Expose an API in grpcio::env that shuts-down AND joins the threads

`grpcio::env` impl of `Drop` issues commands to request all the
completion queues to shutdown, but does not actually join the threads.

For a lot of webservers this works fine, but for some tests, it
creates a problem.

In my usecase I have a server containing SGX enclaves and a database,
and I want to validate that even if the server goes down and comes back
repeatedly, the users are able to recover their data from the database.

```
    let users = ... mock user set

    for phase_count in 0..NUM_PHASES {
        log::info!(logger, "Phase {}/{}", phase_count + 1, NUM_PHASES);

        // First make grpcio env
        let grpcio_env = mobile_acct_api::make_env();

        ... make server, make client,
        ... make requests for each mock user,
        ... validate results
    }
```

Unfortunately for me, even though `grpcio_env` is scoped to the loop
body, the threads actually leak out because the implementation of
`Drop` does not join the threads.

Unfortunately, this consistently causes crashes in tests because intel
sgx sdk contains a `SimEnclaveMgr` object which has a static lifetime
and is torn down at process destruction.

I believe that with the current API, I cannot guarantee that
my grpcio threads are torn down BEFORE that object is. The only way
that I can do that is if there is some API on `grpcio::Environment`
that actually joins the threads.

In the actual rust tests that validate `grpcio::Environment`, you
yourselves have written code that joins the join handles. I would
like to be able to do that in my tests at the end of my loop body.

This commit exposes an API on grpcio::Environment that both issues
the shutdown command, AND joins the join handles. It also makes
the rust unit test, in that same file, use this API.

This is not a breaking change, since we don't change the implementation
of `Drop` or any other public api.

Signed-off-by: Chris Beck <[email protected]>

Author: cbeck88

src/env.rs

@@ -129,6 +129,17 @@ impl Environment {
         let idx = self.idx.fetch_add(1, Ordering::Relaxed);
         self.cqs[idx % self.cqs.len()].clone()
     }
+
+    /// Shutdown the completion queues and join all threads
+    pub fn shutdown_and_join(&mut self) {
+        for cq in self.completion_queues() {
+            cq.shutdown();
+        }
+
+        for handle in self._handles.drain(..) {
+            handle.join().unwrap();
+        }
+    }
 }
 
 impl Drop for Environment {
@@ -163,12 +174,6 @@ mod tests {
         }
 
         assert_eq!(env.completion_queues().len(), 2);
-        for cq in env.completion_queues() {
-            cq.shutdown();
-        }
-
-        for handle in env._handles.drain(..) {
-            handle.join().unwrap();
-        }
+        env.shutdown_and_join();
     }
 }