rt: add Runtime::shutdown_timeout (#2186)

Provides an API for forcing a runtime to shutdown even if there are
still running tasks.

tokio/src/runtime/blocking/pool.rs

@@ -101,19 +101,30 @@ impl BlockingPool {
     pub(crate) fn spawner(&self) -> &Spawner {
         &self.spawner
     }
-}
 
-impl Drop for BlockingPool {
-    fn drop(&mut self) {
+    pub(crate) fn shutdown(&mut self, timeout: Option<Duration>) {
         let mut shared = self.spawner.inner.shared.lock().unwrap();
 
+        // The function can be called multiple times. First, by explicitly
+        // calling `shutdown` then by the drop handler calling `shutdown`. This
+        // prevents shutting down twice.
+        if shared.shutdown {
+            return;
+        }
+
         shared.shutdown = true;
         shared.shutdown_tx = None;
         self.spawner.inner.condvar.notify_all();
 
         drop(shared);
 
-        self.shutdown_rx.wait();
+        self.shutdown_rx.wait(timeout);
+    }
+}
+
+impl Drop for BlockingPool {
+    fn drop(&mut self) {
+        self.shutdown(None);
     }
 }