From 427342f6423f84f4c4afdc8283173ee8c957b9e4 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Thu, 24 Sep 2026 00:00:00 +0000 Subject: [PATCH 1/3] refactor(rt/tokio): move `TokioExecutor` into submodule this commit moves `rt::tokio::TokioExecutor` down a level into a new private `rt::tokio::executor` submodule. a `pub use` statement is placed into `rt::tokio` so that this retains an identical public interface. Signed-off-by: katelyn martin --- src/rt/tokio.rs | 91 ++-------------------------------------- src/rt/tokio/executor.rs | 87 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 88 deletions(-) create mode 100644 src/rt/tokio/executor.rs diff --git a/src/rt/tokio.rs b/src/rt/tokio.rs index dee88e05..96ceb36b 100644 --- a/src/rt/tokio.rs +++ b/src/rt/tokio.rs @@ -57,35 +57,15 @@ use std::{ time::{Duration, Instant}, }; -use hyper::rt::{Executor, Sleep, Timer}; +use hyper::rt::{Sleep, Timer}; use pin_project_lite::pin_project; -#[cfg(feature = "rt-tracing-exec-force")] -use tracing::instrument::Instrument; - -pub use self::{with_hyper_io::WithHyperIo, with_tokio_io::WithTokioIo}; +pub use self::{executor::TokioExecutor, with_hyper_io::WithHyperIo, with_tokio_io::WithTokioIo}; +mod executor; mod with_hyper_io; mod with_tokio_io; -/// Future executor that utilises `tokio` threads. -/// -/// Spawned futures do not inherit the current tracing span, even when the -/// `tracing` feature is enabled. To propagate spans, wrap this executor in -/// one of the components from [`rt::tracing`](crate::rt::tracing), such as -/// [`CurrentSpanExecutor`](crate::rt::CurrentSpanExecutor) (available with -/// the `tracing` feature). -/// -/// See the module-level documentation of [`rt::tracing`](crate::rt::tracing) -/// for more information about propagating [`tracing`] spans to spawned tasks. -/// -/// The temporary `rt-tracing-exec-force` feature restores propagation of the -/// current span for libraries that do not allow customizing their executor. -/// It is excluded from `full` and may be removed in a future breaking release. -#[non_exhaustive] -#[derive(Default, Debug, Clone)] -pub struct TokioExecutor {} - pin_project! { /// A wrapper that implements Tokio's IO traits for an inner type that /// implements hyper's IO traits, or vice versa (implements hyper's IO @@ -112,29 +92,6 @@ pin_project! { } } -// ===== impl TokioExecutor ===== - -impl Executor for TokioExecutor -where - Fut: Future + Send + 'static, - Fut::Output: Send + 'static, -{ - fn execute(&self, fut: Fut) { - #[cfg(feature = "rt-tracing-exec-force")] - tokio::spawn(fut.in_current_span()); - - #[cfg(not(feature = "rt-tracing-exec-force"))] - tokio::spawn(fut); - } -} - -impl TokioExecutor { - /// Create new executor that relies on [`tokio::spawn`] to execute futures. - pub fn new() -> Self { - Self {} - } -} - // ==== impl TokioIo ===== impl TokioIo { @@ -335,45 +292,3 @@ impl TokioSleep { self.project().inner.as_mut().reset(deadline.into()); } } - -#[cfg(test)] -mod tests { - use crate::rt::TokioExecutor; - use hyper::rt::Executor; - use tokio::sync::oneshot; - - #[tokio::test] - async fn simple_execute() -> Result<(), Box> { - let (tx, rx) = oneshot::channel(); - let executor = TokioExecutor::new(); - executor.execute(async move { - tx.send(()).unwrap(); - }); - rx.await.map_err(Into::into) - } - - #[cfg(feature = "tracing")] - #[tokio::test] - async fn execute_tracing_span() { - // The current-thread runtime keeps the subscriber active while the - // spawned future is polled, after the caller has exited its span. - let _subscriber = tracing::subscriber::set_default(tracing_subscriber::registry()); - let span = tracing::info_span!("caller"); - assert!(span.id().is_some()); - let (tx, rx) = oneshot::channel(); - - { - let _entered = span.enter(); - TokioExecutor::new().execute(async move { - tx.send(tracing::Span::current().id()).unwrap(); - }); - } - - let spawned_span = rx.await.unwrap(); - if cfg!(feature = "rt-tracing-exec-force") { - assert_eq!(spawned_span, span.id()); - } else { - assert_eq!(spawned_span, None); - } - } -} diff --git a/src/rt/tokio/executor.rs b/src/rt/tokio/executor.rs new file mode 100644 index 00000000..0525e479 --- /dev/null +++ b/src/rt/tokio/executor.rs @@ -0,0 +1,87 @@ +use hyper::rt::Executor; + +#[cfg(feature = "rt-tracing-exec-force")] +use tracing::instrument::Instrument; + +/// Future executor that utilises `tokio` threads. +/// +/// Spawned futures do not inherit the current tracing span, even when the +/// `tracing` feature is enabled. To propagate spans, wrap this executor in +/// one of the components from [`rt::tracing`](crate::rt::tracing), such as +/// [`CurrentSpanExecutor`](crate::rt::CurrentSpanExecutor) (available with +/// the `tracing` feature). +/// +/// See the module-level documentation of [`rt::tracing`](crate::rt::tracing) +/// for more information about propagating [`tracing`] spans to spawned tasks. +/// +/// The temporary `rt-tracing-exec-force` feature restores propagation of the +/// current span for libraries that do not allow customizing their executor. +/// It is excluded from `full` and may be removed in a future breaking release. +#[non_exhaustive] +#[derive(Default, Debug, Clone)] +pub struct TokioExecutor {} + +// ===== impl TokioExecutor ===== + +impl Executor for TokioExecutor +where + Fut: Future + Send + 'static, + Fut::Output: Send + 'static, +{ + fn execute(&self, fut: Fut) { + #[cfg(feature = "rt-tracing-exec-force")] + tokio::spawn(fut.in_current_span()); + + #[cfg(not(feature = "rt-tracing-exec-force"))] + tokio::spawn(fut); + } +} + +impl TokioExecutor { + /// Create new executor that relies on [`tokio::spawn`] to execute futures. + pub fn new() -> Self { + Self {} + } +} + +#[cfg(test)] +mod tests { + use crate::rt::TokioExecutor; + use hyper::rt::Executor; + use tokio::sync::oneshot; + + #[tokio::test] + async fn simple_execute() -> Result<(), Box> { + let (tx, rx) = oneshot::channel(); + let executor = TokioExecutor::new(); + executor.execute(async move { + tx.send(()).unwrap(); + }); + rx.await.map_err(Into::into) + } + + #[cfg(feature = "tracing")] + #[tokio::test] + async fn execute_tracing_span() { + // The current-thread runtime keeps the subscriber active while the + // spawned future is polled, after the caller has exited its span. + let _subscriber = tracing::subscriber::set_default(tracing_subscriber::registry()); + let span = tracing::info_span!("caller"); + assert!(span.id().is_some()); + let (tx, rx) = oneshot::channel(); + + { + let _entered = span.enter(); + TokioExecutor::new().execute(async move { + tx.send(tracing::Span::current().id()).unwrap(); + }); + } + + let spawned_span = rx.await.unwrap(); + if cfg!(feature = "rt-tracing-exec-force") { + assert_eq!(spawned_span, span.id()); + } else { + assert_eq!(spawned_span, None); + } + } +} From dc1b313b135db83776db5e003ae9e26c88d5ff31 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Thu, 24 Sep 2026 00:00:00 +0000 Subject: [PATCH 2/3] feat(rt/tokio): introduce `LocalExecutor` the `rt::tokio::TokioExecutor` type allows for hyper clients and servers to be run within a tokio runtime, and generally satisfies for most common use cases. however, `tokio::spawn()` has a trait bound that is not satisfiable in some cases: it mandates that spawned futures be `Send`. some asynchronous applications are not designed to permit work-stealing between threads, and expect that a spawned future be polled to completion on the same thread that spawned it. in order to facilitate such uses, this commit proposes the introduction of another `Executor` implementation. this commit introduces `LocalExecutor`, which makes use of `tokio::task::spawn_local()` rather than `tokio::task::spawn()` to spawn background tasks. two test cases showing that this can run a future that interacts with single-threaded reference counting storage (`Rc`) is included to confirm that this works as expected, with both `LocalSet` and `LocalRuntime`. Signed-off-by: katelyn martin --- src/rt/tokio.rs | 6 +- src/rt/tokio/executor.rs | 138 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 142 insertions(+), 2 deletions(-) diff --git a/src/rt/tokio.rs b/src/rt/tokio.rs index 96ceb36b..7ea7ddcc 100644 --- a/src/rt/tokio.rs +++ b/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::{LocalExecutor, TokioExecutor}, + with_hyper_io::WithHyperIo, + with_tokio_io::WithTokioIo, +}; mod executor; mod with_hyper_io; diff --git a/src/rt/tokio/executor.rs b/src/rt/tokio/executor.rs index 0525e479..8b438dfe 100644 --- a/src/rt/tokio/executor.rs +++ b/src/rt/tokio/executor.rs @@ -21,6 +21,48 @@ use tracing::instrument::Instrument; #[derive(Default, Debug, Clone)] pub struct TokioExecutor {} +/// Future executor that utilises local `tokio` threads. +/// +/// This executor relies on [`tokio::task::spawn_local()`] to execute futures. +/// This is of use when dealing with a server or client implementation that is +/// `!Send`, i.e. it must be run on the same thread. +/// +/// *Note:* This cannot be used within a task spawned with [`tokio::spawn()`]. +/// +/// # Examples +/// +/// Execute tasks within a [`LocalSet`][tokio::task::LocalSet]. +/// +/// ``` +/// use hyper_util::rt::tokio::LocalExecutor; +/// +/// let runtime = tokio::runtime::Builder::new_current_thread() +/// .build() +/// .unwrap(); +/// let local_set = tokio::task::LocalSet::new(); +/// local_set.block_on(&runtime, async move { +/// let executor = LocalExecutor::new(); +/// +/// // Use the executor... +/// }); +/// ``` +/// +/// Execute tasks within a [`LocalRuntime`][tokio::runtime::LocalRuntime]. +/// +/// ``` +/// use hyper_util::rt::tokio::LocalExecutor; +/// +/// let runtime = tokio::runtime::LocalRuntime::new().unwrap(); +/// runtime.block_on(async move { +/// let executor = LocalExecutor::new(); +/// +/// // Use the executor... +/// }); +/// ``` +#[non_exhaustive] +#[derive(Default, Debug, Clone)] +pub struct LocalExecutor {} + // ===== impl TokioExecutor ===== impl Executor for TokioExecutor @@ -44,9 +86,28 @@ impl TokioExecutor { } } +// ===== impl LocalExecutor ===== + +impl LocalExecutor { + /// Create a new executor that relies on [`tokio::task::spawn_local()`] to execute futures. + pub fn new() -> Self { + Self {} + } +} + +impl Executor for LocalExecutor +where + Fut: Future + 'static, + Fut::Output: 'static, +{ + fn execute(&self, fut: Fut) { + tokio::task::spawn_local(fut); + } +} + #[cfg(test)] mod tests { - use crate::rt::TokioExecutor; + use crate::rt::{TokioExecutor, tokio::executor::LocalExecutor}; use hyper::rt::Executor; use tokio::sync::oneshot; @@ -84,4 +145,79 @@ mod tests { assert_eq!(spawned_span, None); } } + + #[test] + fn local_executor_works_with_local_set() { + // A future that will spawn a background task that increments a + // (single-threaded) reference-counted integer, indicating via a + // oneshot channel when that has been done. + // + // This is a `!Send` future, because it uses `Rc`. + let fut = async { + use std::{rc::Rc, sync::Mutex}; + let here = Rc::new(Mutex::new(0)); + let (tx, rx) = oneshot::channel(); + + // Spawn the background task on the local set. + let executor = LocalExecutor::new(); + let there = here.clone(); + let fut = async move { + // *there += 42; + *there.lock().unwrap() = 42; + tx.send(()).unwrap(); + }; + + executor.execute(fut); + + rx.await.unwrap(); + assert_eq!(*here.lock().unwrap(), 42); + }; + + // NOTE: We can't assert negative bounds but this can be uncommented + // to check that `fut` above is a `!Send` future. + // fn assert_sync(_: &T) {} + // assert_sync(&fut); + + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap(); + let local_set = tokio::task::LocalSet::new(); + local_set.block_on(&runtime, fut); + } + + #[test] + fn local_executor_works_with_local_runtime() { + // A future that will spawn a background task that increments a + // (single-threaded) reference-counted integer, indicating via a + // oneshot channel when that has been done. + // + // This is a `!Send` future, because it uses `Rc`. + let fut = async { + use std::{rc::Rc, sync::Mutex}; + let here = Rc::new(Mutex::new(0)); + let (tx, rx) = oneshot::channel(); + + // Spawn the background task on the local set. + let executor = LocalExecutor::new(); + let there = here.clone(); + let fut = async move { + // *there += 42; + *there.lock().unwrap() = 42; + tx.send(()).unwrap(); + }; + + executor.execute(fut); + + rx.await.unwrap(); + assert_eq!(*here.lock().unwrap(), 42); + }; + + // NOTE: We can't assert negative bounds but this can be uncommented + // to check that `fut` above is a `!Send` future. + // fn assert_sync(_: &T) {} + // assert_sync(&fut); + + let runtime = tokio::runtime::LocalRuntime::new().unwrap(); + runtime.block_on(fut); + } } From 82393df203002d9a4de2462f78f9615516e1ef13 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Thu, 24 Sep 2026 00:00:00 +0000 Subject: [PATCH 3/3] review(rt/tokio): rename `LocalExecutor` to `TokioLocalExecutor` to avoid conflicts with potential future additions integrating with other asynchronous runtimes, this commit adds a `Tokio` prefix to the name of our local task executor. Signed-off-by: katelyn martin --- src/rt/tokio.rs | 2 +- src/rt/tokio/executor.rs | 22 +++++++++++----------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/rt/tokio.rs b/src/rt/tokio.rs index 7ea7ddcc..4df10273 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::{LocalExecutor, TokioExecutor}, + executor::{TokioExecutor, TokioLocalExecutor}, with_hyper_io::WithHyperIo, with_tokio_io::WithTokioIo, }; diff --git a/src/rt/tokio/executor.rs b/src/rt/tokio/executor.rs index 8b438dfe..d18d990c 100644 --- a/src/rt/tokio/executor.rs +++ b/src/rt/tokio/executor.rs @@ -34,14 +34,14 @@ pub struct TokioExecutor {} /// Execute tasks within a [`LocalSet`][tokio::task::LocalSet]. /// /// ``` -/// use hyper_util::rt::tokio::LocalExecutor; +/// use hyper_util::rt::tokio::TokioLocalExecutor; /// /// let runtime = tokio::runtime::Builder::new_current_thread() /// .build() /// .unwrap(); /// let local_set = tokio::task::LocalSet::new(); /// local_set.block_on(&runtime, async move { -/// let executor = LocalExecutor::new(); +/// let executor = TokioLocalExecutor::new(); /// /// // Use the executor... /// }); @@ -50,18 +50,18 @@ pub struct TokioExecutor {} /// Execute tasks within a [`LocalRuntime`][tokio::runtime::LocalRuntime]. /// /// ``` -/// use hyper_util::rt::tokio::LocalExecutor; +/// use hyper_util::rt::tokio::TokioLocalExecutor; /// /// let runtime = tokio::runtime::LocalRuntime::new().unwrap(); /// runtime.block_on(async move { -/// let executor = LocalExecutor::new(); +/// let executor = TokioLocalExecutor::new(); /// /// // Use the executor... /// }); /// ``` #[non_exhaustive] #[derive(Default, Debug, Clone)] -pub struct LocalExecutor {} +pub struct TokioLocalExecutor {} // ===== impl TokioExecutor ===== @@ -86,16 +86,16 @@ impl TokioExecutor { } } -// ===== impl LocalExecutor ===== +// ===== impl TokioLocalExecutor ===== -impl LocalExecutor { +impl TokioLocalExecutor { /// Create a new executor that relies on [`tokio::task::spawn_local()`] to execute futures. pub fn new() -> Self { Self {} } } -impl Executor for LocalExecutor +impl Executor for TokioLocalExecutor where Fut: Future + 'static, Fut::Output: 'static, @@ -107,7 +107,7 @@ where #[cfg(test)] mod tests { - use crate::rt::{TokioExecutor, tokio::executor::LocalExecutor}; + use crate::rt::{TokioExecutor, tokio::executor::TokioLocalExecutor}; use hyper::rt::Executor; use tokio::sync::oneshot; @@ -159,7 +159,7 @@ mod tests { let (tx, rx) = oneshot::channel(); // Spawn the background task on the local set. - let executor = LocalExecutor::new(); + let executor = TokioLocalExecutor::new(); let there = here.clone(); let fut = async move { // *there += 42; @@ -198,7 +198,7 @@ mod tests { let (tx, rx) = oneshot::channel(); // Spawn the background task on the local set. - let executor = LocalExecutor::new(); + let executor = TokioLocalExecutor::new(); let there = here.clone(); let fut = async move { // *there += 42;