Skip to content

Commit

Permalink
rt: split internal runtime::Handle concerns (#5022)
Browse files Browse the repository at this point in the history
The `runtime::Handle` struct is part of the public API but is also used
internally. This has created a bit of tension. An earlier patch made
defined Handle as a private struct in some cases when `rt` is not
enabled.

This patch splits out internal handle concerns into a new
`scheduler::Handle` type, which will only be internal. This also defines
a `Handle` type for each scheduler variant. Eventually, the
per-scheduler `Handle` types will replace the per-scheduler `Spawner`
types, but more work is needed before we can make that change.
  • Loading branch information
carllerche committed Sep 16, 2022
1 parent b5709ba commit ebeb78e
Show file tree
Hide file tree
Showing 16 changed files with 649 additions and 602 deletions.
4 changes: 2 additions & 2 deletions tokio/src/runtime/blocking/pool.rs
Expand Up @@ -154,7 +154,7 @@ cfg_fs! {
R: Send + 'static,
{
let rt = context::current();
rt.as_inner().blocking_spawner.spawn_mandatory_blocking(&rt, func)
rt.inner.blocking_spawner().spawn_mandatory_blocking(&rt, func)
}
}

Expand Down Expand Up @@ -419,7 +419,7 @@ impl Spawner {
builder.spawn(move || {
// Only the reference should be moved into the closure
let _enter = crate::runtime::context::enter(rt.clone());
rt.as_inner().blocking_spawner.inner.run(id);
rt.inner.blocking_spawner().inner.run(id);
drop(shutdown_tx);
})
}
Expand Down
21 changes: 11 additions & 10 deletions tokio/src/runtime/builder.rs
@@ -1,5 +1,5 @@
use crate::runtime::handle::Handle;
use crate::runtime::{blocking, driver, Callback, Runtime, Spawner};
use crate::runtime::{blocking, driver, Callback, Runtime};
use crate::util::{RngSeed, RngSeedGenerator};

use std::fmt;
Expand Down Expand Up @@ -874,7 +874,8 @@ impl Builder {
}

fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::{Config, CurrentThread, HandleInner, Scheduler};
use crate::runtime::scheduler::{self, current_thread, CurrentThread};
use crate::runtime::{Config, Scheduler};
use std::sync::Arc;

let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?;
Expand All @@ -900,14 +901,13 @@ impl Builder {
seed_generator: self.seed_generator.next_generator(),
},
);
let spawner = Spawner::CurrentThread(scheduler.spawner().clone());

let inner = Arc::new(HandleInner {
spawner,
let inner = Arc::new(current_thread::Handle {
spawner: scheduler.spawner().clone(),
driver: driver_handle,
blocking_spawner,
seed_generator: self.seed_generator.next_generator(),
});
let inner = scheduler::Handle::CurrentThread(inner);

Ok(Runtime {
scheduler: Scheduler::CurrentThread(scheduler),
Expand Down Expand Up @@ -993,7 +993,8 @@ cfg_rt_multi_thread! {
impl Builder {
fn build_threaded_runtime(&mut self) -> io::Result<Runtime> {
use crate::loom::sys::num_cpus;
use crate::runtime::{Config, HandleInner, Scheduler, MultiThread};
use crate::runtime::{Config, Scheduler};
use crate::runtime::scheduler::{self, multi_thread, MultiThread};
use std::sync::Arc;

let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
Expand All @@ -1019,14 +1020,14 @@ cfg_rt_multi_thread! {
seed_generator: self.seed_generator.next_generator(),
},
);
let spawner = Spawner::MultiThread(scheduler.spawner().clone());

let inner = Arc::new(HandleInner {
spawner,
let inner = Arc::new(multi_thread::Handle {
spawner: scheduler.spawner().clone(),
driver: driver_handle,
blocking_spawner,
seed_generator: self.seed_generator.next_generator(),
});
let inner = scheduler::Handle::MultiThread(inner);

// Create the runtime handle
let handle = Handle { inner };
Expand Down
27 changes: 8 additions & 19 deletions tokio/src/runtime/context.rs
Expand Up @@ -31,8 +31,8 @@ cfg_io_driver! {
let ctx = ctx.borrow();
ctx.as_ref()
.expect(crate::util::error::CONTEXT_MISSING_ERROR)
.as_inner()
.driver
.inner
.driver()
.io
.clone()
}) {
Expand All @@ -49,9 +49,8 @@ cfg_signal_internal! {
let ctx = ctx.borrow();
ctx.as_ref()
.expect(crate::util::error::CONTEXT_MISSING_ERROR)
.as_inner()
.driver
.signal
.inner
.signal()
.clone()
}) {
Ok(signal_handle) => signal_handle,
Expand All @@ -65,10 +64,9 @@ cfg_time! {
pub(crate) fn clock() -> Option<crate::runtime::driver::Clock> {
match CONTEXT.try_with(|ctx| {
let ctx = ctx.borrow();
ctx.as_ref()
.map(|ctx| {
ctx.as_inner().driver.clock.clone()
})
ctx
.as_ref()
.map(|ctx| ctx.inner.clock().clone())
}) {
Ok(clock) => clock,
Err(_) => panic!("{}", crate::util::error::THREAD_LOCAL_DESTROYED_ERROR),
Expand All @@ -77,15 +75,6 @@ cfg_time! {
}
}

cfg_rt! {
pub(crate) fn spawn_handle() -> Option<crate::runtime::Spawner> {
match CONTEXT.try_with(|ctx| (*ctx.borrow()).as_ref().map(|ctx| ctx.inner.spawner.clone())) {
Ok(spawner) => spawner,
Err(_) => panic!("{}", crate::util::error::THREAD_LOCAL_DESTROYED_ERROR),
}
}
}

/// Sets this [`Handle`] as the current active [`Handle`].
///
/// [`Handle`]: Handle
Expand All @@ -100,7 +89,7 @@ pub(crate) fn enter(new: Handle) -> EnterGuard {
///
/// [`Handle`]: Handle
pub(crate) fn try_enter(new: Handle) -> Option<EnterGuard> {
let rng_seed = new.as_inner().seed_generator.next_seed();
let rng_seed = new.inner.seed_generator().next_seed();
let old_handle = CONTEXT.try_with(|ctx| ctx.borrow_mut().replace(new)).ok()?;

let old_seed = replace_thread_rng(rng_seed);
Expand Down

0 comments on commit ebeb78e

Please sign in to comment.