tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-18 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

rt: internally split Handle into two structs (#4629)

Browse Source 
Previously, `runtime::Handle` was a single struct composed of the
internal handles for each runtime component. This patch splits the
`Handle` struct into a `HandleInner` which contains everything
**except** the task scheduler handle. Now, `HandleInner` is passed to
the task scheduler during creation and the task scheduler is responsible
for storing it. `Handle` only  needs to hold the scheduler handle and
can access the rest of the component handles by querying the task
scheduler.

The motivation for this change is it now enables the multi-threaded
scheduler to have direct access to the blocking spawner handle.
Previously, when spawning a new thread, the multi-threaded scheduler had
to access the blocking spawner by accessing a thread-local variable.
Now, in theory, the multi-threaded scheduler can use `HandleInner`
directly. However, this change hasn't been done in this PR yet.

Also, now the `Handle` struct is much smaller.

This change is intended to make it easier for the multi-threaded
scheduler to shutdown idle threads and respawn them on demand.
This commit is contained in:
Carl Lerche
2022-04-20 12:56:55 -07:00
committed by GitHub
parent d590a369d5
commit 911a0efa87
16 changed files with 235 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/runtime/basic_scheduler.rs
+10 -1
View File
@@ -5,7 +5,7 @@ use crate::park::{Park, Unpark};
use crate::runtime::context::EnterGuard;
use crate::runtime::driver::Driver;
use crate::runtime::task::{self, JoinHandle, OwnedTasks, Schedule, Task};
use crate::runtime::Callback;
use crate::runtime::{Callback, HandleInner};
use crate::runtime::{MetricsBatch, SchedulerMetrics, WorkerMetrics};
use crate::sync::notify::Notify;
use crate::util::atomic_cell::AtomicCell;
@@ -78,6 +78,9 @@ struct Shared {
/// Indicates whether the blocked on thread was woken.
woken: AtomicBool,
/// Handle to I/O driver, timer, blocking pool, ...
handle_inner: HandleInner,
/// Callback for a worker parking itself
before_park: Option<Callback>,
@@ -119,6 +122,7 @@ scoped_thread_local!(static CURRENT: Context);
impl BasicScheduler {
pub(crate) fn new(
driver: Driver,
handle_inner: HandleInner,
before_park: Option<Callback>,
after_unpark: Option<Callback>,
) -> BasicScheduler {
@@ -130,6 +134,7 @@ impl BasicScheduler {
owned: OwnedTasks::new(),
unpark,
woken: AtomicBool::new(false),
handle_inner,
before_park,
after_unpark,
scheduler_metrics: SchedulerMetrics::new(),
@@ -397,6 +402,10 @@ impl Spawner {
pub(crate) fn reset_woken(&self) -> bool {
self.shared.woken.swap(false, AcqRel)
}
pub(crate) fn as_handle_inner(&self) -> &HandleInner {
&self.shared.handle_inner
}
}
cfg_metrics! {
tokio/src/runtime/blocking/mod.rs
-25
View File
@@ -21,28 +21,3 @@ use crate::runtime::Builder;
pub(crate) fn create_blocking_pool(builder: &Builder, thread_cap: usize) -> BlockingPool {
BlockingPool::new(builder, thread_cap)
}
/*
cfg_not_blocking_impl! {
use crate::runtime::Builder;
use std::time::Duration;
#[derive(Debug, Clone)]
pub(crate) struct BlockingPool {}
pub(crate) use BlockingPool as Spawner;
pub(crate) fn create_blocking_pool(_builder: &Builder, _thread_cap: usize) -> BlockingPool {
BlockingPool {}
}
impl BlockingPool {
pub(crate) fn spawner(&self) -> &BlockingPool {
self
}
pub(crate) fn shutdown(&mut self, _duration: Option<Duration>) {
}
}
}
*/
tokio/src/runtime/blocking/pool.rs
+6 -6
View File
@@ -7,7 +7,7 @@ use crate::runtime::blocking::shutdown;
use crate::runtime::builder::ThreadNameFn;
use crate::runtime::context;
use crate::runtime::task::{self, JoinHandle};
use crate::runtime::{Builder, Callback, Handle};
use crate::runtime::{Builder, Callback, ToHandle};
use std::collections::{HashMap, VecDeque};
use std::fmt;
@@ -129,7 +129,7 @@ cfg_fs! {
R: Send + 'static,
{
let rt = context::current();
rt.spawn_mandatory_blocking(func)
rt.as_inner().spawn_mandatory_blocking(&rt, func)
}
}
@@ -220,7 +220,7 @@ impl fmt::Debug for BlockingPool {
// ===== impl Spawner =====
impl Spawner {
pub(crate) fn spawn(&self, task: Task, rt: &Handle) -> Result<(), ()> {
pub(crate) fn spawn(&self, task: Task, rt: &dyn ToHandle) -> Result<(), ()> {
let mut shared = self.inner.shared.lock();
if shared.shutdown {
@@ -283,7 +283,7 @@ impl Spawner {
fn spawn_thread(
&self,
shutdown_tx: shutdown::Sender,
rt: &Handle,
rt: &dyn ToHandle,
id: usize,
) -> std::io::Result<thread::JoinHandle<()>> {
let mut builder = thread::Builder::new().name((self.inner.thread_name)());
@@ -292,12 +292,12 @@ impl Spawner {
builder = builder.stack_size(stack_size);
}
let rt = rt.clone();
let rt = rt.to_handle();
builder.spawn(move || {
// Only the reference should be moved into the closure
let _enter = crate::runtime::context::enter(rt.clone());
rt.blocking_spawner.inner.run(id);
rt.as_inner().blocking_spawner.inner.run(id);
drop(shutdown_tx);
})
}
tokio/src/runtime/builder.rs
+32 -25
View File
@@ -555,32 +555,37 @@ impl Builder {
}
fn build_basic_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::{BasicScheduler, Kind};
use crate::runtime::{BasicScheduler, HandleInner, Kind};
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
// And now put a single-threaded scheduler on top of the timer. When
// there are no futures ready to do something, it'll let the timer or
// the reactor to generate some new stimuli for the futures to continue
// in their life.
let scheduler =
BasicScheduler::new(driver, self.before_park.clone(), self.after_unpark.clone());
let spawner = Spawner::Basic(scheduler.spawner().clone());
// Blocking pool
let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads);
let blocking_spawner = blocking_pool.spawner().clone();
let handle_inner = HandleInner {
io_handle: resources.io_handle,
time_handle: resources.time_handle,
signal_handle: resources.signal_handle,
clock: resources.clock,
blocking_spawner,
};
// And now put a single-threaded scheduler on top of the timer. When
// there are no futures ready to do something, it'll let the timer or
// the reactor to generate some new stimuli for the futures to continue
// in their life.
let scheduler = BasicScheduler::new(
driver,
handle_inner,
self.before_park.clone(),
self.after_unpark.clone(),
);
let spawner = Spawner::Basic(scheduler.spawner().clone());
Ok(Runtime {
kind: Kind::CurrentThread(scheduler),
handle: Handle {
spawner,
io_handle: resources.io_handle,
time_handle: resources.time_handle,
signal_handle: resources.signal_handle,
clock: resources.clock,
blocking_spawner,
},
handle: Handle { spawner },
blocking_pool,
})
}
@@ -662,23 +667,17 @@ cfg_rt_multi_thread! {
impl Builder {
fn build_threaded_runtime(&mut self) -> io::Result<Runtime> {
use crate::loom::sys::num_cpus;
use crate::runtime::{Kind, ThreadPool};
use crate::runtime::park::Parker;
use crate::runtime::{Kind, HandleInner, ThreadPool};
let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
let (scheduler, launch) = ThreadPool::new(core_threads, Parker::new(driver), self.before_park.clone(), self.after_unpark.clone());
let spawner = Spawner::ThreadPool(scheduler.spawner().clone());
// Create the blocking pool
let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads + core_threads);
let blocking_spawner = blocking_pool.spawner().clone();
// Create the runtime handle
let handle = Handle {
spawner,
let handle_inner = HandleInner {
io_handle: resources.io_handle,
time_handle: resources.time_handle,
signal_handle: resources.signal_handle,
@@ -686,6 +685,14 @@ cfg_rt_multi_thread! {
blocking_spawner,
};
let (scheduler, launch) = ThreadPool::new(core_threads, driver, handle_inner, self.before_park.clone(), self.after_unpark.clone());
let spawner = Spawner::ThreadPool(scheduler.spawner().clone());
// Create the runtime handle
let handle = Handle {
spawner,
};
// Spawn the thread pool workers
let _enter = crate::runtime::context::enter(handle.clone());
launch.launch();
tokio/src/runtime/context.rs
+4 -4
View File
@@ -26,7 +26,7 @@ cfg_io_driver! {
pub(crate) fn io_handle() -> crate::runtime::driver::IoHandle {
match CONTEXT.try_with(|ctx| {
let ctx = ctx.borrow();
ctx.as_ref().expect(crate::util::error::CONTEXT_MISSING_ERROR).io_handle.clone()
ctx.as_ref().expect(crate::util::error::CONTEXT_MISSING_ERROR).as_inner().io_handle.clone()
}) {
Ok(io_handle) => io_handle,
Err(_) => panic!("{}", crate::util::error::THREAD_LOCAL_DESTROYED_ERROR),
@@ -39,7 +39,7 @@ cfg_signal_internal! {
pub(crate) fn signal_handle() -> crate::runtime::driver::SignalHandle {
match CONTEXT.try_with(|ctx| {
let ctx = ctx.borrow();
ctx.as_ref().expect(crate::util::error::CONTEXT_MISSING_ERROR).signal_handle.clone()
ctx.as_ref().expect(crate::util::error::CONTEXT_MISSING_ERROR).as_inner().signal_handle.clone()
}) {
Ok(signal_handle) => signal_handle,
Err(_) => panic!("{}", crate::util::error::THREAD_LOCAL_DESTROYED_ERROR),
@@ -51,7 +51,7 @@ cfg_time! {
pub(crate) fn time_handle() -> crate::runtime::driver::TimeHandle {
match CONTEXT.try_with(|ctx| {
let ctx = ctx.borrow();
ctx.as_ref().expect(crate::util::error::CONTEXT_MISSING_ERROR).time_handle.clone()
ctx.as_ref().expect(crate::util::error::CONTEXT_MISSING_ERROR).as_inner().time_handle.clone()
}) {
Ok(time_handle) => time_handle,
Err(_) => panic!("{}", crate::util::error::THREAD_LOCAL_DESTROYED_ERROR),
@@ -60,7 +60,7 @@ cfg_time! {
cfg_test_util! {
pub(crate) fn clock() -> Option<crate::runtime::driver::Clock> {
match CONTEXT.try_with(|ctx| (*ctx.borrow()).as_ref().map(|ctx| ctx.clock.clone())) {
match CONTEXT.try_with(|ctx| (*ctx.borrow()).as_ref().map(|ctx| ctx.as_inner().clock.clone())) {
Ok(clock) => clock,
Err(_) => panic!("{}", crate::util::error::THREAD_LOCAL_DESTROYED_ERROR),
}
tokio/src/runtime/handle.rs
+111 -77
View File
@@ -16,7 +16,11 @@ use std::{error, fmt};
#[derive(Debug, Clone)]
pub struct Handle {
pub(super) spawner: Spawner,
}
/// All internal handles that are *not* the scheduler's spawner.
#[derive(Debug)]
pub(crate) struct HandleInner {
/// Handles to the I/O drivers
#[cfg_attr(
not(any(feature = "net", feature = "process", all(unix, feature = "signal"))),
@@ -47,6 +51,11 @@ pub struct Handle {
pub(super) blocking_spawner: blocking::Spawner,
}
/// Create a new runtime handle.
pub(crate) trait ToHandle {
fn to_handle(&self) -> Handle;
}
/// Runtime context guard.
///
/// Returned by [`Runtime::enter`] and [`Handle::enter`], the context guard exits
@@ -196,85 +205,11 @@ impl Handle {
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let (join_handle, _was_spawned) =
if cfg!(debug_assertions) && std::mem::size_of::<F>() > 2048 {
self.spawn_blocking_inner(Box::new(func), blocking::Mandatory::NonMandatory, None)
} else {
self.spawn_blocking_inner(func, blocking::Mandatory::NonMandatory, None)
};
join_handle
self.as_inner().spawn_blocking(self, func)
}
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<F, R>(&self, func: F) -> Option<JoinHandle<R>>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let (join_handle, was_spawned) = if cfg!(debug_assertions) && std::mem::size_of::<F>() > 2048 {
self.spawn_blocking_inner(
Box::new(func),
blocking::Mandatory::Mandatory,
None
)
} else {
self.spawn_blocking_inner(
func,
blocking::Mandatory::Mandatory,
None
)
};
if was_spawned {
Some(join_handle)
} else {
None
}
}
}
#[track_caller]
pub(crate) fn spawn_blocking_inner<F, R>(
&self,
func: F,
is_mandatory: blocking::Mandatory,
name: Option<&str>,
) -> (JoinHandle<R>, bool)
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let fut = BlockingTask::new(func);
#[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(),
"fn" = %std::any::type_name::<F>(),
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);
let spawned = self
.blocking_spawner
.spawn(blocking::Task::new(task, is_mandatory), self);
(handle, spawned.is_ok())
pub(crate) fn as_inner(&self) -> &HandleInner {
self.spawner.as_handle_inner()
}
/// Runs a future to completion on this `Handle`'s associated `Runtime`.
@@ -369,6 +304,12 @@ impl Handle {
}
}
impl ToHandle for Handle {
fn to_handle(&self) -> Handle {
self.clone()
}
}
cfg_metrics! {
use crate::runtime::RuntimeMetrics;
@@ -381,6 +322,99 @@ cfg_metrics! {
}
}
impl HandleInner {
#[track_caller]
pub(crate) fn spawn_blocking<F, R>(&self, rt: &dyn ToHandle, func: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let (join_handle, _was_spawned) = if cfg!(debug_assertions)
&& std::mem::size_of::<F>() > 2048
{
self.spawn_blocking_inner(Box::new(func), blocking::Mandatory::NonMandatory, None, rt)
} else {
self.spawn_blocking_inner(func, blocking::Mandatory::NonMandatory, None, rt)
};
join_handle
}
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<F, R>(&self, rt: &dyn ToHandle, func: F) -> Option<JoinHandle<R>>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let (join_handle, was_spawned) = if cfg!(debug_assertions) && std::mem::size_of::<F>() > 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 was_spawned {
Some(join_handle)
} else {
None
}
}
}
#[track_caller]
pub(crate) fn spawn_blocking_inner<F, R>(
&self,
func: F,
is_mandatory: blocking::Mandatory,
name: Option<&str>,
rt: &dyn ToHandle,
) -> (JoinHandle<R>, bool)
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let fut = BlockingTask::new(func);
#[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(),
"fn" = %std::any::type_name::<F>(),
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);
let spawned = self
.blocking_spawner
.spawn(blocking::Task::new(task, is_mandatory), rt);
(handle, spawned.is_ok())
}
}
/// Error returned by `try_current` when no Runtime has been started
#[derive(Debug)]
pub struct TryCurrentError {
tokio/src/runtime/metrics/runtime.rs
+1
View File
@@ -526,6 +526,7 @@ cfg_net! {
// TODO: Investigate if this should return 0, most of our metrics always increase
// thus this breaks that guarantee.
self.handle
.as_inner()
.io_handle
.as_ref()
.and_then(|h| h.with_io_driver_metrics(f))
tokio/src/runtime/mod.rs
+3 -7
View File
@@ -218,24 +218,20 @@ cfg_rt! {
pub use self::builder::Builder;
pub(crate) mod context;
pub(crate) mod driver;
mod driver;
use self::enter::enter;
mod handle;
pub use handle::{EnterGuard, Handle, TryCurrentError};
pub(crate) use handle::{HandleInner, ToHandle};
mod spawner;
use self::spawner::Spawner;
}
cfg_rt_multi_thread! {
mod park;
use park::Parker;
}
cfg_rt_multi_thread! {
mod queue;
use driver::Driver;
pub(crate) mod thread_pool;
use self::thread_pool::ThreadPool;
tokio/src/runtime/spawner.rs
+9 -1
View File
@@ -1,5 +1,5 @@
use crate::future::Future;
use crate::runtime::basic_scheduler;
use crate::runtime::{basic_scheduler, HandleInner};
use crate::task::JoinHandle;
cfg_rt_multi_thread! {
@@ -34,6 +34,14 @@ impl Spawner {
Spawner::ThreadPool(spawner) => spawner.spawn(future),
}
}
pub(crate) fn as_handle_inner(&self) -> &HandleInner {
match self {
Spawner::Basic(spawner) => spawner.as_handle_inner(),
#[cfg(feature = "rt-multi-thread")]
Spawner::ThreadPool(spawner) => spawner.as_handle_inner(),
}
}
}
cfg_metrics! {
tokio/src/runtime/tests/loom_queue.rs
+2 -1
View File
@@ -1,6 +1,7 @@
use crate::runtime::blocking::NoopSchedule;
use crate::runtime::task::Inject;
use crate::runtime::{queue, MetricsBatch};
use crate::runtime::thread_pool::queue;
use crate::runtime::MetricsBatch;
use loom::thread;
tokio/src/runtime/tests/queue.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use crate::runtime::queue;
use crate::runtime::task::{self, Inject, Schedule, Task};
use crate::runtime::thread_pool::queue;
use crate::runtime::MetricsBatch;
use std::thread;
tokio/src/runtime/thread_pool/mod.rs
+15 -3
View File
@@ -3,6 +3,11 @@
mod idle;
use self::idle::Idle;
mod park;
pub(crate) use park::{Parker, Unparker};
pub(super) mod queue;
mod worker;
pub(crate) use worker::Launch;
@@ -10,7 +15,7 @@ pub(crate) use worker::block_in_place;
use crate::loom::sync::Arc;
use crate::runtime::task::JoinHandle;
use crate::runtime::{Callback, Parker};
use crate::runtime::{Callback, Driver, HandleInner};
use std::fmt;
use std::future::Future;
@@ -42,11 +47,14 @@ pub(crate) struct Spawner {
impl ThreadPool {
pub(crate) fn new(
size: usize,
parker: Parker,
driver: Driver,
handle_inner: HandleInner,
before_park: Option<Callback>,
after_unpark: Option<Callback>,
) -> (ThreadPool, Launch) {
let (shared, launch) = worker::create(size, parker, before_park, after_unpark);
let parker = Parker::new(driver);
let (shared, launch) =
worker::create(size, parker, handle_inner, before_park, after_unpark);
let spawner = Spawner { shared };
let thread_pool = ThreadPool { spawner };
@@ -101,6 +109,10 @@ impl Spawner {
pub(crate) fn shutdown(&mut self) {
self.shared.close();
}
pub(crate) fn as_handle_inner(&self) -> &HandleInner {
self.shared.as_handle_inner()
}
}
cfg_metrics! {
tokio/src/runtime/park.rs → tokio/src/runtime/thread_pool/park.rs
View File
tokio/src/runtime/queue.rs → tokio/src/runtime/thread_pool/queue.rs
+10 -10
View File
@@ -11,14 +11,14 @@ use std::ptr;
use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release};
/// Producer handle. May only be used from a single thread.
pub(super) struct Local<T: 'static> {
pub(crate) struct Local<T: 'static> {
inner: Arc<Inner<T>>,
}
/// Consumer handle. May be used from many threads.
pub(super) struct Steal<T: 'static>(Arc<Inner<T>>);
pub(crate) struct Steal<T: 'static>(Arc<Inner<T>>);
pub(super) struct Inner<T: 'static> {
pub(crate) struct Inner<T: 'static> {
/// Concurrently updated by many threads.
///
/// Contains two `u16` values. The LSB byte is the "real" head of the queue.
@@ -65,7 +65,7 @@ fn make_fixed_size<T>(buffer: Box<[T]>) -> Box<[T; LOCAL_QUEUE_CAPACITY]> {
}
/// Create a new local run-queue
pub(super) fn local<T: 'static>() -> (Steal<T>, Local<T>) {
pub(crate) fn local<T: 'static>() -> (Steal<T>, Local<T>) {
let mut buffer = Vec::with_capacity(LOCAL_QUEUE_CAPACITY);
for _ in 0..LOCAL_QUEUE_CAPACITY {
@@ -89,7 +89,7 @@ pub(super) fn local<T: 'static>() -> (Steal<T>, Local<T>) {
impl<T> Local<T> {
/// Returns true if the queue has entries that can be stealed.
pub(super) fn is_stealable(&self) -> bool {
pub(crate) fn is_stealable(&self) -> bool {
!self.inner.is_empty()
}
@@ -97,12 +97,12 @@ impl<T> Local<T> {
///
/// Separate to is_stealable so that refactors of is_stealable to "protect"
/// some tasks from stealing won't affect this
pub(super) fn has_tasks(&self) -> bool {
pub(crate) fn has_tasks(&self) -> bool {
!self.inner.is_empty()
}
/// Pushes a task to the back of the local queue, skipping the LIFO slot.
pub(super) fn push_back(
pub(crate) fn push_back(
&mut self,
mut task: task::Notified<T>,
inject: &Inject<T>,
@@ -259,7 +259,7 @@ impl<T> Local<T> {
}
/// Pops a task from the local queue.
pub(super) fn pop(&mut self) -> Option<task::Notified<T>> {
pub(crate) fn pop(&mut self) -> Option<task::Notified<T>> {
let mut head = self.inner.head.load(Acquire);
let idx = loop {
@@ -301,12 +301,12 @@ impl<T> Local<T> {
}
impl<T> Steal<T> {
pub(super) fn is_empty(&self) -> bool {
pub(crate) fn is_empty(&self) -> bool {
self.0.is_empty()
}
/// Steals half the tasks from self and place them into `dst`.
pub(super) fn steal_into(
pub(crate) fn steal_into(
&self,
dst: &mut Local<T>,
dst_metrics: &mut MetricsBatch,
tokio/src/runtime/thread_pool/worker.rs
+24 -3
View File
@@ -63,10 +63,9 @@ use crate::loom::sync::{Arc, Mutex};
use crate::park::{Park, Unpark};
use crate::runtime;
use crate::runtime::enter::EnterContext;
use crate::runtime::park::{Parker, Unparker};
use crate::runtime::task::{Inject, JoinHandle, OwnedTasks};
use crate::runtime::thread_pool::Idle;
use crate::runtime::{queue, task, Callback, MetricsBatch, SchedulerMetrics, WorkerMetrics};
use crate::runtime::thread_pool::{queue, Idle, Parker, Unparker};
use crate::runtime::{task, Callback, HandleInner, MetricsBatch, SchedulerMetrics, WorkerMetrics};
use crate::util::atomic_cell::AtomicCell;
use crate::util::FastRand;
@@ -122,6 +121,9 @@ struct Core {
/// State shared across all workers
pub(super) struct Shared {
/// Handle to the I/O driver, timer, blocking spawner, ...
handle_inner: HandleInner,
/// Per-worker remote state. All other workers have access to this and is
/// how they communicate between each other.
remotes: Box<[Remote]>,
@@ -193,6 +195,7 @@ scoped_thread_local!(static CURRENT: Context);
pub(super) fn create(
size: usize,
park: Parker,
handle_inner: HandleInner,
before_park: Option<Callback>,
after_unpark: Option<Callback>,
) -> (Arc<Shared>, Launch) {
@@ -223,6 +226,7 @@ pub(super) fn create(
}
let shared = Arc::new(Shared {
handle_inner,
remotes: remotes.into_boxed_slice(),
inject: Inject::new(),
idle: Idle::new(size),
@@ -715,6 +719,10 @@ impl task::Schedule for Arc<Shared> {
}
impl Shared {
pub(crate) fn as_handle_inner(&self) -> &HandleInner {
&self.handle_inner
}
pub(super) fn bind_new_task<T>(me: &Arc<Self>, future: T) -> JoinHandle<T::Output>
where
T: Future + Send + 'static,
@@ -853,6 +861,19 @@ impl Shared {
}
}
impl crate::runtime::ToHandle for Arc<Shared> {
fn to_handle(&self) -> crate::runtime::Handle {
use crate::runtime::thread_pool::Spawner;
use crate::runtime::{self, Handle};
Handle {
spawner: runtime::Spawner::ThreadPool(Spawner {
shared: self.clone(),
}),
}
}
}
cfg_metrics! {
impl Shared {
pub(super) fn injection_queue_depth(&self) -> usize {
tokio/src/task/builder.rs
+7 -2
View File
@@ -108,8 +108,13 @@ impl<'a> Builder<'a> {
Output: Send + 'static,
{
use crate::runtime::Mandatory;
let (join_handle, _was_spawned) =
context::current().spawn_blocking_inner(function, Mandatory::NonMandatory, self.name);
let handle = context::current();
let (join_handle, _was_spawned) = handle.as_inner().spawn_blocking_inner(
function,
Mandatory::NonMandatory,
self.name,
&handle,
);
join_handle
}
}