diff --git a/tokio-test/src/clock.rs b/tokio-test/src/clock.rs index c5fac787f..38417539e 100644 --- a/tokio-test/src/clock.rs +++ b/tokio-test/src/clock.rs @@ -22,7 +22,7 @@ //! }); //! ``` -use tokio::executor::park::{Park, Unpark}; +use tokio::runtime::{Park, Unpark}; use tokio::timer::clock::{Clock, Now}; use tokio::timer::Timer; diff --git a/tokio/src/executor/mod.rs b/tokio/src/executor/mod.rs deleted file mode 100644 index 6134ea3b8..000000000 --- a/tokio/src/executor/mod.rs +++ /dev/null @@ -1,74 +0,0 @@ -//! Task execution related traits and utilities. -//! -//! In the Tokio execution model, futures are lazy. When a future is created, no -//! work is performed. In order for the work defined by the future to happen, -//! the future must be submitted to an executor. A future that is submitted to -//! an executor is called a "task". -//! -//! The executor is responsible for ensuring that [`Future::poll`] is called -//! whenever the task is notified. Notification happens when the internal -//! state of a task transitions from *not ready* to *ready*. For example, a -//! socket might have received data and a call to `read` will now be able to -//! succeed. -//! -//! The specific strategy used to manage the tasks is left up to the -//! executor. There are two main flavors of executors: single-threaded and -//! multi-threaded. Tokio provides implementation for both of these in the -//! [`runtime`] module. -//! -//! # `Executor` trait. -//! -//! This module provides the [`Executor`] trait (re-exported from -//! [`tokio-executor`]), which describes the API that all executors must -//! implement. -//! -//! A free [`spawn`] function is provided that allows spawning futures onto the -//! default executor (tracked via a thread-local variable) without referencing a -//! handle. It is expected that all executors will set a value for the default -//! executor. This value will often be set to the executor itself, but it is -//! possible that the default executor might be set to a different executor. -//! -//! For example, a single threaded executor might set the default executor to a -//! thread pool instead of itself, allowing futures to spawn new tasks onto the -//! thread pool when those tasks are `Send`. -//! -//! [`Future::poll`]: https://docs.rs/futures/0.1/futures/future/trait.Future.html#tymethod.poll -//! [notified]: https://docs.rs/futures/0.1/futures/executor/trait.Notify.html#tymethod.notify -//! [`runtime`]: ../runtime/index.html -//! [`tokio-executor`]: https://docs.rs/tokio-executor/0.1 -//! [`Executor`]: trait.Executor.html -//! [`spawn`]: fn.spawn.html#[cfg(all(test, loom))] - -// At the top due to macros -#[cfg(test)] -#[macro_use] -mod tests; - -#[cfg(feature = "rt-current-thread")] -mod enter; -#[cfg(feature = "rt-current-thread")] -pub(crate) use self::enter::enter; - -mod global; -pub use self::global::spawn; - -pub mod park; - -#[cfg(feature = "rt-current-thread")] -mod task; -#[cfg(feature = "rt-current-thread")] -pub use self::task::{JoinError, JoinHandle}; - -#[cfg(feature = "rt-full")] -mod util; - -#[cfg(all(not(feature = "blocking"), feature = "rt-full"))] -mod blocking; -#[cfg(feature = "blocking")] -pub mod blocking; - -#[cfg(feature = "rt-current-thread")] -pub(crate) mod current_thread; - -#[cfg(feature = "rt-full")] -pub(crate) mod thread_pool; diff --git a/tokio/src/fs/mod.rs b/tokio/src/fs/mod.rs index c1b4ed71a..ed3b4162d 100644 --- a/tokio/src/fs/mod.rs +++ b/tokio/src/fs/mod.rs @@ -91,5 +91,5 @@ where mod sys { pub(crate) use std::fs::File; - pub(crate) use crate::executor::blocking::{run, Blocking}; + pub(crate) use crate::runtime::blocking::{run, Blocking}; } diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index 89547c1d8..b7b99ac6d 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -72,6 +72,7 @@ macro_rules! if_runtime { ($($i:item)*) => ($( #[cfg(any( + feature = "blocking", feature = "rt-full", feature = "rt-current-thread", ))] @@ -98,7 +99,6 @@ pub mod io; #[cfg(feature = "net-driver")] pub mod net; -#[cfg(any(feature = "sync", feature = "blocking", feature = "rt-current-thread"))] mod loom; pub mod prelude; @@ -106,6 +106,8 @@ pub mod prelude; #[cfg(all(feature = "process", not(loom)))] pub mod process; +pub mod runtime; + #[cfg(feature = "signal")] #[cfg(not(loom))] pub mod signal; @@ -118,14 +120,13 @@ pub mod sync; #[cfg(feature = "timer")] pub mod timer; -#[cfg(feature = "executor-core")] -pub mod executor; +#[cfg(feature = "rt-full")] +mod util; if_runtime! { - pub mod runtime; #[doc(inline)] - pub use crate::executor::spawn; + pub use crate::runtime::spawn; #[cfg(not(test))] // Work around for rust-lang/rust#62127 #[cfg(feature = "macros")] diff --git a/tokio/src/loom/std/mod.rs b/tokio/src/loom/std/mod.rs index 5bbf15319..e0aafa83a 100644 --- a/tokio/src/loom/std/mod.rs +++ b/tokio/src/loom/std/mod.rs @@ -12,6 +12,7 @@ pub(crate) mod cell { pub(crate) use super::causal_cell::{CausalCell, CausalCheck}; } +#[cfg(feature = "sync")] pub(crate) mod future { pub(crate) use crate::sync::AtomicWaker; } diff --git a/tokio/src/net/addr.rs b/tokio/src/net/addr.rs index 812964dae..8fe02b4ea 100644 --- a/tokio/src/net/addr.rs +++ b/tokio/src/net/addr.rs @@ -1,4 +1,4 @@ -use crate::executor::blocking; +use crate::runtime::blocking; use futures_util::future; use std::io; @@ -143,7 +143,7 @@ pub(crate) mod sealed { //! part of the `ToSocketAddrs` public API. The details will change over //! time. - use crate::executor::blocking::Blocking; + use crate::runtime::blocking::Blocking; use futures_core::ready; use std::future::Future; diff --git a/tokio/src/net/driver/reactor/mod.rs b/tokio/src/net/driver/reactor/mod.rs index 3a2f35150..8bcded411 100644 --- a/tokio/src/net/driver/reactor/mod.rs +++ b/tokio/src/net/driver/reactor/mod.rs @@ -1,6 +1,6 @@ -use crate::executor::park::{Park, Unpark}; use crate::loom::sync::atomic::AtomicUsize; use crate::net::driver::platform; +use crate::runtime::{Park, Unpark}; use std::sync::atomic::Ordering::SeqCst; diff --git a/tokio/src/executor/blocking/builder.rs b/tokio/src/runtime/blocking/builder.rs similarity index 97% rename from tokio/src/executor/blocking/builder.rs rename to tokio/src/runtime/blocking/builder.rs index e755ae238..59178f25f 100644 --- a/tokio/src/executor/blocking/builder.rs +++ b/tokio/src/runtime/blocking/builder.rs @@ -1,5 +1,5 @@ -use crate::executor::blocking::Pool; use crate::loom::thread; +use crate::runtime::blocking::Pool; use std::usize; diff --git a/tokio/src/executor/blocking/mod.rs b/tokio/src/runtime/blocking/mod.rs similarity index 98% rename from tokio/src/executor/blocking/mod.rs rename to tokio/src/runtime/blocking/mod.rs index d105fe22a..e634ea599 100644 --- a/tokio/src/executor/blocking/mod.rs +++ b/tokio/src/runtime/blocking/mod.rs @@ -262,7 +262,7 @@ impl Drop for PoolWaiter { /// /// ``` /// # async fn docs() { -/// tokio::executor::blocking::in_place(move || { +/// tokio::runtime::blocking::in_place(move || { /// // do some compute-heavy work or call synchronous code /// }); /// # } @@ -272,9 +272,9 @@ pub fn in_place(f: F) -> R where F: FnOnce() -> R, { - use crate::executor; + use crate::runtime::{enter, thread_pool}; - executor::enter::exit(|| executor::thread_pool::blocking(f)) + enter::exit(|| thread_pool::blocking(f)) } /// Run the provided closure on a thread where blocking is acceptable. @@ -288,7 +288,7 @@ where /// /// ``` /// # async fn docs() { -/// tokio::executor::blocking::run(move || { +/// tokio::runtime::blocking::run(move || { /// // do some compute-heavy work or call synchronous code /// }).await; /// # } diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index 1ed38c8a0..92fdfa1aa 100644 --- a/tokio/src/runtime/builder.rs +++ b/tokio/src/runtime/builder.rs @@ -1,14 +1,13 @@ -use crate::executor::blocking::{Pool, PoolWaiter}; -use crate::executor::current_thread::CurrentThread; +#[cfg(feature = "blocking")] +use crate::runtime::blocking::{Pool, PoolWaiter}; +#[cfg(feature = "rt-current-thread")] +use crate::runtime::current_thread::CurrentThread; #[cfg(feature = "rt-full")] -use crate::executor::thread_pool; -use crate::net::driver::Reactor; -use crate::runtime::{Runtime, Kind}; -use crate::timer::clock::Clock; -use crate::timer::timer::Timer; +use crate::runtime::thread_pool; +use crate::runtime::{io, timer, Runtime}; +use std::fmt; use std::sync::Arc; -use std::{fmt, io}; /// Builds Tokio Runtime with custom configuration values. /// @@ -43,8 +42,8 @@ use std::{fmt, io}; /// } /// ``` pub struct Builder { - /// When `true`, use the current-thread executor. - current_thread: bool, + /// The task execution model to use. + kind: Kind, /// The number of worker threads. /// @@ -64,7 +63,16 @@ pub struct Builder { before_stop: Option>, /// The clock to use - clock: Clock, + clock: timer::Clock, +} + +#[derive(Debug)] +enum Kind { + Shell, + #[cfg(feature = "rt-current-thread")] + CurrentThread, + #[cfg(feature = "rt-full")] + ThreadPool, } impl Builder { @@ -74,8 +82,8 @@ impl Builder { /// Configuration methods can be chained on the return value. pub fn new() -> Builder { Builder { - // Use the thread-pool executor by default - current_thread: false, + // No task execution by default + kind: Kind::Shell, // Default to use an equal number of threads to number of CPU cores num_threads: crate::loom::sys::num_cpus(), @@ -91,7 +99,7 @@ impl Builder { before_stop: None, // Default clock - clock: Clock::new(), + clock: timer::Clock::default(), } } @@ -119,12 +127,20 @@ impl Builder { self } - /// Use only the current thread for the runtime. + /// Use only the current thread for executing tasks. /// /// The network driver, timer, and executor will all be run on the current /// thread during `block_on` calls. + #[cfg(feature = "rt-current-thread")] pub fn current_thread(&mut self) -> &mut Self { - self.current_thread = true; + self.kind = Kind::CurrentThread; + self + } + + /// Use a thread-pool for executing tasks. + #[cfg(feature = "rt-full")] + pub fn thread_pool(&mut self) -> &mut Self { + self.kind = Kind::ThreadPool; self } @@ -224,7 +240,7 @@ impl Builder { } /// Set the `Clock` instance that will be used by the runtime. - pub fn clock(&mut self, clock: Clock) -> &mut Self { + pub fn clock(&mut self, clock: timer::Clock) -> &mut Self { self.clock = clock; self } @@ -245,20 +261,44 @@ impl Builder { /// }); /// ``` pub fn build(&mut self) -> io::Result { - if self.current_thread { - self.build_current_thread() - } else { - self.build_threadpool() + match self.kind { + Kind::Shell => self.build_shell(), + #[cfg(feature = "rt-current-thread")] + Kind::CurrentThread => self.build_current_thread(), + #[cfg(feature = "rt-full")] + Kind::ThreadPool => self.build_threadpool(), } } - fn build_current_thread(&mut self) -> io::Result { - // Create network driver - let net = Reactor::new()?; - let net_handles = vec![net.handle()]; + fn build_shell(&mut self) -> io::Result { + use crate::runtime::Kind; - let timer = Timer::new_with_clock(net, self.clock.clone()); - let timer_handles = vec![timer.handle()]; + // Create network driver + let (net, handle) = io::create()?; + let net_handles = vec![handle]; + + let (_timer, handle) = timer::create(net, self.clock.clone()); + let timer_handles = vec![handle]; + + Ok(Runtime { + kind: Kind::Shell, + net_handles, + timer_handles, + #[cfg(feature = "blocking")] + blocking_pool: PoolWaiter::from(Pool::default()), + }) + } + + #[cfg(feature = "rt-current-thread")] + fn build_current_thread(&mut self) -> io::Result { + use crate::runtime::Kind; + + // Create network driver + let (net, handle) = io::create()?; + let net_handles = vec![handle]; + + let (timer, handle) = timer::create(net, self.clock.clone()); + let timer_handles = vec![handle]; // And now put a single-threaded executor on top of the timer. When // there are no futures ready to do something, it'll let the timer or @@ -277,16 +317,10 @@ impl Builder { }) } - // Without rt-full, the "threadpool" variant just uses current-thread - #[cfg(not(feature = "rt-full"))] - fn build_threadpool(&mut self) -> io::Result { - self.build_current_thread() - } - #[cfg(feature = "rt-full")] fn build_threadpool(&mut self) -> io::Result { - use crate::net::driver; - use crate::timer::{clock, timer}; + use crate::runtime::Kind; + use crate::timer::clock; use std::sync::Mutex; let mut net_handles = Vec::new(); @@ -294,13 +328,13 @@ impl Builder { let mut timers = Vec::new(); for _ in 0..self.num_threads { - // Create network driver - let net = Reactor::new()?; - net_handles.push(net.handle()); + // Create network driver and handle + let (net, handle) = io::create()?; + net_handles.push(handle); // Create a new timer. - let timer = Timer::new_with_clock(net, self.clock.clone()); - timer_handles.push(timer.handle()); + let (timer, handle) = timer::create(net, self.clock.clone()); + timer_handles.push(handle); timers.push(Mutex::new(Some(timer))); } @@ -328,7 +362,7 @@ impl Builder { builder .around_worker(move |index, next| { // Configure the network driver - let _net = driver::set_default(&net_handles[index]); + let _net = io::set_default(&net_handles[index]); // Configure the clock clock::with_default(&clock, || { @@ -370,7 +404,7 @@ impl Default for Builder { impl fmt::Debug for Builder { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("Builder") - .field("current_thread", &self.current_thread) + .field("kind", &self.kind) .field("num_threads", &self.num_threads) .field("thread_name", &self.thread_name) .field("thread_stack_size", &self.thread_stack_size) diff --git a/tokio/src/executor/current_thread/mod.rs b/tokio/src/runtime/current_thread/mod.rs similarity index 96% rename from tokio/src/executor/current_thread/mod.rs rename to tokio/src/runtime/current_thread/mod.rs index 24fee3a87..d63b4ccfc 100644 --- a/tokio/src/executor/current_thread/mod.rs +++ b/tokio/src/runtime/current_thread/mod.rs @@ -1,5 +1,5 @@ -use crate::executor::park::{Park, Unpark}; -use crate::executor::task::{self, JoinHandle, Schedule, Task}; +use crate::runtime::park::{Park, Unpark}; +use crate::runtime::task::{self, JoinHandle, Schedule, Task}; use std::cell::UnsafeCell; use std::collections::VecDeque; @@ -130,6 +130,7 @@ where where F: Future, { + use crate::runtime; use std::pin::Pin; use std::task::Context; use std::task::Poll::Ready; @@ -137,9 +138,9 @@ where let local = &mut self.local; let scheduler = &*self.scheduler; - crate::executor::global::with_current_thread(scheduler, || { + runtime::global::with_current_thread(scheduler, || { let mut _enter = - crate::executor::enter().expect("attempting to block while on a Tokio executor"); + runtime::enter::enter().expect("attempting to block while on a Tokio executor"); let raw_waker = RawWaker::new( scheduler as *const Scheduler as *const (), @@ -272,7 +273,7 @@ impl Schedule for Scheduler { } fn schedule(&self, task: Task) { - use crate::executor::global; + use crate::runtime::global; if global::current_thread_is_current(self) { unsafe { self.schedule_local(task) }; diff --git a/tokio/src/executor/enter.rs b/tokio/src/runtime/enter.rs similarity index 95% rename from tokio/src/executor/enter.rs rename to tokio/src/runtime/enter.rs index 19ae26e0c..78ddf8ccd 100644 --- a/tokio/src/executor/enter.rs +++ b/tokio/src/runtime/enter.rs @@ -1,7 +1,6 @@ use std::cell::{Cell, RefCell}; use std::error::Error; use std::fmt; -#[cfg(feature = "rt-full")] use std::future::Future; use std::marker::PhantomData; @@ -101,14 +100,13 @@ pub(crate) fn exit R, R>(f: F) -> R { impl Enter { /// Blocks the thread on the specified future, returning the value with /// which that future completes. - #[cfg(feature = "rt-full")] pub(crate) fn block_on(&mut self, mut f: F) -> F::Output { - use crate::executor::park::{Park, ParkThread}; + use crate::runtime::park::{CachedParkThread, Park}; use std::pin::Pin; use std::task::Context; use std::task::Poll::Ready; - let mut park = ParkThread::new(); + let mut park = CachedParkThread::new(); let waker = park.unpark().into_waker(); let mut cx = Context::from_waker(&waker); diff --git a/tokio/src/executor/global.rs b/tokio/src/runtime/global.rs similarity index 98% rename from tokio/src/executor/global.rs rename to tokio/src/runtime/global.rs index ddb9b1ab0..a1b1bed0d 100644 --- a/tokio/src/executor/global.rs +++ b/tokio/src/runtime/global.rs @@ -1,8 +1,8 @@ #[cfg(feature = "rt-current-thread")] -use crate::executor::current_thread; +use crate::runtime::current_thread; #[cfg(feature = "rt-full")] -use crate::executor::thread_pool; +use crate::runtime::thread_pool; use std::cell::Cell; use std::future::Future; diff --git a/tokio/src/runtime/io.rs b/tokio/src/runtime/io.rs new file mode 100644 index 000000000..a1ede23a5 --- /dev/null +++ b/tokio/src/runtime/io.rs @@ -0,0 +1,56 @@ +pub(crate) use self::variant::*; + +/// Re-exported for convenience. +pub(crate) use std::io::Result; + +#[cfg(feature = "net-driver")] +mod variant { + use crate::net::driver; + + use std::io; + + /// The driver value the runtime passes to the `timer` layer. + /// + /// When the `io-driver` feature is enabled, this is the "real" I/O driver + /// backed by Mio. Without the `io-driver` feature, this is a thread parker + /// backed by a condition variable. + pub(crate) type Driver = driver::Reactor; + + /// The handle the runtime stores for future use. + /// + /// When the `io-driver` feature is **not** enabled, this is `()`. + pub(crate) type Handle = driver::Handle; + + pub(crate) fn create() -> io::Result<(Driver, Handle)> { + let driver = driver::Reactor::new()?; + let handle = driver.handle(); + + Ok((driver, handle)) + } + + pub(crate) fn set_default(handle: &Handle) -> driver::DefaultGuard<'_> { + driver::set_default(handle) + } +} + +#[cfg(not(feature = "net-driver"))] +mod variant { + use crate::runtime::park::ParkThread; + + use std::io; + + /// I/O is not enabled, use a condition variable based parker + pub(crate) type Driver = ParkThread; + + /// There is no handle + pub(crate) type Handle = (); + + pub(crate) fn create() -> io::Result<(Driver, Handle)> { + let driver = ParkThread::new(); + + Ok((driver, ())) + } + + #[cfg(feature = "blocking")] + pub(crate) fn set_default(_handle: &Handle) {} +} diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index e351737c6..ac5bee974 100644 --- a/tokio/src/runtime/mod.rs +++ b/tokio/src/runtime/mod.rs @@ -128,24 +128,57 @@ //! [`tokio::spawn`]: ../executor/fn.spawn.html //! [`tokio::main`]: ../../tokio_macros/attr.main.html +// At the top due to macros +#[cfg(test)] +#[macro_use] +mod tests; + +#[cfg(all(not(feature = "blocking"), feature = "rt-full"))] +mod blocking; +#[cfg(feature = "blocking")] +pub mod blocking; + mod builder; pub use self::builder::Builder; +#[cfg(feature = "rt-current-thread")] +mod current_thread; + +#[cfg(feature = "blocking")] +mod enter; + +mod global; +pub use self::global::spawn; + +mod io; + +mod park; +pub use self::park::{Park, Unpark}; + +#[cfg(feature = "rt-current-thread")] mod spawner; +#[cfg(feature = "rt-current-thread")] pub use self::spawner::Spawner; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use crate::executor::{JoinError, JoinHandle}; +#[cfg(feature = "rt-current-thread")] +mod task; +#[cfg(feature = "rt-current-thread")] +pub use self::task::{JoinError, JoinHandle}; + +mod timer; -use crate::executor::blocking::{self, PoolWaiter}; -use crate::executor::current_thread::CurrentThread; #[cfg(feature = "rt-full")] -use crate::executor::thread_pool::ThreadPool; -use crate::net::{self, driver}; -use crate::timer::timer; +pub(crate) mod thread_pool; +#[cfg(feature = "blocking")] +use crate::runtime::blocking::PoolWaiter; +#[cfg(feature = "rt-current-thread")] +use crate::runtime::current_thread::CurrentThread; +#[cfg(feature = "rt-full")] +use crate::runtime::thread_pool::ThreadPool; + +#[cfg(feature = "blocking")] use std::future::Future; -use std::io; /// The Tokio runtime, includes a reactor as well as an executor for running /// tasks. @@ -178,21 +211,30 @@ pub struct Runtime { kind: Kind, /// Handles to the network drivers - net_handles: Vec, + net_handles: Vec, /// Timer handles timer_handles: Vec, /// Blocking pool handle + #[cfg(feature = "blocking")] blocking_pool: PoolWaiter, } /// The runtime executor is either a thread-pool or a current-thread executor. #[derive(Debug)] enum Kind { + /// Not able to execute concurrent tasks. This variant is mostly used to get + /// access to the driver handles. + Shell, + + /// Execute all tasks on the current-thread. + #[cfg(feature = "rt-current-thread")] + CurrentThread(CurrentThread), + + /// Execute tasks across multiple threads. #[cfg(feature = "rt-full")] ThreadPool(ThreadPool), - CurrentThread(CurrentThread>), } impl Runtime { @@ -222,7 +264,16 @@ impl Runtime { /// /// [mod]: index.html pub fn new() -> io::Result { - Builder::new().build() + #[cfg(feature = "rt-full")] + let ret = Builder::new().thread_pool().build(); + + #[cfg(all(not(feature = "rt-full"), feature = "rt-current-thread"))] + let ret = Builder::new().current_thread().build(); + + #[cfg(not(feature = "rt-current-thread"))] + let ret = Builder::new().build(); + + ret } /// Spawn a future onto the Tokio runtime. @@ -255,11 +306,13 @@ impl Runtime { /// /// This function panics if the spawn fails. Failure occurs if the executor /// is currently at capacity and is unable to spawn a new future. + #[cfg(feature = "rt-current-thread")] pub fn spawn(&self, future: F) -> JoinHandle where F: Future + Send + 'static, { match &self.kind { + Kind::Shell => panic!("task execution disabled"), #[cfg(feature = "rt-full")] Kind::ThreadPool(exec) => exec.spawn(future), Kind::CurrentThread(exec) => exec.spawn(future), @@ -279,18 +332,22 @@ impl Runtime { /// /// This function panics if the executor is at capacity, if the provided /// future panics, or if called within an asynchronous execution context. + #[cfg(feature = "blocking")] // TODO: remove this pub fn block_on(&mut self, future: F) -> F::Output { - let _net = driver::set_default(&self.net_handles[0]); + let _net = io::set_default(&self.net_handles[0]); let _timer = timer::set_default(&self.timer_handles[0]); let kind = &mut self.kind; - blocking::with_pool(&self.blocking_pool, || { - match kind { - #[cfg(feature = "rt-full")] - Kind::ThreadPool(exec) => exec.block_on(future), - Kind::CurrentThread(exec) => exec.block_on(future), + blocking::with_pool(&self.blocking_pool, || match kind { + Kind::Shell => { + let mut enter = enter::enter().unwrap(); + enter.block_on(future) } + #[cfg(feature = "rt-current-thread")] + Kind::CurrentThread(exec) => exec.block_on(future), + #[cfg(feature = "rt-full")] + Kind::ThreadPool(exec) => exec.block_on(future), }) } @@ -310,11 +367,14 @@ impl Runtime { /// /// spawner.spawn(async { println!("hello"); }); /// ``` + #[cfg(feature = "rt-current-thread")] pub fn spawner(&self) -> Spawner { match &self.kind { + Kind::Shell => Spawner::shell(), + #[cfg(feature = "rt-current-thread")] + Kind::CurrentThread(exec) => Spawner::current_thread(exec.spawner()), #[cfg(feature = "rt-full")] Kind::ThreadPool(exec) => Spawner::thread_pool(exec.spawner().clone()), - Kind::CurrentThread(exec) => Spawner::current_thread(exec.spawner()), } } } diff --git a/tokio/src/executor/park/mod.rs b/tokio/src/runtime/park/mod.rs similarity index 97% rename from tokio/src/executor/park/mod.rs rename to tokio/src/runtime/park/mod.rs index 9d6d508f6..b8fa81aba 100644 --- a/tokio/src/executor/park/mod.rs +++ b/tokio/src/runtime/park/mod.rs @@ -44,9 +44,10 @@ //! [up]: trait.Unpark.html //! [mio]: https://docs.rs/mio/0.6/mio/struct.Poll.html -#[cfg(feature = "rt-full")] mod thread; -#[cfg(feature = "rt-full")] +#[cfg(feature = "blocking")] +pub(crate) use self::thread::CachedParkThread; +#[cfg(not(feature = "net-driver"))] pub(crate) use self::thread::ParkThread; use std::sync::Arc; diff --git a/tokio/src/executor/park/thread.rs b/tokio/src/runtime/park/thread.rs similarity index 86% rename from tokio/src/executor/park/thread.rs rename to tokio/src/runtime/park/thread.rs index 880dfb494..beebe2a4b 100644 --- a/tokio/src/executor/park/thread.rs +++ b/tokio/src/runtime/park/thread.rs @@ -1,6 +1,6 @@ -use crate::executor::park::{Park, Unpark}; use crate::loom::sync::atomic::AtomicUsize; use crate::loom::sync::{Arc, Condvar, Mutex}; +use crate::runtime::park::{Park, Unpark}; use std::marker::PhantomData; use std::rc::Rc; @@ -18,10 +18,14 @@ use std::time::Duration; /// means that an instance of `ParkThread` might be unblocked by a handle /// associated with a different `ParkThread` instance. #[derive(Debug)] -pub(crate) struct ParkThread { +pub(crate) struct CachedParkThread { _anchor: PhantomData>, } +pub(crate) struct ParkThread { + inner: Arc, +} + /// Error returned by [`ParkThread`] /// /// This currently is never returned, but might at some point in the future. @@ -32,10 +36,6 @@ pub(crate) struct ParkError { _p: (), } -struct Parker { - unparker: Arc, -} - /// Unblocks a thread that was blocked by `ParkThread`. #[derive(Clone, Debug)] pub(crate) struct UnparkThread { @@ -54,32 +54,38 @@ const NOTIFY: usize = 1; const SLEEP: usize = 2; thread_local! { - static CURRENT_PARKER: Parker = Parker::new(); + static CURRENT_PARKER: ParkThread = ParkThread::new(); } -// ==== impl Parker ==== +// ==== impl ParkThread ==== -impl Parker { - fn new() -> Self { +impl ParkThread { + pub(crate) fn new() -> Self { Self { - unparker: Arc::new(Inner { + inner: Arc::new(Inner { state: AtomicUsize::new(IDLE), mutex: Mutex::new(()), condvar: Condvar::new(), }), } } +} - fn unparker(&self) -> &Arc { - &self.unparker +impl Park for ParkThread { + type Unpark = UnparkThread; + type Error = ParkError; + + fn unpark(&self) -> Self::Unpark { + let inner = self.inner.clone(); + UnparkThread { inner } } - fn park(&self) -> Result<(), ParkError> { - self.unparker.park(None) + fn park(&mut self) -> Result<(), Self::Error> { + self.inner.park(None) } - fn park_timeout(&self, timeout: Duration) -> Result<(), ParkError> { - self.unparker.park(Some(timeout)) + fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> { + self.inner.park(Some(duration)) } } @@ -156,13 +162,14 @@ impl Inner { // ===== impl ParkThread ===== -impl ParkThread { +impl CachedParkThread { /// Create a new `ParkThread` handle for the current thread. /// /// This type cannot be moved to other threads, so it should be created on /// the thread that the caller intends to park. - pub(crate) fn new() -> ParkThread { - ParkThread { + #[cfg(feature = "blocking")] + pub(crate) fn new() -> CachedParkThread { + CachedParkThread { _anchor: PhantomData, } } @@ -170,28 +177,27 @@ impl ParkThread { /// Get a reference to the `ParkThread` handle for this thread. fn with_current(&self, f: F) -> R where - F: FnOnce(&Parker) -> R, + F: FnOnce(&ParkThread) -> R, { CURRENT_PARKER.with(|inner| f(inner)) } } -impl Park for ParkThread { +impl Park for CachedParkThread { type Unpark = UnparkThread; type Error = ParkError; fn unpark(&self) -> Self::Unpark { - let inner = self.with_current(|inner| inner.unparker().clone()); - UnparkThread { inner } + self.with_current(|park_thread| park_thread.unpark()) } fn park(&mut self) -> Result<(), Self::Error> { - self.with_current(|inner| inner.park())?; + self.with_current(|park_thread| park_thread.inner.park(None))?; Ok(()) } fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> { - self.with_current(|inner| inner.park_timeout(duration))?; + self.with_current(|park_thread| park_thread.inner.park(Some(duration)))?; Ok(()) } } @@ -210,7 +216,7 @@ impl Unpark for UnparkThread { } } -#[cfg(feature = "rt-full")] +#[cfg(feature = "blocking")] mod waker { use super::{Inner, UnparkThread}; use crate::loom::sync::Arc; diff --git a/tokio/src/runtime/spawner.rs b/tokio/src/runtime/spawner.rs index c5a076154..678f09912 100644 --- a/tokio/src/runtime/spawner.rs +++ b/tokio/src/runtime/spawner.rs @@ -1,6 +1,6 @@ -use crate::executor::current_thread; +use crate::runtime::current_thread; #[cfg(feature = "rt-full")] -use crate::executor::thread_pool; +use crate::runtime::thread_pool; use crate::runtime::JoinHandle; use std::future::Future; @@ -13,24 +13,33 @@ use std::future::Future; /// For more details, see the [module level](index.html) documentation. #[derive(Debug, Clone)] pub struct Spawner { - kind: Kind + kind: Kind, } #[derive(Debug, Clone)] enum Kind { + Shell, #[cfg(feature = "rt-full")] ThreadPool(thread_pool::Spawner), CurrentThread(current_thread::Spawner), } impl Spawner { + pub(super) fn shell() -> Spawner { + Spawner { kind: Kind::Shell } + } + #[cfg(feature = "rt-full")] pub(super) fn thread_pool(spawner: thread_pool::Spawner) -> Spawner { - Spawner { kind: Kind::ThreadPool(spawner) } + Spawner { + kind: Kind::ThreadPool(spawner), + } } pub(super) fn current_thread(spawner: current_thread::Spawner) -> Spawner { - Spawner { kind: Kind::CurrentThread(spawner) } + Spawner { + kind: Kind::CurrentThread(spawner), + } } /// Spawn a future onto the Tokio runtime. @@ -69,6 +78,7 @@ impl Spawner { F: Future + Send + 'static, { match &self.kind { + Kind::Shell => panic!("spawning not enabled for runtime"), #[cfg(feature = "rt-full")] Kind::ThreadPool(spawner) => spawner.spawn(future), Kind::CurrentThread(spawner) => spawner.spawn(future), diff --git a/tokio/src/executor/task/core.rs b/tokio/src/runtime/task/core.rs similarity index 96% rename from tokio/src/executor/task/core.rs rename to tokio/src/runtime/task/core.rs index 7a3016ee4..957a4a836 100644 --- a/tokio/src/executor/task/core.rs +++ b/tokio/src/runtime/task/core.rs @@ -1,9 +1,9 @@ -use crate::executor::task::raw::{self, Vtable}; -use crate::executor::task::state::State; -use crate::executor::task::waker::waker_ref; -use crate::executor::task::Schedule; use crate::loom::alloc::Track; use crate::loom::cell::CausalCell; +use crate::runtime::task::raw::{self, Vtable}; +use crate::runtime::task::state::State; +use crate::runtime::task::waker::waker_ref; +use crate::runtime::task::Schedule; use std::cell::UnsafeCell; use std::future::Future; diff --git a/tokio/src/executor/task/error.rs b/tokio/src/runtime/task/error.rs similarity index 100% rename from tokio/src/executor/task/error.rs rename to tokio/src/runtime/task/error.rs diff --git a/tokio/src/executor/task/harness.rs b/tokio/src/runtime/task/harness.rs similarity index 99% rename from tokio/src/executor/task/harness.rs rename to tokio/src/runtime/task/harness.rs index 56b3cfbb4..39cda5dd0 100644 --- a/tokio/src/executor/task/harness.rs +++ b/tokio/src/runtime/task/harness.rs @@ -1,8 +1,8 @@ -use crate::executor::task::core::{Cell, Core, Header, Trailer}; -use crate::executor::task::state::Snapshot; -use crate::executor::task::{JoinError, Schedule, Task}; use crate::loom::alloc::Track; use crate::loom::cell::CausalCheck; +use crate::runtime::task::core::{Cell, Core, Header, Trailer}; +use crate::runtime::task::state::Snapshot; +use crate::runtime::task::{JoinError, Schedule, Task}; use std::future::Future; use std::marker::PhantomData; diff --git a/tokio/src/executor/task/join.rs b/tokio/src/runtime/task/join.rs similarity index 98% rename from tokio/src/executor/task/join.rs rename to tokio/src/runtime/task/join.rs index faf6eb56c..a59cfde84 100644 --- a/tokio/src/executor/task/join.rs +++ b/tokio/src/runtime/task/join.rs @@ -1,5 +1,5 @@ -use crate::executor::task::raw::RawTask; use crate::loom::alloc::Track; +use crate::runtime::task::RawTask; use std::fmt; use std::future::Future; diff --git a/tokio/src/executor/task/list.rs b/tokio/src/runtime/task/list.rs similarity index 98% rename from tokio/src/executor/task/list.rs rename to tokio/src/runtime/task/list.rs index 0ff887a70..7ccc57950 100644 --- a/tokio/src/executor/task/list.rs +++ b/tokio/src/runtime/task/list.rs @@ -1,4 +1,4 @@ -use crate::executor::task::{Header, Task}; +use crate::runtime::task::{Header, Task}; use std::fmt; use std::marker::PhantomData; diff --git a/tokio/src/executor/task/mod.rs b/tokio/src/runtime/task/mod.rs similarity index 97% rename from tokio/src/executor/task/mod.rs rename to tokio/src/runtime/task/mod.rs index 14d378a88..7e361e4fb 100644 --- a/tokio/src/executor/task/mod.rs +++ b/tokio/src/runtime/task/mod.rs @@ -1,4 +1,5 @@ mod core; +use self::core::Cell; pub(crate) use self::core::Header; mod error; @@ -6,6 +7,7 @@ mod error; pub use self::error::JoinError; mod harness; +use self::harness::Harness; mod join; #[cfg(any(feature = "rt-current-thread", feature = "rt-full"))] @@ -16,19 +18,20 @@ mod list; pub(crate) use self::list::OwnedList; mod raw; +use self::raw::RawTask; mod stack; pub(crate) use self::stack::TransferStack; mod state; +use self::state::{Snapshot, State}; + mod waker; /// Unit tests #[cfg(test)] mod tests; -use self::raw::RawTask; - use std::future::Future; use std::marker::PhantomData; use std::ptr::NonNull; diff --git a/tokio/src/executor/task/raw.rs b/tokio/src/runtime/task/raw.rs similarity index 96% rename from tokio/src/executor/task/raw.rs rename to tokio/src/runtime/task/raw.rs index 6a34a3d41..51d9d41c2 100644 --- a/tokio/src/executor/task/raw.rs +++ b/tokio/src/runtime/task/raw.rs @@ -1,8 +1,8 @@ -use crate::executor::task::core::Cell; -use crate::executor::task::harness::Harness; -use crate::executor::task::state::{Snapshot, State}; -use crate::executor::task::{Header, Schedule}; use crate::loom::alloc::Track; +use crate::runtime::task::Cell; +use crate::runtime::task::Harness; +use crate::runtime::task::{Header, Schedule}; +use crate::runtime::task::{Snapshot, State}; use std::future::Future; use std::ptr::NonNull; diff --git a/tokio/src/executor/task/stack.rs b/tokio/src/runtime/task/stack.rs similarity index 98% rename from tokio/src/executor/task/stack.rs rename to tokio/src/runtime/task/stack.rs index cd675e10d..7e13d7d57 100644 --- a/tokio/src/executor/task/stack.rs +++ b/tokio/src/runtime/task/stack.rs @@ -1,5 +1,5 @@ -use crate::executor::task::{Header, Task}; use crate::loom::sync::atomic::AtomicPtr; +use crate::runtime::task::{Header, Task}; use std::marker::PhantomData; use std::ptr::{self, NonNull}; diff --git a/tokio/src/executor/task/state.rs b/tokio/src/runtime/task/state.rs similarity index 100% rename from tokio/src/executor/task/state.rs rename to tokio/src/runtime/task/state.rs diff --git a/tokio/src/executor/task/tests/loom.rs b/tokio/src/runtime/task/tests/loom.rs similarity index 98% rename from tokio/src/executor/task/tests/loom.rs rename to tokio/src/runtime/task/tests/loom.rs index b377aeb99..b62e0f36e 100644 --- a/tokio/src/executor/task/tests/loom.rs +++ b/tokio/src/runtime/task/tests/loom.rs @@ -1,5 +1,5 @@ -use crate::executor::task; -use crate::executor::tests::loom_schedule::LoomSchedule; +use crate::runtime::task; +use crate::runtime::tests::loom_schedule::LoomSchedule; use tokio_test::{assert_err, assert_ok}; diff --git a/tokio/src/executor/task/tests/mod.rs b/tokio/src/runtime/task/tests/mod.rs similarity index 100% rename from tokio/src/executor/task/tests/mod.rs rename to tokio/src/runtime/task/tests/mod.rs diff --git a/tokio/src/executor/task/tests/task.rs b/tokio/src/runtime/task/tests/task.rs similarity index 98% rename from tokio/src/executor/task/tests/task.rs rename to tokio/src/runtime/task/tests/task.rs index ab942af7b..4b2dd2d89 100644 --- a/tokio/src/executor/task/tests/task.rs +++ b/tokio/src/runtime/task/tests/task.rs @@ -1,7 +1,7 @@ -use crate::executor::task::{self, Header}; -use crate::executor::tests::backoff::*; -use crate::executor::tests::mock_schedule::{mock, Mock}; -use crate::executor::tests::track_drop::track_drop; +use crate::runtime::task::{self, Header}; +use crate::runtime::tests::backoff::*; +use crate::runtime::tests::mock_schedule::{mock, Mock}; +use crate::runtime::tests::track_drop::track_drop; use crate::sync::oneshot; use tokio_test::task::spawn; diff --git a/tokio/src/executor/task/waker.rs b/tokio/src/runtime/task/waker.rs similarity index 96% rename from tokio/src/executor/task/waker.rs rename to tokio/src/runtime/task/waker.rs index 6bc121449..00c1e6d46 100644 --- a/tokio/src/executor/task/waker.rs +++ b/tokio/src/runtime/task/waker.rs @@ -1,5 +1,5 @@ -use crate::executor::task::harness::Harness; -use crate::executor::task::{Header, Schedule}; +use crate::runtime::task::harness::Harness; +use crate::runtime::task::{Header, Schedule}; use std::future::Future; use std::marker::PhantomData; diff --git a/tokio/src/executor/tests/backoff.rs b/tokio/src/runtime/tests/backoff.rs similarity index 100% rename from tokio/src/executor/tests/backoff.rs rename to tokio/src/runtime/tests/backoff.rs diff --git a/tokio/src/executor/tests/loom_oneshot.rs b/tokio/src/runtime/tests/loom_oneshot.rs similarity index 100% rename from tokio/src/executor/tests/loom_oneshot.rs rename to tokio/src/runtime/tests/loom_oneshot.rs diff --git a/tokio/src/executor/tests/loom_schedule.rs b/tokio/src/runtime/tests/loom_schedule.rs similarity index 96% rename from tokio/src/executor/tests/loom_schedule.rs rename to tokio/src/runtime/tests/loom_schedule.rs index 7999dd974..1dad60621 100644 --- a/tokio/src/executor/tests/loom_schedule.rs +++ b/tokio/src/runtime/tests/loom_schedule.rs @@ -1,4 +1,4 @@ -use crate::executor::task::{Schedule, Task}; +use crate::runtime::task::{Schedule, Task}; use loom::sync::Notify; use std::collections::VecDeque; diff --git a/tokio/src/executor/tests/mock_park.rs b/tokio/src/runtime/tests/mock_park.rs similarity index 96% rename from tokio/src/executor/tests/mock_park.rs rename to tokio/src/runtime/tests/mock_park.rs index 2bc9edf5e..0fe28b361 100644 --- a/tokio/src/executor/tests/mock_park.rs +++ b/tokio/src/runtime/tests/mock_park.rs @@ -1,6 +1,6 @@ #![allow(warnings)] -use crate::executor::park::{Park, Unpark}; +use crate::runtime::{Park, Unpark}; use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering::SeqCst}; diff --git a/tokio/src/executor/tests/mock_schedule.rs b/tokio/src/runtime/tests/mock_schedule.rs similarity index 98% rename from tokio/src/executor/tests/mock_schedule.rs rename to tokio/src/runtime/tests/mock_schedule.rs index daf66fa3a..ab15c54e2 100644 --- a/tokio/src/executor/tests/mock_schedule.rs +++ b/tokio/src/runtime/tests/mock_schedule.rs @@ -1,6 +1,6 @@ #![allow(warnings)] -use crate::executor::task::{Header, Schedule, Task}; +use crate::runtime::task::{Header, Schedule, Task}; use std::collections::VecDeque; use std::sync::Mutex; diff --git a/tokio/src/executor/tests/mod.rs b/tokio/src/runtime/tests/mod.rs similarity index 100% rename from tokio/src/executor/tests/mod.rs rename to tokio/src/runtime/tests/mod.rs diff --git a/tokio/src/executor/tests/track_drop.rs b/tokio/src/runtime/tests/track_drop.rs similarity index 100% rename from tokio/src/executor/tests/track_drop.rs rename to tokio/src/runtime/tests/track_drop.rs diff --git a/tokio/src/executor/thread_pool/builder.rs b/tokio/src/runtime/thread_pool/builder.rs similarity index 93% rename from tokio/src/executor/thread_pool/builder.rs rename to tokio/src/runtime/thread_pool/builder.rs index c1b1ed19a..04b5fa238 100644 --- a/tokio/src/executor/thread_pool/builder.rs +++ b/tokio/src/runtime/thread_pool/builder.rs @@ -1,7 +1,7 @@ -use crate::executor::park::Park; -use crate::executor::thread_pool::{shutdown, worker, worker::Worker, Spawner, ThreadPool}; use crate::loom::sync::Arc; use crate::loom::sys::num_cpus; +use crate::runtime::park::Park; +use crate::runtime::thread_pool::{shutdown, Spawner, ThreadPool, Worker}; use std::{fmt, usize}; @@ -90,6 +90,8 @@ impl Builder { F: FnMut(usize) -> P, P: Park + Send + 'static, { + use crate::runtime::thread_pool::worker; + let (shutdown_tx, shutdown_rx) = shutdown::channel(); let around_worker = self.around_worker.as_ref().map(Arc::clone); @@ -134,7 +136,7 @@ impl Builder { }) as Box>) -> Box + Send + Sync>); - let mut blocking = crate::executor::blocking::Builder::default(); + let mut blocking = crate::runtime::blocking::Builder::default(); blocking.name(self.name.clone()); if let Some(ss) = self.stack_size { blocking.stack_size(ss); @@ -150,11 +152,11 @@ impl Builder { // Spawn threads for each worker for worker in workers { - crate::executor::blocking::Pool::spawn(&blocking, launch_worker(worker)) + crate::runtime::blocking::Pool::spawn(&blocking, launch_worker(worker)) } let spawner = Spawner::new(pool); - let blocking = crate::executor::blocking::PoolWaiter::from(blocking); + let blocking = crate::runtime::blocking::PoolWaiter::from(blocking); ThreadPool::from_parts(spawner, shutdown_rx, blocking) } } @@ -189,7 +191,7 @@ impl

Park for BoxedPark

where P: Park, { - type Unpark = Box; + type Unpark = Box; type Error = P::Error; fn unpark(&self) -> Self::Unpark { diff --git a/tokio/src/executor/thread_pool/current.rs b/tokio/src/runtime/thread_pool/current.rs similarity index 95% rename from tokio/src/executor/thread_pool/current.rs rename to tokio/src/runtime/thread_pool/current.rs index 7bbe3e34a..1a8113e37 100644 --- a/tokio/src/executor/thread_pool/current.rs +++ b/tokio/src/runtime/thread_pool/current.rs @@ -1,6 +1,6 @@ -use crate::executor::park::Unpark; -use crate::executor::thread_pool::{worker, Owned}; use crate::loom::sync::Arc; +use crate::runtime::park::Unpark; +use crate::runtime::thread_pool::{worker, Owned}; use std::cell::Cell; use std::ptr; diff --git a/tokio/src/executor/thread_pool/idle.rs b/tokio/src/runtime/thread_pool/idle.rs similarity index 100% rename from tokio/src/executor/thread_pool/idle.rs rename to tokio/src/runtime/thread_pool/idle.rs diff --git a/tokio/src/executor/thread_pool/mod.rs b/tokio/src/runtime/thread_pool/mod.rs similarity index 96% rename from tokio/src/executor/thread_pool/mod.rs rename to tokio/src/runtime/thread_pool/mod.rs index 0c7ce42ae..5dce2a10f 100644 --- a/tokio/src/executor/thread_pool/mod.rs +++ b/tokio/src/runtime/thread_pool/mod.rs @@ -27,6 +27,8 @@ use self::shared::Shared; mod shutdown; mod worker; +use self::worker::Worker; + #[cfg(feature = "blocking")] pub(crate) use worker::blocking; diff --git a/tokio/src/executor/thread_pool/owned.rs b/tokio/src/runtime/thread_pool/owned.rs similarity index 94% rename from tokio/src/executor/thread_pool/owned.rs rename to tokio/src/runtime/thread_pool/owned.rs index f0aece571..04a2f9316 100644 --- a/tokio/src/executor/thread_pool/owned.rs +++ b/tokio/src/runtime/thread_pool/owned.rs @@ -1,6 +1,6 @@ -use crate::executor::task::{self, Task}; -use crate::executor::thread_pool::{queue, Shared}; -use crate::executor::util::FastRand; +use crate::runtime::task::{self, Task}; +use crate::runtime::thread_pool::{queue, Shared}; +use crate::util::FastRand; use std::cell::Cell; diff --git a/tokio/src/executor/thread_pool/pool.rs b/tokio/src/runtime/thread_pool/pool.rs similarity index 81% rename from tokio/src/executor/thread_pool/pool.rs rename to tokio/src/runtime/thread_pool/pool.rs index d3219d878..06d5a61ae 100644 --- a/tokio/src/executor/thread_pool/pool.rs +++ b/tokio/src/runtime/thread_pool/pool.rs @@ -1,6 +1,6 @@ -use crate::executor::blocking::PoolWaiter; -use crate::executor::task::JoinHandle; -use crate::executor::thread_pool::{shutdown, Spawner}; +use crate::runtime::blocking::PoolWaiter; +use crate::runtime::task::JoinHandle; +use crate::runtime::thread_pool::{shutdown, Spawner}; use std::fmt; use std::future::Future; @@ -54,10 +54,10 @@ impl ThreadPool { where F: Future, { - crate::executor::global::with_thread_pool(self.spawner(), || { - let mut enter = - crate::executor::enter().expect("attempting to block while on a Tokio executor"); - crate::executor::blocking::with_pool(self.spawner.blocking_pool(), || { + crate::runtime::global::with_thread_pool(self.spawner(), || { + let mut enter = crate::runtime::enter::enter() + .expect("attempting to block while on a Tokio executor"); + crate::runtime::blocking::with_pool(self.spawner.blocking_pool(), || { enter.block_on(future) }) }) diff --git a/tokio/src/executor/thread_pool/queue/global.rs b/tokio/src/runtime/thread_pool/queue/global.rs similarity index 99% rename from tokio/src/executor/thread_pool/queue/global.rs rename to tokio/src/runtime/thread_pool/queue/global.rs index 20a9d36f8..edac6edea 100644 --- a/tokio/src/executor/thread_pool/queue/global.rs +++ b/tokio/src/runtime/thread_pool/queue/global.rs @@ -1,6 +1,6 @@ -use crate::executor::task::{Header, Task}; use crate::loom::sync::atomic::AtomicUsize; use crate::loom::sync::Mutex; +use crate::runtime::task::{Header, Task}; use std::marker::PhantomData; use std::ptr::{self, NonNull}; diff --git a/tokio/src/executor/thread_pool/queue/inject.rs b/tokio/src/runtime/thread_pool/queue/inject.rs similarity index 92% rename from tokio/src/executor/thread_pool/queue/inject.rs rename to tokio/src/runtime/thread_pool/queue/inject.rs index 045ce9227..f0f92fb2c 100644 --- a/tokio/src/executor/thread_pool/queue/inject.rs +++ b/tokio/src/runtime/thread_pool/queue/inject.rs @@ -1,6 +1,6 @@ -use crate::executor::task::Task; -use crate::executor::thread_pool::queue::Cluster; use crate::loom::sync::Arc; +use crate::runtime::task::Task; +use crate::runtime::thread_pool::queue::Cluster; pub(crate) struct Inject { cluster: Arc>, diff --git a/tokio/src/executor/thread_pool/queue/local.rs b/tokio/src/runtime/thread_pool/queue/local.rs similarity index 98% rename from tokio/src/executor/thread_pool/queue/local.rs rename to tokio/src/runtime/thread_pool/queue/local.rs index 62472b0bf..14f34832f 100644 --- a/tokio/src/executor/thread_pool/queue/local.rs +++ b/tokio/src/runtime/thread_pool/queue/local.rs @@ -1,8 +1,8 @@ -use crate::executor::task::Task; -use crate::executor::thread_pool::queue::global; -use crate::executor::thread_pool::LOCAL_QUEUE_CAPACITY; use crate::loom::cell::{CausalCell, CausalCheck}; use crate::loom::sync::atomic::{self, AtomicU32}; +use crate::runtime::task::Task; +use crate::runtime::thread_pool::queue::global; +use crate::runtime::thread_pool::LOCAL_QUEUE_CAPACITY; use std::fmt; use std::mem::MaybeUninit; diff --git a/tokio/src/executor/thread_pool/queue/mod.rs b/tokio/src/runtime/thread_pool/queue/mod.rs similarity index 100% rename from tokio/src/executor/thread_pool/queue/mod.rs rename to tokio/src/runtime/thread_pool/queue/mod.rs diff --git a/tokio/src/executor/thread_pool/queue/worker.rs b/tokio/src/runtime/thread_pool/queue/worker.rs similarity index 97% rename from tokio/src/executor/thread_pool/queue/worker.rs rename to tokio/src/runtime/thread_pool/queue/worker.rs index be2552252..67a2a1b85 100644 --- a/tokio/src/executor/thread_pool/queue/worker.rs +++ b/tokio/src/runtime/thread_pool/queue/worker.rs @@ -1,6 +1,6 @@ -use crate::executor::task::Task; -use crate::executor::thread_pool::queue::{local, Cluster, Inject}; use crate::loom::sync::Arc; +use crate::runtime::task::Task; +use crate::runtime::thread_pool::queue::{local, Cluster, Inject}; use std::cell::Cell; use std::fmt; diff --git a/tokio/src/executor/thread_pool/set.rs b/tokio/src/runtime/thread_pool/set.rs similarity index 92% rename from tokio/src/executor/thread_pool/set.rs rename to tokio/src/runtime/thread_pool/set.rs index c55f5d7d3..73555f821 100644 --- a/tokio/src/executor/thread_pool/set.rs +++ b/tokio/src/runtime/thread_pool/set.rs @@ -2,12 +2,12 @@ //! //! - Attempt to spin. -use crate::executor::park::Unpark; -use crate::executor::task::{self, JoinHandle, Task}; -use crate::executor::thread_pool::{current, queue, Idle, Owned, Shared}; -use crate::executor::util::{CachePadded, FastRand}; use crate::loom::rand::seed; use crate::loom::sync::Arc; +use crate::runtime::park::Unpark; +use crate::runtime::task::{self, JoinHandle, Task}; +use crate::runtime::thread_pool::{current, queue, Idle, Owned, Shared}; +use crate::util::{CachePadded, FastRand}; use std::cell::UnsafeCell; use std::future::Future; @@ -29,7 +29,7 @@ where idle: Idle, /// Pool where blocking tasks should be spawned. - pub(crate) blocking: Arc, + pub(crate) blocking: Arc, } unsafe impl Send for Set

{} @@ -43,7 +43,7 @@ where pub(crate) fn new( num_workers: usize, mut mk_unpark: F, - blocking: Arc, + blocking: Arc, ) -> Self where F: FnMut(usize) -> P, @@ -112,7 +112,7 @@ where self.schedule(task); } - pub(super) fn blocking_pool(&self) -> &Arc { + pub(super) fn blocking_pool(&self) -> &Arc { &self.blocking } diff --git a/tokio/src/executor/thread_pool/shared.rs b/tokio/src/runtime/thread_pool/shared.rs similarity index 94% rename from tokio/src/executor/thread_pool/shared.rs rename to tokio/src/runtime/thread_pool/shared.rs index bf19e1971..e0a1987e2 100644 --- a/tokio/src/executor/thread_pool/shared.rs +++ b/tokio/src/runtime/thread_pool/shared.rs @@ -1,6 +1,6 @@ -use crate::executor::park::Unpark; -use crate::executor::task::{self, Schedule, Task}; -use crate::executor::thread_pool::worker; +use crate::runtime::park::Unpark; +use crate::runtime::task::{self, Schedule, Task}; +use crate::runtime::thread_pool::worker; use std::ptr; diff --git a/tokio/src/executor/thread_pool/shutdown.rs b/tokio/src/runtime/thread_pool/shutdown.rs similarity index 96% rename from tokio/src/executor/thread_pool/shutdown.rs rename to tokio/src/runtime/thread_pool/shutdown.rs index 9a0ee479d..7f5548983 100644 --- a/tokio/src/executor/thread_pool/shutdown.rs +++ b/tokio/src/runtime/thread_pool/shutdown.rs @@ -27,7 +27,7 @@ pub(super) fn channel() -> (Sender, Receiver) { impl Receiver { /// Block the current thread until all `Sender` handles drop. pub(crate) fn wait(&mut self) { - use crate::executor::enter; + use crate::runtime::enter::enter; let mut e = match enter() { Ok(e) => e, diff --git a/tokio/src/executor/thread_pool/spawner.rs b/tokio/src/runtime/thread_pool/spawner.rs similarity index 89% rename from tokio/src/executor/thread_pool/spawner.rs rename to tokio/src/runtime/thread_pool/spawner.rs index 28ebe1646..dd80f4a67 100644 --- a/tokio/src/executor/thread_pool/spawner.rs +++ b/tokio/src/runtime/thread_pool/spawner.rs @@ -1,7 +1,7 @@ -use crate::executor::park::Unpark; -use crate::executor::task::JoinHandle; -use crate::executor::thread_pool::worker; use crate::loom::sync::Arc; +use crate::runtime::park::Unpark; +use crate::runtime::task::JoinHandle; +use crate::runtime::thread_pool::worker; use std::fmt; use std::future::Future; @@ -45,7 +45,7 @@ impl Spawner { self.workers.spawn_background(future); } - pub(super) fn blocking_pool(&self) -> &Arc { + pub(super) fn blocking_pool(&self) -> &Arc { self.workers.blocking_pool() } diff --git a/tokio/src/executor/thread_pool/tests/loom_pool.rs b/tokio/src/runtime/thread_pool/tests/loom_pool.rs similarity index 97% rename from tokio/src/executor/thread_pool/tests/loom_pool.rs rename to tokio/src/runtime/thread_pool/tests/loom_pool.rs index fff00d677..b7be932cc 100644 --- a/tokio/src/executor/thread_pool/tests/loom_pool.rs +++ b/tokio/src/runtime/thread_pool/tests/loom_pool.rs @@ -1,6 +1,6 @@ -use crate::executor::park::{Park, Unpark}; -use crate::executor::tests::loom_oneshot as oneshot; -use crate::executor::thread_pool::{self, Builder}; +use crate::runtime::tests::loom_oneshot as oneshot; +use crate::runtime::thread_pool::{self, Builder}; +use crate::runtime::{Park, Unpark}; use crate::spawn; use loom::sync::atomic::{AtomicBool, AtomicUsize}; diff --git a/tokio/src/executor/thread_pool/tests/loom_queue.rs b/tokio/src/runtime/thread_pool/tests/loom_queue.rs similarity index 91% rename from tokio/src/executor/thread_pool/tests/loom_queue.rs rename to tokio/src/runtime/thread_pool/tests/loom_queue.rs index cc9ae4494..b7c86f6c9 100644 --- a/tokio/src/executor/thread_pool/tests/loom_queue.rs +++ b/tokio/src/runtime/thread_pool/tests/loom_queue.rs @@ -1,6 +1,6 @@ -use crate::executor::task::{self, Task}; -use crate::executor::tests::mock_schedule::{Noop, NOOP_SCHEDULE}; -use crate::executor::thread_pool::queue; +use crate::runtime::task::{self, Task}; +use crate::runtime::tests::mock_schedule::{Noop, NOOP_SCHEDULE}; +use crate::runtime::thread_pool::queue; use loom::thread; diff --git a/tokio/src/executor/thread_pool/tests/mod.rs b/tokio/src/runtime/thread_pool/tests/mod.rs similarity index 100% rename from tokio/src/executor/thread_pool/tests/mod.rs rename to tokio/src/runtime/thread_pool/tests/mod.rs diff --git a/tokio/src/executor/thread_pool/tests/pool.rs b/tokio/src/runtime/thread_pool/tests/pool.rs similarity index 98% rename from tokio/src/executor/thread_pool/tests/pool.rs rename to tokio/src/runtime/thread_pool/tests/pool.rs index 90a2eda78..050ffc7d0 100644 --- a/tokio/src/executor/thread_pool/tests/pool.rs +++ b/tokio/src/runtime/thread_pool/tests/pool.rs @@ -1,7 +1,6 @@ #![warn(rust_2018_idioms)] -use crate::executor::park::{Park, Unpark}; -use crate::executor::thread_pool; +use crate::runtime::{thread_pool, Park, Unpark}; use futures_util::future::poll_fn; use std::future::Future; diff --git a/tokio/src/executor/thread_pool/tests/queue.rs b/tokio/src/runtime/thread_pool/tests/queue.rs similarity index 97% rename from tokio/src/executor/thread_pool/tests/queue.rs rename to tokio/src/runtime/thread_pool/tests/queue.rs index 51b653065..86f32ed29 100644 --- a/tokio/src/executor/thread_pool/tests/queue.rs +++ b/tokio/src/runtime/thread_pool/tests/queue.rs @@ -1,6 +1,6 @@ -use crate::executor::task::{self, Task}; -use crate::executor::tests::mock_schedule::{Noop, NOOP_SCHEDULE}; -use crate::executor::thread_pool::{queue, LOCAL_QUEUE_CAPACITY}; +use crate::runtime::task::{self, Task}; +use crate::runtime::tests::mock_schedule::{Noop, NOOP_SCHEDULE}; +use crate::runtime::thread_pool::{queue, LOCAL_QUEUE_CAPACITY}; macro_rules! assert_pop { ($q:expr, $expect:expr) => { diff --git a/tokio/src/executor/thread_pool/tests/worker.rs b/tokio/src/runtime/thread_pool/tests/worker.rs similarity index 86% rename from tokio/src/executor/thread_pool/tests/worker.rs rename to tokio/src/runtime/thread_pool/tests/worker.rs index 20b640c86..edb971604 100644 --- a/tokio/src/executor/thread_pool/tests/worker.rs +++ b/tokio/src/runtime/thread_pool/tests/worker.rs @@ -1,5 +1,5 @@ -use crate::executor::tests::track_drop::track_drop; -use crate::executor::thread_pool; +use crate::runtime::tests::track_drop::track_drop; +use crate::runtime::thread_pool; use tokio_test::assert_ok; @@ -11,8 +11,8 @@ macro_rules! pool { (pool, w.remove(0), w.remove(0), mock_park) }}; (! $n:expr) => {{ - let mut mock_park = crate::executor::tests::mock_park::MockPark::new(); - let blocking = std::sync::Arc::new(crate::executor::blocking::Pool::default()); + let mut mock_park = crate::runtime::tests::mock_park::MockPark::new(); + let blocking = std::sync::Arc::new(crate::runtime::blocking::Pool::default()); let (pool, workers) = thread_pool::worker::create_set( $n, |index| Box::new(mock_park.mk_park(index)), diff --git a/tokio/src/executor/thread_pool/worker.rs b/tokio/src/runtime/thread_pool/worker.rs similarity index 97% rename from tokio/src/executor/thread_pool/worker.rs rename to tokio/src/runtime/thread_pool/worker.rs index 10aeeae4e..462fb2e86 100644 --- a/tokio/src/executor/thread_pool/worker.rs +++ b/tokio/src/runtime/thread_pool/worker.rs @@ -1,7 +1,7 @@ -use crate::executor::park::{Park, Unpark}; -use crate::executor::task::Task; -use crate::executor::thread_pool::{current, Owned, Shared, Spawner}; use crate::loom::sync::Arc; +use crate::runtime::park::{Park, Unpark}; +use crate::runtime::task::Task; +use crate::runtime::thread_pool::{current, Owned, Shared, Spawner}; use std::cell::Cell; use std::ops::{Deref, DerefMut}; @@ -38,7 +38,7 @@ where } // TODO: remove this re-export -pub(super) use crate::executor::thread_pool::set::Set; +pub(super) use crate::runtime::thread_pool::set::Set; pub(crate) struct Worker { /// Entry in the set of workers. @@ -58,7 +58,7 @@ pub(super) fn create_set( pool_size: usize, mk_park: F, launch_worker: LaunchWorker

, - blocking: Arc, + blocking: Arc, ) -> (Arc>, Vec>) where P: Send + Park, @@ -131,10 +131,11 @@ where // Track the current worker current::set(&pool, index, || { - let _enter = crate::executor::enter().expect("executor already running on thread"); + let _enter = + crate::runtime::enter::enter().expect("executor already running on thread"); - crate::executor::global::with_thread_pool(&spawner, || { - crate::executor::blocking::with_pool(blocking, || { + crate::runtime::global::with_thread_pool(&spawner, || { + crate::runtime::blocking::with_pool(blocking, || { ON_BLOCK.with(|ob| { // Ensure that the ON_BLOCK is removed from the thread-local context // when leaving the scope. This handles cases that involve panicking. @@ -220,7 +221,7 @@ where // instances of the Worker, and compare_exchange it to true afterwards // in an attempt to take it back. if it succeeds, we just resume where // we were. if it fails, another thread has already stolen the Worker. - crate::executor::blocking::Pool::spawn( + crate::runtime::blocking::Pool::spawn( &pool.blocking, launch_worker(worker), ); diff --git a/tokio/src/runtime/timer.rs b/tokio/src/runtime/timer.rs new file mode 100644 index 000000000..987324cd2 --- /dev/null +++ b/tokio/src/runtime/timer.rs @@ -0,0 +1,41 @@ +pub(crate) use self::variant::*; + +#[cfg(feature = "timer")] +mod variant { + use crate::runtime::io; + use crate::timer::{clock, timer}; + + pub(crate) type Clock = clock::Clock; + pub(crate) type Driver = timer::Timer; + pub(crate) type Handle = timer::Handle; + + /// Create a new timer driver / handle pair + pub(crate) fn create(io_driver: io::Driver, clock: Clock) -> (Driver, Handle) { + let driver = timer::Timer::new_with_clock(io_driver, clock); + let handle = driver.handle(); + + (driver, handle) + } + + #[cfg(feature = "blocking")] + pub(crate) fn set_default(handle: &Handle) -> timer::DefaultGuard<'_> { + timer::set_default(handle) + } +} + +#[cfg(not(feature = "timer"))] +mod variant { + use crate::runtime::io; + + pub(crate) type Clock = (); + pub(crate) type Driver = io::Driver; + pub(crate) type Handle = (); + + /// Create a new timer driver / handle pair + pub(crate) fn create(io_driver: io::Driver, _clock: Clock) -> (Driver, Handle) { + (io_driver, ()) + } + + #[cfg(feature = "blocking")] + pub(crate) fn set_default(_handle: &Handle) {} +} diff --git a/tokio/src/timer/timer/mod.rs b/tokio/src/timer/timer/mod.rs index c42b629b9..cce30b668 100644 --- a/tokio/src/timer/timer/mod.rs +++ b/tokio/src/timer/timer/mod.rs @@ -33,7 +33,7 @@ use self::entry::Entry; mod handle; pub(crate) use self::handle::HandlePriv; -pub use self::handle::{set_default, Handle}; +pub use self::handle::{set_default, DefaultGuard, Handle}; mod registration; pub(crate) use self::registration::Registration; @@ -41,7 +41,7 @@ pub(crate) use self::registration::Registration; mod stack; use self::stack::Stack; -use crate::executor::park::{Park, Unpark}; +use crate::runtime::{Park, Unpark}; use crate::timer::atomic::AtomicU64; use crate::timer::clock::Clock; use crate::timer::wheel; diff --git a/tokio/src/executor/util/mod.rs b/tokio/src/util/mod.rs similarity index 100% rename from tokio/src/executor/util/mod.rs rename to tokio/src/util/mod.rs index f16514439..44377fcf9 100644 --- a/tokio/src/executor/util/mod.rs +++ b/tokio/src/util/mod.rs @@ -1,5 +1,5 @@ mod pad; -mod rand; - pub(crate) use self::pad::CachePadded; + +mod rand; pub(crate) use self::rand::FastRand; diff --git a/tokio/src/executor/util/pad.rs b/tokio/src/util/pad.rs similarity index 100% rename from tokio/src/executor/util/pad.rs rename to tokio/src/util/pad.rs diff --git a/tokio/src/executor/util/rand.rs b/tokio/src/util/rand.rs similarity index 100% rename from tokio/src/executor/util/rand.rs rename to tokio/src/util/rand.rs diff --git a/tokio/tests/clock.rs b/tokio/tests/clock.rs index 29035bfb4..a9c2e3d29 100644 --- a/tokio/tests/clock.rs +++ b/tokio/tests/clock.rs @@ -20,7 +20,11 @@ fn clock_and_timer_concurrent() { let when = Instant::now() + Duration::from_millis(5_000); let clock = Clock::new_with_now(MockNow(when)); - let mut rt = runtime::Builder::new().clock(clock).build().unwrap(); + let mut rt = runtime::Builder::new() + .thread_pool() + .clock(clock) + .build() + .unwrap(); let (tx, rx) = mpsc::channel(); diff --git a/tokio/tests/rt_thread_pool.rs b/tokio/tests/rt_thread_pool.rs index 432497870..fcf5ef850 100644 --- a/tokio/tests/rt_thread_pool.rs +++ b/tokio/tests/rt_thread_pool.rs @@ -16,7 +16,7 @@ use std::task::{Context, Poll}; #[test] fn single_thread() { // No panic when starting a runtime w/ a single thread - let _ = runtime::Builder::new().num_threads(1).build(); + let _ = runtime::Builder::new().thread_pool().num_threads(1).build(); } #[test] @@ -185,6 +185,7 @@ fn drop_threadpool_drops_futures() { let b = num_dec.clone(); let rt = runtime::Builder::new() + .thread_pool() .after_start(move || { a.fetch_add(1, Relaxed); }) @@ -223,6 +224,7 @@ fn after_start_and_before_stop_is_called() { let after_inner = after_start.clone(); let before_inner = before_stop.clone(); let mut rt = tokio::runtime::Builder::new() + .thread_pool() .after_start(move || { after_inner.clone().fetch_add(1, Ordering::Relaxed); }) @@ -263,7 +265,7 @@ fn blocking() { for _ in 0..4 { let block = block.clone(); rt.spawn(async move { - tokio::executor::blocking::in_place(move || { + tokio::runtime::blocking::in_place(move || { block.wait(); block.wait(); })