diff --git a/tokio/src/runtime/blocking/mod.rs b/tokio/src/runtime/blocking/mod.rs index 7633299b3..88bdcfd64 100644 --- a/tokio/src/runtime/blocking/mod.rs +++ b/tokio/src/runtime/blocking/mod.rs @@ -4,15 +4,20 @@ //! compilation. mod pool; -pub(crate) use pool::{spawn_blocking, BlockingPool, Mandatory, SpawnError, Spawner, Task}; +pub(crate) use pool::{spawn_blocking, BlockingPool, Spawner}; cfg_fs! { pub(crate) use pool::spawn_mandatory_blocking; } +cfg_trace! { + pub(crate) use pool::Mandatory; +} + mod schedule; mod shutdown; mod task; +#[cfg(all(test, not(tokio_wasm)))] pub(crate) use schedule::NoopSchedule; pub(crate) use task::BlockingTask; diff --git a/tokio/src/runtime/blocking/pool.rs b/tokio/src/runtime/blocking/pool.rs index 193153885..d5953c20b 100644 --- a/tokio/src/runtime/blocking/pool.rs +++ b/tokio/src/runtime/blocking/pool.rs @@ -3,7 +3,7 @@ use crate::loom::sync::{Arc, Condvar, Mutex}; use crate::loom::thread; use crate::runtime::blocking::schedule::NoopSchedule; -use crate::runtime::blocking::shutdown; +use crate::runtime::blocking::{shutdown, BlockingTask}; use crate::runtime::builder::ThreadNameFn; use crate::runtime::context; use crate::runtime::task::{self, JoinHandle}; @@ -150,7 +150,7 @@ cfg_fs! { R: Send + 'static, { let rt = context::current(); - rt.as_inner().spawn_mandatory_blocking(&rt, func) + rt.as_inner().blocking_spawner.spawn_mandatory_blocking(&rt, func) } } @@ -241,7 +241,103 @@ impl fmt::Debug for BlockingPool { // ===== impl Spawner ===== impl Spawner { - pub(crate) fn spawn(&self, task: Task, rt: &dyn ToHandle) -> Result<(), SpawnError> { + #[track_caller] + pub(crate) fn spawn_blocking(&self, rt: &dyn ToHandle, func: F) -> JoinHandle + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + let (join_handle, spawn_result) = + if cfg!(debug_assertions) && std::mem::size_of::() > 2048 { + self.spawn_blocking_inner(Box::new(func), Mandatory::NonMandatory, None, rt) + } else { + self.spawn_blocking_inner(func, Mandatory::NonMandatory, None, rt) + }; + + match spawn_result { + Ok(()) => join_handle, + // Compat: do not panic here, return the join_handle even though it will never resolve + Err(SpawnError::ShuttingDown) => join_handle, + Err(SpawnError::NoThreads(e)) => { + panic!("OS can't spawn worker thread: {}", e) + } + } + } + + cfg_fs! { + #[track_caller] + #[cfg_attr(any( + all(loom, not(test)), // the function is covered by loom tests + test + ), allow(dead_code))] + pub(crate) fn spawn_mandatory_blocking(&self, rt: &dyn ToHandle, func: F) -> Option> + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + let (join_handle, spawn_result) = if cfg!(debug_assertions) && std::mem::size_of::() > 2048 { + self.spawn_blocking_inner( + Box::new(func), + Mandatory::Mandatory, + None, + rt, + ) + } else { + self.spawn_blocking_inner( + func, + Mandatory::Mandatory, + None, + rt, + ) + }; + + if spawn_result.is_ok() { + Some(join_handle) + } else { + None + } + } + } + + #[track_caller] + pub(crate) fn spawn_blocking_inner( + &self, + func: F, + is_mandatory: Mandatory, + name: Option<&str>, + rt: &dyn ToHandle, + ) -> (JoinHandle, Result<(), SpawnError>) + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + let fut = BlockingTask::new(func); + let id = task::Id::next(); + #[cfg(all(tokio_unstable, feature = "tracing"))] + let fut = { + use tracing::Instrument; + let location = std::panic::Location::caller(); + let span = tracing::trace_span!( + target: "tokio::task::blocking", + "runtime.spawn", + kind = %"blocking", + task.name = %name.unwrap_or_default(), + task.id = id.as_u64(), + "fn" = %std::any::type_name::(), + spawn.location = %format_args!("{}:{}:{}", location.file(), location.line(), location.column()), + ); + fut.instrument(span) + }; + + #[cfg(not(all(tokio_unstable, feature = "tracing")))] + let _ = name; + + let (task, handle) = task::unowned(fut, NoopSchedule, id); + let spawned = self.spawn_task(Task::new(task, is_mandatory), rt); + (handle, spawned) + } + + fn spawn_task(&self, task: Task, rt: &dyn ToHandle) -> Result<(), SpawnError> { let mut shared = self.inner.shared.lock(); if shared.shutdown { diff --git a/tokio/src/runtime/handle.rs b/tokio/src/runtime/handle.rs index 075792a30..4675d7af9 100644 --- a/tokio/src/runtime/handle.rs +++ b/tokio/src/runtime/handle.rs @@ -1,5 +1,4 @@ -use crate::runtime::blocking::{BlockingTask, NoopSchedule}; -use crate::runtime::task::{self, JoinHandle}; +use crate::runtime::task::JoinHandle; use crate::runtime::{blocking, context, driver, Spawner}; use crate::util::error::{CONTEXT_MISSING_ERROR, THREAD_LOCAL_DESTROYED_ERROR}; @@ -52,7 +51,7 @@ pub(crate) struct HandleInner { pub(super) clock: driver::Clock, /// Blocking pool spawner - pub(super) blocking_spawner: blocking::Spawner, + pub(crate) blocking_spawner: blocking::Spawner, } /// Create a new runtime handle. @@ -208,7 +207,7 @@ impl Handle { F: FnOnce() -> R + Send + 'static, R: Send + 'static, { - self.as_inner().spawn_blocking(self, func) + self.as_inner().blocking_spawner.spawn_blocking(self, func) } pub(crate) fn as_inner(&self) -> &HandleInner { @@ -338,107 +337,6 @@ cfg_metrics! { } } -impl HandleInner { - #[track_caller] - pub(crate) fn spawn_blocking(&self, rt: &dyn ToHandle, func: F) -> JoinHandle - where - F: FnOnce() -> R + Send + 'static, - R: Send + 'static, - { - let (join_handle, spawn_result) = if cfg!(debug_assertions) - && std::mem::size_of::() > 2048 - { - self.spawn_blocking_inner(Box::new(func), blocking::Mandatory::NonMandatory, None, rt) - } else { - self.spawn_blocking_inner(func, blocking::Mandatory::NonMandatory, None, rt) - }; - - match spawn_result { - Ok(()) => join_handle, - // Compat: do not panic here, return the join_handle even though it will never resolve - Err(blocking::SpawnError::ShuttingDown) => join_handle, - Err(blocking::SpawnError::NoThreads(e)) => { - panic!("OS can't spawn worker thread: {}", e) - } - } - } - - cfg_fs! { - #[track_caller] - #[cfg_attr(any( - all(loom, not(test)), // the function is covered by loom tests - test - ), allow(dead_code))] - pub(crate) fn spawn_mandatory_blocking(&self, rt: &dyn ToHandle, func: F) -> Option> - where - F: FnOnce() -> R + Send + 'static, - R: Send + 'static, - { - let (join_handle, spawn_result) = if cfg!(debug_assertions) && std::mem::size_of::() > 2048 { - self.spawn_blocking_inner( - Box::new(func), - blocking::Mandatory::Mandatory, - None, - rt, - ) - } else { - self.spawn_blocking_inner( - func, - blocking::Mandatory::Mandatory, - None, - rt, - ) - }; - - if spawn_result.is_ok() { - Some(join_handle) - } else { - None - } - } - } - - #[track_caller] - pub(crate) fn spawn_blocking_inner( - &self, - func: F, - is_mandatory: blocking::Mandatory, - name: Option<&str>, - rt: &dyn ToHandle, - ) -> (JoinHandle, Result<(), blocking::SpawnError>) - where - F: FnOnce() -> R + Send + 'static, - R: Send + 'static, - { - let fut = BlockingTask::new(func); - let id = super::task::Id::next(); - #[cfg(all(tokio_unstable, feature = "tracing"))] - let fut = { - use tracing::Instrument; - let location = std::panic::Location::caller(); - let span = tracing::trace_span!( - target: "tokio::task::blocking", - "runtime.spawn", - kind = %"blocking", - task.name = %name.unwrap_or_default(), - task.id = id.as_u64(), - "fn" = %std::any::type_name::(), - spawn.location = %format_args!("{}:{}:{}", location.file(), location.line(), location.column()), - ); - fut.instrument(span) - }; - - #[cfg(not(all(tokio_unstable, feature = "tracing")))] - let _ = name; - - let (task, handle) = task::unowned(fut, NoopSchedule, id); - let spawned = self - .blocking_spawner - .spawn(blocking::Task::new(task, is_mandatory), rt); - (handle, spawned) - } -} - /// Error returned by `try_current` when no Runtime has been started #[derive(Debug)] pub struct TryCurrentError { diff --git a/tokio/src/task/builder.rs b/tokio/src/task/builder.rs index c7a101b29..e18636d96 100644 --- a/tokio/src/task/builder.rs +++ b/tokio/src/task/builder.rs @@ -187,7 +187,7 @@ impl<'a> Builder<'a> { Output: Send + 'static, { use crate::runtime::Mandatory; - let (join_handle, spawn_result) = handle.as_inner().spawn_blocking_inner( + let (join_handle, spawn_result) = handle.as_inner().blocking_spawner.spawn_blocking_inner( function, Mandatory::NonMandatory, self.name,