From a997fc7199f4db26a7b128c8b0fce1c7af3fa8c2 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Thu, 24 Sep 2026 00:00:00 +0000 Subject: [PATCH 1/2] chore(deps): update tokio from 1.13 to 1.49 see: https://github.com/tokio-rs/tokio/blob/master/tokio/CHANGELOG.md#1490-january-3rd-2026 this does not break hyper-util's MSRV, this tokio release supports 1.71 and newer. this release includes the stabilization of `tokio::runtime::Handle::id()`, which will permit us to write unit tests that inspect which runtime a spawned future is being run on. Signed-off-by: katelyn martin --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 363ad770..9e5769b0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ percent-encoding = { version = "2.3", optional = true } pin-project-lite = "0.2.4" socket2 = { version = "0.6", optional = true, features = ["all"] } tracing = { version = "0.1.36", default-features = false, features = ["std"], optional = true } -tokio = { version = "1.13", optional = true, default-features = false } +tokio = { version = "1.49", optional = true, default-features = false } tower-layer = { version = "0.3", optional = true } tower-service = { version = "0.3", optional = true } From eee3d2362900967d8e4c7862aad436ac26b027d0 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Thu, 24 Sep 2026 00:00:00 +0000 Subject: [PATCH 2/2] feat(rt/tokio): introduce `TokioHandleExecutor` this commit introduces an `hyper::rt::Executor` implementation, `TokioHandleExecutor`. 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 `TokioHandleExecutor` 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 --- src/rt/tokio.rs | 2 +- src/rt/tokio/executor.rs | 129 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 129 insertions(+), 2 deletions(-) diff --git a/src/rt/tokio.rs b/src/rt/tokio.rs index 4df10273..ea93b8ec 100644 --- a/src/rt/tokio.rs +++ b/src/rt/tokio.rs @@ -61,7 +61,7 @@ use hyper::rt::{Sleep, Timer}; use pin_project_lite::pin_project; pub use self::{ - executor::{TokioExecutor, TokioLocalExecutor}, + executor::{TokioExecutor, TokioHandleExecutor, TokioLocalExecutor}, with_hyper_io::WithHyperIo, with_tokio_io::WithTokioIo, }; diff --git a/src/rt/tokio/executor.rs b/src/rt/tokio/executor.rs index d18d990c..dd5e84b2 100644 --- a/src/rt/tokio/executor.rs +++ b/src/rt/tokio/executor.rs @@ -63,6 +63,45 @@ pub struct TokioExecutor {} #[derive(Default, Debug, Clone)] pub struct TokioLocalExecutor {} +/// 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::TokioHandleExecutor; +/// +/// let runtime = tokio::runtime::Builder::new_current_thread() +/// .build() +/// .unwrap(); +/// let handle = runtime.handle().clone(); +/// let executor = TokioHandleExecutor::new(handle); +/// ``` +/// +/// [`Handle`]: tokio::runtime::Handle +/// [`Handle::spawn()`]: tokio::runtime::Handle::spawn +/// [`Runtime::handle()`]: tokio::runtime::Runtime::handle +pub struct TokioHandleExecutor { + handle: tokio::runtime::Handle, +} + // ===== impl TokioExecutor ===== impl Executor for TokioExecutor @@ -105,9 +144,31 @@ where } } +// ===== impl TokioHandleExecutor ===== + +impl TokioHandleExecutor { + /// TK + pub fn new(handle: tokio::runtime::Handle) -> Self { + Self { handle } + } +} + +impl Executor for TokioHandleExecutor +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, tokio::executor::TokioLocalExecutor}; + use crate::rt::{ + TokioExecutor, + tokio::{TokioHandleExecutor, TokioLocalExecutor}, + }; use hyper::rt::Executor; use tokio::sync::oneshot; @@ -220,4 +281,70 @@ mod tests { let runtime = tokio::runtime::LocalRuntime::new().unwrap(); runtime.block_on(fut); } + + #[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 = TokioHandleExecutor::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"); + } }