feat(rt/tokio): introduce HandleExecutor

this commit introduces an `hyper::rt::Executor` implementation,
`HandleExecutor`. this executor is used to run tasks on a tokio runtime,
but spawns tasks onto a particular runtime using a provided
`tokio::runtime::Handle` rather than by calling `tokio::spawn()`

this is useful for situations in which you wish to run tasks on a
*separate* runtime, for example. this may be applicable to those
configuring hyper clients and servers in applications running on NUMA
(Non-Uniform Memory Awareneses) systems, or those that wish to manage
provision separate resources for background tasks. see the
[`tokio::runtime`] documentation for more information about choosing the
correct runtime for an application.

a unit test is included which demonstrates that `HandleExecutor` can be
used to spawn background tasks onto a separate tokio runtime.

[`tokio::runtime`]: https://docs.rs/tokio/latest/tokio/runtime/index.html#numa-awareness

Signed-off-by: katelyn martin <git@katelyn.world>

Author: cratelyn

src/rt/tokio.rs

@@ -60,7 +60,11 @@ use std::{
 use hyper::rt::{Sleep, Timer};
 use pin_project_lite::pin_project;
 
-pub use self::{executor::TokioExecutor, with_hyper_io::WithHyperIo, with_tokio_io::WithTokioIo};
+pub use self::{
+    executor::{HandleExecutor, TokioExecutor},
+    with_hyper_io::WithHyperIo,
+    with_tokio_io::WithTokioIo,
+};
 
 mod executor;
 mod with_hyper_io;

src/rt/tokio/executor.rs

@@ -21,6 +21,45 @@ use tracing::instrument::Instrument;
 #[derive(Default, Debug, Clone)]
 pub struct TokioExecutor {}
 
+/// Future executor backed by a runtime [`Handle`].
+///
+/// This executor, like [`TokioExecutor`], utilises [`tokio`] threads. This
+/// executor spawns tasks using [`Handle::spawn()`] rather than
+/// [`tokio::spawn()`], however.
+///
+/// A runtime handle may be obtained by calling [`Runtime::handle()`].
+///
+/// This is useful for situations in which you wish to run tasks on a
+/// *separate* runtime. If your application only runs using a single tokio
+/// runtime, [`TokioExecutor`] should be used instead.
+///
+/// This may be applicable to those configuring hyper clients and servers in
+/// applications running on NUMA (Non-Uniform Memory Awareneses) systems, or
+/// if you wish to manage provision separate resources for background tasks
+/// associated with a client or server.
+///
+/// See the [`tokio::runtime`] documentation for more information about
+/// choosing the correct runtime for your application.
+///
+/// # Examples
+///
+/// ```
+/// use hyper_util::rt::tokio::HandleExecutor;
+///
+/// let runtime = tokio::runtime::Builder::new_current_thread()
+///     .build()
+///     .unwrap();
+/// let handle = runtime.handle().clone();
+/// let executor = HandleExecutor::new(handle);
+/// ```
+///
+/// [`Handle`]: tokio::runtime::Handle
+/// [`Handle::spawn()`]: tokio::runtime::Handle::spawn
+/// [`Runtime::handle()`]: tokio::runtime::Runtime::handle
+pub struct HandleExecutor {
+    handle: tokio::runtime::Handle,
+}
+
 // ===== impl TokioExecutor =====
 
 impl<Fut> Executor<Fut> for TokioExecutor
@@ -44,9 +83,28 @@ impl TokioExecutor {
     }
 }
 
+// ===== impl HandleExecutor =====
+
+impl HandleExecutor {
+    /// TK
+    pub fn new(handle: tokio::runtime::Handle) -> Self {
+        Self { handle }
+    }
+}
+
+impl<Fut> Executor<Fut> for HandleExecutor
+where
+    Fut: Future + Send + 'static,
+    Fut::Output: Send + 'static,
+{
+    fn execute(&self, fut: Fut) {
+        self.handle.spawn(fut);
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use crate::rt::TokioExecutor;
+    use crate::rt::tokio::{HandleExecutor, TokioExecutor};
     use hyper::rt::Executor;
     use tokio::sync::oneshot;
 
@@ -84,4 +142,70 @@ mod tests {
             assert_eq!(spawned_span, None);
         }
     }
+
+    #[test]
+    fn handle_executor_can_execute_task_on_separate_runtime() {
+        // Create a "foreground" runtime we will run our top-level on.
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .worker_threads(1)
+            .name("foreground")
+            .build()
+            .unwrap();
+
+        // Create a "background" runtime, whose handle will be used to spawn
+        // background tasks by our executor.
+        let background = tokio::runtime::Builder::new_current_thread()
+            .worker_threads(1)
+            .name("background")
+            .build()
+            .unwrap();
+        let handle = background.handle().clone();
+        let executor = HandleExecutor::new(handle);
+
+        // Begin running the background runtime on a separate worker thread.
+        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+        let worker = std::thread::Builder::new()
+            .name("execute-task-on-separate-runtime-worker".into())
+            .spawn(move || {
+                use futures_util::FutureExt;
+                let fut = shutdown_rx.map(drop);
+                background.block_on(fut);
+            })
+            .expect("should spawn thread");
+
+        // Run a future that, when polled, spawns a background task onto the
+        // handle executor. This background task retrieves the name of the
+        // runtime that it is running on, and sends the name back to its
+        // caller. The parent then asserts that the child was run on the
+        // "background" runtime.
+        rt.block_on(async move {
+            let handle = tokio::runtime::Handle::current();
+            let name = handle.name().unwrap().to_string();
+            assert_eq!(
+                name, "foreground",
+                "future should be spawned onto foreground runtime"
+            );
+
+            let (tx, rx) = oneshot::channel();
+            let fut = async move {
+                let handle = tokio::runtime::Handle::current();
+                let name = handle.name().unwrap().to_string();
+                tx.send(name).unwrap();
+            };
+
+            executor.execute(fut);
+            let name = rx.await.unwrap();
+            assert_eq!(
+                name, "background",
+                "worker should be spawned onto background runtime"
+            );
+        });
+
+        // Signal to the background runtime that it should shutdown now, and
+        // then wait for the thread running it to finish.
+        shutdown_tx
+            .send(())
+            .expect("shutdown signal should be sent");
+        worker.join().expect("worker thread should finish");
+    }
 }