From 0d38936b35779b604770120da2e98560bbb6241f Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 18 Nov 2019 07:00:55 -0800 Subject: [PATCH] chore: refine feature flags (#1785) Removes dependencies between Tokio feature flags. For example, `process` should not depend on `sync` simply because it uses the `mpsc` channel. Instead, feature flags represent **public** APIs that become available with the feature enabled. When the feature is not enabled, the functionality is removed. If another Tokio component requires the functionality, it is stays as `pub(crate)`. The threaded scheduler is now exposed under `rt-threaded`. This feature flag only enables the threaded scheduler and does not include I/O, networking, or time. Those features must be explictly enabled. A `full` feature flag is added that enables all features. `stdin`, `stdout`, `stderr` are exposed under `io-std`. Macros are used to scope code by feature flag. --- tokio-macros/Cargo.toml | 2 +- tokio/Cargo.toml | 33 ++- tokio/src/blocking/mod.rs | 12 +- tokio/src/blocking/pool.rs | 22 +- tokio/src/io/blocking.rs | 30 +- tokio/src/io/mod.rs | 79 ++--- tokio/src/io/util/mod.rs | 110 +++---- tokio/src/lib.rs | 81 +++--- tokio/src/loom/std/mod.rs | 8 +- tokio/src/macros/assert.rs | 19 ++ tokio/src/macros/cfg.rs | 217 ++++++++++++++ tokio/src/macros/loom.rs | 12 + tokio/src/macros/mod.rs | 17 ++ tokio/src/macros/ready.rs | 8 + tokio/src/macros/thread_local.rs | 4 + tokio/src/net/addr.rs | 270 +++++++++--------- tokio/src/net/mod.rs | 34 ++- tokio/src/prelude.rs | 8 +- tokio/src/process/mod.rs | 4 +- tokio/src/runtime/blocking.rs | 8 +- tokio/src/runtime/builder.rs | 218 +++++++------- tokio/src/runtime/enter.rs | 96 ++++--- tokio/src/runtime/global.rs | 22 +- tokio/src/runtime/handle.rs | 117 ++++---- tokio/src/runtime/io.rs | 8 +- tokio/src/runtime/mod.rs | 39 +-- tokio/src/runtime/park/mod.rs | 2 +- tokio/src/runtime/park/thread.rs | 4 +- tokio/src/runtime/thread_pool/mod.rs | 4 +- tokio/src/runtime/thread_pool/worker.rs | 34 +-- tokio/src/runtime/time.rs | 8 +- tokio/src/sync/mod.rs | 67 +++-- tokio/src/sync/mpsc/bounded.rs | 11 +- tokio/src/sync/mpsc/chan.rs | 6 - tokio/src/sync/mpsc/list.rs | 7 - tokio/src/sync/mpsc/mod.rs | 2 + tokio/src/sync/oneshot.rs | 2 + tokio/src/sync/semaphore.rs | 108 ++----- tokio/src/sync/task/atomic_waker.rs | 8 - tokio/src/sync/tests/loom_list.rs | 4 - tokio/src/sync/tests/mod.rs | 27 +- .../sync/tests/semaphore.rs} | 4 +- tokio/src/task/blocking.rs | 113 ++++---- tokio/src/task/mod.rs | 52 ++-- tokio/src/task/raw.rs | 19 +- tokio/src/task/state.rs | 2 +- tokio/src/tests/mod.rs | 22 -- tokio/src/time/clock.rs | 13 +- tokio/tests/fs_file_mocked.rs | 10 + 49 files changed, 1092 insertions(+), 915 deletions(-) create mode 100644 tokio/src/macros/assert.rs create mode 100644 tokio/src/macros/cfg.rs create mode 100644 tokio/src/macros/loom.rs create mode 100644 tokio/src/macros/mod.rs create mode 100644 tokio/src/macros/ready.rs create mode 100644 tokio/src/macros/thread_local.rs rename tokio/{tests/sync_semaphore.rs => src/sync/tests/semaphore.rs} (97%) diff --git a/tokio-macros/Cargo.toml b/tokio-macros/Cargo.toml index 21fe9117a..4bf87927e 100644 --- a/tokio-macros/Cargo.toml +++ b/tokio-macros/Cargo.toml @@ -29,7 +29,7 @@ quote = "1" syn = { version = "1.0.3", features = ["full"] } [dev-dependencies] -tokio = { version = "=0.2.0-alpha.6", path = "../tokio", default-features = false, features = ["rt-full"] } +tokio = { version = "=0.2.0-alpha.6", path = "../tokio" } [package.metadata.docs.rs] all-features = true diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 1c2b24f16..9364ea827 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -24,14 +24,21 @@ categories = ["asynchronous", "network-programming"] keywords = ["io", "async", "non-blocking", "futures"] [features] -default = [ +default = ["full"] + +# enable everything +full = [ "blocking", + "dns", "fs", + "io-driver", "io-util", - "io", + "io-std", + "macros", "net", "process", - "rt-full", + "rt-core", + "rt-threaded", "signal", "stream", "sync", @@ -39,15 +46,16 @@ default = [ ] blocking = ["rt-core"] -dns = ["blocking"] -fs = ["blocking"] -io-driver = ["mio", "lazy_static", "sync"] # TODO: get rid of sync +dns = ["rt-core"] +fs = ["rt-core"] +io-driver = ["rt-core", "mio", "lazy_static"] io-util = ["memchr"] -io = ["io-util", "blocking"] +# stdin, stdout, stderr +io-std = ["rt-core"] macros = ["tokio-macros"] net = ["dns", "tcp", "udp", "uds"] process = [ - "io-util", # TODO: Get rid of + "io-driver", "libc", "mio-named-pipes", "signal", @@ -58,14 +66,9 @@ process = [ ] # Includes basic task execution capabilities rt-core = [] -# TODO: rename this -> `rt-threaded` -rt-full = [ - "macros", +rt-threaded = [ "num_cpus", - "net", "rt-core", - "sync", - "time", ] signal = [ "io-driver", @@ -80,7 +83,7 @@ stream = ["futures-core"] sync = ["fnv"] test-util = [] tcp = ["io-driver"] -time = ["rt-core", "sync", "slab"] +time = ["rt-core", "slab"] udp = ["io-driver"] uds = ["io-driver", "mio-uds", "libc"] diff --git a/tokio/src/blocking/mod.rs b/tokio/src/blocking/mod.rs index ae830b49a..500055ae9 100644 --- a/tokio/src/blocking/mod.rs +++ b/tokio/src/blocking/mod.rs @@ -1,7 +1,11 @@ +#![cfg_attr(not(feature = "blocking"), allow(dead_code, unused_imports))] + //! Perform blocking operations from an asynchronous context. -mod pool; -pub(crate) use self::pool::{spawn_blocking, BlockingPool, Spawner}; +cfg_blocking_impl! { + mod pool; + pub(crate) use pool::{spawn_blocking, BlockingPool, Spawner}; -mod schedule; -mod task; + mod schedule; + mod task; +} diff --git a/tokio/src/blocking/pool.rs b/tokio/src/blocking/pool.rs index fccfc18e5..f75af7802 100644 --- a/tokio/src/blocking/pool.rs +++ b/tokio/src/blocking/pool.rs @@ -116,16 +116,20 @@ impl fmt::Debug for BlockingPool { // ===== impl Spawner ===== -impl Spawner { - #[cfg(feature = "rt-full")] - pub(crate) fn spawn_background(&self, func: F) - where - F: FnOnce() + Send + 'static, - { - let task = task::background(BlockingTask::new(func)); - self.schedule(task); - } +cfg_rt_threaded! { + impl Spawner { + pub(crate) fn spawn_background(&self, func: F) + where + F: FnOnce() + Send + 'static, + { + let task = task::background(BlockingTask::new(func)); + self.schedule(task); + } + } +} + +impl Spawner { /// Set the blocking pool for the duration of the closure /// /// If a blocking pool is already set, it will be restored when the closure diff --git a/tokio/src/io/blocking.rs b/tokio/src/io/blocking.rs index 79e2e0b73..87b251b76 100644 --- a/tokio/src/io/blocking.rs +++ b/tokio/src/io/blocking.rs @@ -34,13 +34,14 @@ enum State { Busy(sys::Blocking<(io::Result, Buf, T)>), } -impl Blocking { - #[cfg(feature = "io")] - pub(crate) fn new(inner: T) -> Blocking { - Blocking { - inner: Some(inner), - state: State::Idle(Some(Buf::with_capacity(0))), - need_flush: false, +cfg_io_std! { + impl Blocking { + pub(crate) fn new(inner: T) -> Blocking { + Blocking { + inner: Some(inner), + state: State::Idle(Some(Buf::with_capacity(0))), + need_flush: false, + } } } } @@ -264,12 +265,15 @@ impl Buf { self.buf.clear(); res } +} - #[cfg(feature = "fs")] - pub(crate) fn discard_read(&mut self) -> i64 { - let ret = -(self.bytes().len() as i64); - self.pos = 0; - self.buf.truncate(0); - ret +cfg_fs! { + impl Buf { + pub(crate) fn discard_read(&mut self) -> i64 { + let ret = -(self.bytes().len() as i64); + self.pos = 0; + self.buf.truncate(0); + ret + } } } diff --git a/tokio/src/io/mod.rs b/tokio/src/io/mod.rs index df84efd7d..df1888cea 100644 --- a/tokio/src/io/mod.rs +++ b/tokio/src/io/mod.rs @@ -36,8 +36,9 @@ //! [`ErrorKind`]: enum.ErrorKind.html //! [`Result`]: type.Result.html -#[cfg(any(feature = "io", feature = "fs"))] -pub(crate) mod blocking; +cfg_io_blocking! { + pub(crate) mod blocking; +} mod async_buf_read; pub use self::async_buf_read::AsyncBufRead; @@ -48,43 +49,43 @@ pub use self::async_read::AsyncRead; mod async_write; pub use self::async_write::AsyncWrite; -#[cfg(feature = "io-util")] -pub mod split; -#[cfg(feature = "io-util")] -pub use self::split::split; +cfg_io_std! { + mod stderr; + pub use stderr::{stderr, Stderr}; -#[cfg(feature = "io-util")] -mod util; -#[cfg(feature = "io-util")] -pub use self::util::{ - copy, empty, repeat, sink, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufStream, - BufWriter, Copy, Empty, Lines, Repeat, Sink, Split, Take, -}; + mod stdin; + pub use stdin::{stdin, Stdin}; -#[cfg(feature = "io")] -mod stderr; -#[cfg(feature = "io")] -pub use self::stderr::{stderr, Stderr}; - -#[cfg(feature = "io")] -mod stdin; -#[cfg(feature = "io")] -pub use self::stdin::{stdin, Stdin}; - -#[cfg(feature = "io")] -mod stdout; -#[cfg(feature = "io")] -pub use self::stdout::{stdout, Stdout}; - -// Re-export io::Error so that users don't have to deal -// with conflicts when `use`ing `tokio::io` and `std::io`. -#[cfg(feature = "io-util")] -pub use std::io::{Error, ErrorKind, Result}; - -/// Types in this module can be mocked out in tests. -#[cfg(any(feature = "io", feature = "fs"))] -mod sys { - // TODO: don't rename - pub(crate) use crate::blocking::spawn_blocking as run; - pub(crate) use crate::task::JoinHandle as Blocking; + mod stdout; + pub use stdout::{stdout, Stdout}; +} + +cfg_io_util! { + pub mod split; + pub use split::split; + + pub(crate) mod util; + pub use util::{ + copy, empty, repeat, sink, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufStream, + BufWriter, Copy, Empty, Lines, Repeat, Sink, Split, Take, + }; + + // Re-export io::Error so that users don't have to deal with conflicts when + // `use`ing `tokio::io` and `std::io`. + pub use std::io::{Error, ErrorKind, Result}; +} + +cfg_not_io_util! { + cfg_process! { + pub(crate) mod util; + } +} + +cfg_io_blocking! { + /// Types in this module can be mocked out in tests. + mod sys { + // TODO: don't rename + pub(crate) use crate::blocking::spawn_blocking as run; + pub(crate) use crate::task::JoinHandle as Blocking; + } } diff --git a/tokio/src/io/util/mod.rs b/tokio/src/io/util/mod.rs index 8d4e7678f..24451ef6a 100644 --- a/tokio/src/io/util/mod.rs +++ b/tokio/src/io/util/mod.rs @@ -1,71 +1,75 @@ -mod async_buf_read_ext; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::async_buf_read_ext::AsyncBufReadExt; +#![allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -mod async_read_ext; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::async_read_ext::AsyncReadExt; +cfg_io_util! { + mod async_buf_read_ext; + pub use async_buf_read_ext::AsyncBufReadExt; -mod async_write_ext; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::async_write_ext::AsyncWriteExt; + mod async_read_ext; + pub use async_read_ext::AsyncReadExt; -mod buf_reader; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::buf_reader::BufReader; + mod async_write_ext; + pub use async_write_ext::AsyncWriteExt; -mod buf_stream; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::buf_stream::BufStream; + mod buf_reader; + pub use buf_reader::BufReader; -mod buf_writer; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::buf_writer::BufWriter; + mod buf_stream; + pub use buf_stream::BufStream; -mod chain; + mod buf_writer; + pub use buf_writer::BufWriter; -mod copy; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::copy::{copy, Copy}; + mod chain; -mod empty; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::empty::{empty, Empty}; + mod copy; + pub use copy::{copy, Copy}; -mod flush; + mod empty; + pub use empty::{empty, Empty}; -mod lines; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::lines::Lines; + mod flush; -mod read; -mod read_exact; -mod read_line; -mod read_to_end; -mod read_to_string; -mod read_until; + mod lines; + pub use lines::Lines; -mod repeat; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::repeat::{repeat, Repeat}; + mod read; + mod read_exact; + mod read_line; -mod shutdown; + mod read_to_end; + cfg_process! { + pub(crate) use read_to_end::read_to_end; + } -mod sink; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::sink::{sink, Sink}; + mod read_to_string; + mod read_until; -mod split; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::split::Split; + mod repeat; + pub use repeat::{repeat, Repeat}; -mod take; -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::take::Take; + mod shutdown; -mod write; -mod write_all; + mod sink; + pub use sink::{sink, Sink}; -// used by `BufReader` and `BufWriter` -// https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1 -const DEFAULT_BUF_SIZE: usize = 8 * 1024; + mod split; + pub use split::Split; + + mod take; + pub use take::Take; + + mod write; + mod write_all; + + // used by `BufReader` and `BufWriter` + // https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1 + const DEFAULT_BUF_SIZE: usize = 8 * 1024; +} + +cfg_not_io_util! { + cfg_process! { + mod read_to_end; + // Used by process + pub(crate) use read_to_end::read_to_end; + } +} diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index 81d93c68b..d82d58d8a 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -70,75 +70,68 @@ //! } //! ``` -#[cfg(all(loom, test))] -macro_rules! thread_local { - ($($tts:tt)+) => { loom::thread_local!{ $($tts)+ } } -} - -macro_rules! ready { - ($e:expr $(,)?) => { - match $e { - std::task::Poll::Ready(t) => t, - std::task::Poll::Pending => return std::task::Poll::Pending, - } - }; -} - -// At the top due to macros -#[cfg(test)] +// macros used internally #[macro_use] -mod tests; +mod macros; -#[cfg(feature = "blocking")] +// Blocking task implementation pub(crate) mod blocking; -#[cfg(feature = "fs")] -pub mod fs; +cfg_fs! { + pub mod fs; +} mod future; pub mod io; -#[cfg(feature = "io-driver")] pub mod net; mod loom; pub mod prelude; -#[cfg(feature = "process")] -#[cfg(not(loom))] -pub mod process; +cfg_process! { + pub mod process; +} pub mod runtime; -#[cfg(feature = "signal")] -#[cfg(not(loom))] -pub mod signal; +cfg_signal! { + pub mod signal; +} -#[cfg(feature = "sync")] -pub mod sync; +cfg_sync! { + pub mod sync; +} +cfg_not_sync! { + mod sync; +} -#[cfg(feature = "rt-core")] -pub mod task; -#[cfg(feature = "rt-core")] -pub use crate::task::spawn; +cfg_rt_core! { + pub mod task; + pub use task::spawn; +} -#[cfg(feature = "time")] -pub mod time; +cfg_time! { + pub mod time; +} -#[cfg(feature = "rt-full")] -mod util; +cfg_rt_threaded! { + mod util; +} -#[cfg(not(test))] // Work around for rust-lang/rust#62127 -#[cfg(feature = "macros")] -#[doc(inline)] -pub use tokio_macros::main; +cfg_macros! { + #[cfg(not(test))] // Work around for rust-lang/rust#62127 + pub use tokio_macros::main; + pub use tokio_macros::test; +} -#[cfg(feature = "macros")] -#[doc(inline)] -pub use tokio_macros::test; +// Tests +#[cfg(test)] +mod tests; +// TODO: rm #[cfg(feature = "io-util")] #[cfg(test)] fn is_unpin() {} diff --git a/tokio/src/loom/std/mod.rs b/tokio/src/loom/std/mod.rs index c5bd6039f..2c5b7eaa3 100644 --- a/tokio/src/loom/std/mod.rs +++ b/tokio/src/loom/std/mod.rs @@ -1,4 +1,4 @@ -#![cfg_attr(not(feature = "rt-full"), allow(unused_imports, dead_code))] +#![cfg_attr(not(feature = "full"), allow(unused_imports, dead_code))] mod atomic_u32; mod atomic_u64; @@ -11,7 +11,7 @@ pub(crate) mod cell { pub(crate) use super::causal_cell::{CausalCell, CausalCheck}; } -#[cfg(feature = "sync")] +#[cfg(any(feature = "sync", feature = "io-driver"))] pub(crate) mod future { pub(crate) use crate::sync::AtomicWaker; } @@ -51,12 +51,12 @@ pub(crate) mod sync { } pub(crate) mod sys { - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] pub(crate) fn num_cpus() -> usize { usize::max(1, num_cpus::get_physical()) } - #[cfg(not(feature = "rt-full"))] + #[cfg(not(feature = "rt-threaded"))] pub(crate) fn num_cpus() -> usize { 1 } diff --git a/tokio/src/macros/assert.rs b/tokio/src/macros/assert.rs new file mode 100644 index 000000000..fd6601b44 --- /dev/null +++ b/tokio/src/macros/assert.rs @@ -0,0 +1,19 @@ +/// Assert option is some +macro_rules! assert_some { + ($e:expr) => {{ + match $e { + Some(v) => v, + _ => panic!("expected some, was none"), + } + }}; +} + +/// Assert option is none +macro_rules! assert_none { + ($e:expr) => {{ + match $e { + Some(v) => panic!("expected none, was {:?}", v), + _ => {} + } + }}; +} diff --git a/tokio/src/macros/cfg.rs b/tokio/src/macros/cfg.rs new file mode 100644 index 000000000..5e84a3ac1 --- /dev/null +++ b/tokio/src/macros/cfg.rs @@ -0,0 +1,217 @@ +#![allow(unused_macros)] + +macro_rules! cfg_atomic_waker { + ($($item:item)*) => { + $( #[cfg(any(feature = "io-driver", feature = "time"))] $item )* + } +} + +macro_rules! cfg_blocking { + ($($item:item)*) => { + $( #[cfg(feature = "blocking")] $item )* + } +} + +/// Enable blocking API internals +macro_rules! cfg_blocking_impl { + ($($item:item)*) => { + $( + #[cfg(any( + feature = "blocking", + feature = "fs", + feature = "dns", + feature = "io-std", + feature = "rt-threaded", + ))] + $item + )* + } +} + +/// Enable blocking API internals +macro_rules! cfg_not_blocking_impl { + ($($item:item)*) => { + $( + #[cfg(not(any( + feature = "blocking", + feature = "fs", + feature = "dns", + feature = "io-std", + feature = "rt-threaded", + )))] + $item + )* + } +} + +macro_rules! cfg_dns { + ($($item:item)*) => { + $( #[cfg(feature = "dns")] $item )* + } +} + +macro_rules! cfg_fs { + ($($item:item)*) => { $( #[cfg(feature = "fs")] $item )* } +} + +macro_rules! cfg_io_blocking { + ($($item:item)*) => { + $( #[cfg(any(feature = "io-std", feature = "fs"))] $item )* + } +} + +macro_rules! cfg_io_driver { + ($($item:item)*) => { + $( #[cfg(feature = "io-driver")] $item )* + } +} + +macro_rules! cfg_not_io_driver { + ($($item:item)*) => { + $( #[cfg(not(feature = "io-driver"))] $item )* + } +} + +macro_rules! cfg_io_std { + ($($item:item)*) => { + $( #[cfg(feature = "io-std")] $item )* + } +} + +macro_rules! cfg_io_util { + ($($item:item)*) => { + $( #[cfg(feature = "io-util")] $item )* + } +} + +macro_rules! cfg_not_io_util { + ($($item:item)*) => { + $( #[cfg(not(feature = "io-util"))] $item )* + } +} + +macro_rules! cfg_loom { + ($($item:item)*) => { + $( #[cfg(loom)] $item )* + } +} + +macro_rules! cfg_not_loom { + ($($item:item)*) => { + $( #[cfg(not(loom))] $item )* + } +} + +macro_rules! cfg_macros { + ($($item:item)*) => { + $( + #[cfg(feature = "macros")] + #[doc(inline)] + $item + )* + } +} + +macro_rules! cfg_process { + ($($item:item)*) => { + $( + #[cfg(feature = "process")] + #[cfg(not(loom))] + $item + )* + } +} + +macro_rules! cfg_signal { + ($($item:item)*) => { + $( + #[cfg(feature = "signal")] + #[cfg(not(loom))] + $item + )* + } +} + +macro_rules! cfg_stream { + ($($item:item)*) => { + $( #[cfg(feature = "stream")] $item )* + } +} + +macro_rules! cfg_sync { + ($($item:item)*) => { + $( #[cfg(feature = "sync")] $item )* + } +} + +macro_rules! cfg_not_sync { + ($($item:item)*) => { + $( #[cfg(not(feature = "sync"))] $item )* + } +} + +macro_rules! cfg_rt_core { + ($($item:item)*) => { + $( #[cfg(feature = "rt-core")] $item )* + } +} + +macro_rules! cfg_not_rt_core { + ($($item:item)*) => { + $( #[cfg(not(feature = "rt-core"))] $item )* + } +} + +macro_rules! cfg_rt_threaded { + ($($item:item)*) => { + $( #[cfg(feature = "rt-threaded")] $item )* + } +} + +macro_rules! cfg_not_rt_threaded { + ($($item:item)*) => { + $( #[cfg(not(feature = "rt-threaded"))] $item )* + } +} + +macro_rules! cfg_tcp { + ($($item:item)*) => { + $( #[cfg(feature = "tcp")] $item )* + } +} + +macro_rules! cfg_test_util { + ($($item:item)*) => { + $( #[cfg(feature = "test-util")] $item )* + } +} + +macro_rules! cfg_not_test_util { + ($($item:item)*) => { + $( #[cfg(not(feature = "test-util"))] $item )* + } +} + +macro_rules! cfg_time { + ($($item:item)*) => { + $( #[cfg(feature = "time")] $item )* + } +} + +macro_rules! cfg_not_time { + ($($item:item)*) => { + $( #[cfg(not(feature = "time"))] $item )* + } +} + +macro_rules! cfg_udp { + ($($item:item)*) => { + $( #[cfg(feature = "udp")] $item )* + } +} + +macro_rules! cfg_uds { + ($($item:item)*) => { + $( #[cfg(all(unix, feature = "uds"))] $item )* + } +} diff --git a/tokio/src/macros/loom.rs b/tokio/src/macros/loom.rs new file mode 100644 index 000000000..d57d9fb0f --- /dev/null +++ b/tokio/src/macros/loom.rs @@ -0,0 +1,12 @@ +macro_rules! if_loom { + ($($t:tt)*) => {{ + #[cfg(loom)] + const LOOM: bool = true; + #[cfg(not(loom))] + const LOOM: bool = false; + + if LOOM { + $($t)* + } + }} +} diff --git a/tokio/src/macros/mod.rs b/tokio/src/macros/mod.rs new file mode 100644 index 000000000..9136e5944 --- /dev/null +++ b/tokio/src/macros/mod.rs @@ -0,0 +1,17 @@ +#![cfg_attr(not(feature = "full"), allow(unused_macros))] + +#[macro_use] +#[cfg(test)] +mod assert; + +#[macro_use] +mod cfg; + +#[macro_use] +mod loom; + +#[macro_use] +mod ready; + +#[macro_use] +mod thread_local; diff --git a/tokio/src/macros/ready.rs b/tokio/src/macros/ready.rs new file mode 100644 index 000000000..1f48623b8 --- /dev/null +++ b/tokio/src/macros/ready.rs @@ -0,0 +1,8 @@ +macro_rules! ready { + ($e:expr $(,)?) => { + match $e { + std::task::Poll::Ready(t) => t, + std::task::Poll::Pending => return std::task::Poll::Pending, + } + }; +} diff --git a/tokio/src/macros/thread_local.rs b/tokio/src/macros/thread_local.rs new file mode 100644 index 000000000..d84894735 --- /dev/null +++ b/tokio/src/macros/thread_local.rs @@ -0,0 +1,4 @@ +#[cfg(all(loom, test))] +macro_rules! thread_local { + ($($tts:tt)+) => { loom::thread_local!{ $($tts)+ } } +} diff --git a/tokio/src/net/addr.rs b/tokio/src/net/addr.rs index 7c30a7e4a..108557e4d 100644 --- a/tokio/src/net/addr.rs +++ b/tokio/src/net/addr.rs @@ -16,6 +16,22 @@ pub trait ToSocketAddrs: sealed::ToSocketAddrsPriv {} type ReadyFuture = future::Ready>; +// ===== impl &impl ToSocketAddrs ===== + +impl ToSocketAddrs for &T {} + +impl sealed::ToSocketAddrsPriv for &T +where + T: sealed::ToSocketAddrsPriv + ?Sized, +{ + type Iter = T::Iter; + type Future = T::Future; + + fn to_socket_addrs(&self) -> Self::Future { + (**self).to_socket_addrs() + } +} + // ===== impl SocketAddr ===== impl ToSocketAddrs for SocketAddr {} @@ -56,75 +72,6 @@ impl sealed::ToSocketAddrsPriv for SocketAddrV6 { } } -// ===== impl str ===== - -#[cfg(feature = "dns")] -impl ToSocketAddrs for str {} - -#[cfg(feature = "dns")] -impl sealed::ToSocketAddrsPriv for str { - type Iter = sealed::OneOrMore; - type Future = sealed::MaybeReady; - - fn to_socket_addrs(&self) -> Self::Future { - use crate::blocking; - use sealed::MaybeReady; - - // First check if the input parses as a socket address - let res: Result = self.parse(); - - if let Ok(addr) = res { - return MaybeReady::Ready(Some(addr)); - } - - // Run DNS lookup on the blocking pool - let s = self.to_owned(); - - MaybeReady::Blocking(blocking::spawn_blocking(move || { - std::net::ToSocketAddrs::to_socket_addrs(&s) - })) - } -} - -// ===== impl (&str, u16) ===== - -#[cfg(feature = "dns")] -impl ToSocketAddrs for (&str, u16) {} - -#[cfg(feature = "dns")] -impl sealed::ToSocketAddrsPriv for (&str, u16) { - type Iter = sealed::OneOrMore; - type Future = sealed::MaybeReady; - - fn to_socket_addrs(&self) -> Self::Future { - use crate::blocking; - use sealed::MaybeReady; - - let (host, port) = *self; - - // try to parse the host as a regular IP address first - if let Ok(addr) = host.parse::() { - let addr = SocketAddrV4::new(addr, port); - let addr = SocketAddr::V4(addr); - - return MaybeReady::Ready(Some(addr)); - } - - if let Ok(addr) = host.parse::() { - let addr = SocketAddrV6::new(addr, port, 0, 0); - let addr = SocketAddr::V6(addr); - - return MaybeReady::Ready(Some(addr)); - } - - let host = host.to_owned(); - - MaybeReady::Blocking(blocking::spawn_blocking(move || { - std::net::ToSocketAddrs::to_socket_addrs(&(&host[..], port)) - })) - } -} - // ===== impl (IpAddr, u16) ===== impl ToSocketAddrs for (IpAddr, u16) {} @@ -167,34 +114,83 @@ impl sealed::ToSocketAddrsPriv for (Ipv6Addr, u16) { } } -// ===== impl String ===== +cfg_dns! { + // ===== impl str ===== -#[cfg(feature = "dns")] -impl ToSocketAddrs for String {} + impl ToSocketAddrs for str {} -#[cfg(feature = "dns")] -impl sealed::ToSocketAddrsPriv for String { - type Iter = ::Iter; - type Future = ::Future; + impl sealed::ToSocketAddrsPriv for str { + type Iter = sealed::OneOrMore; + type Future = sealed::MaybeReady; - fn to_socket_addrs(&self) -> Self::Future { - (&self[..]).to_socket_addrs() + fn to_socket_addrs(&self) -> Self::Future { + use crate::blocking; + use sealed::MaybeReady; + + // First check if the input parses as a socket address + let res: Result = self.parse(); + + if let Ok(addr) = res { + return MaybeReady::Ready(Some(addr)); + } + + // Run DNS lookup on the blocking pool + let s = self.to_owned(); + + MaybeReady::Blocking(blocking::spawn_blocking(move || { + std::net::ToSocketAddrs::to_socket_addrs(&s) + })) + } } -} -// ===== impl &impl ToSocketAddrs ===== + // ===== impl (&str, u16) ===== -impl ToSocketAddrs for &T {} + impl ToSocketAddrs for (&str, u16) {} -impl sealed::ToSocketAddrsPriv for &T -where - T: sealed::ToSocketAddrsPriv + ?Sized, -{ - type Iter = T::Iter; - type Future = T::Future; + impl sealed::ToSocketAddrsPriv for (&str, u16) { + type Iter = sealed::OneOrMore; + type Future = sealed::MaybeReady; - fn to_socket_addrs(&self) -> Self::Future { - (**self).to_socket_addrs() + fn to_socket_addrs(&self) -> Self::Future { + use crate::blocking; + use sealed::MaybeReady; + + let (host, port) = *self; + + // try to parse the host as a regular IP address first + if let Ok(addr) = host.parse::() { + let addr = SocketAddrV4::new(addr, port); + let addr = SocketAddr::V4(addr); + + return MaybeReady::Ready(Some(addr)); + } + + if let Ok(addr) = host.parse::() { + let addr = SocketAddrV6::new(addr, port, 0, 0); + let addr = SocketAddr::V6(addr); + + return MaybeReady::Ready(Some(addr)); + } + + let host = host.to_owned(); + + MaybeReady::Blocking(blocking::spawn_blocking(move || { + std::net::ToSocketAddrs::to_socket_addrs(&(&host[..], port)) + })) + } + } + + // ===== impl String ===== + + impl ToSocketAddrs for String {} + + impl sealed::ToSocketAddrsPriv for String { + type Iter = ::Iter; + type Future = ::Future; + + fn to_socket_addrs(&self) -> Self::Future { + (&self[..]).to_socket_addrs() + } } } @@ -203,20 +199,18 @@ pub(crate) mod sealed { //! part of the `ToSocketAddrs` public API. The details will change over //! time. - #[cfg(feature = "dns")] - use crate::task::JoinHandle; - use std::future::Future; use std::io; use std::net::SocketAddr; - #[cfg(feature = "dns")] - use std::option; - #[cfg(feature = "dns")] - use std::pin::Pin; - #[cfg(feature = "dns")] - use std::task::{Context, Poll}; - #[cfg(feature = "dns")] - use std::vec; + + cfg_dns! { + use crate::task::JoinHandle; + + use std::option; + use std::pin::Pin; + use std::task::{Context, Poll}; + use std::vec; + } #[doc(hidden)] pub trait ToSocketAddrsPriv { @@ -226,56 +220,54 @@ pub(crate) mod sealed { fn to_socket_addrs(&self) -> Self::Future; } - #[doc(hidden)] - #[derive(Debug)] - #[cfg(feature = "dns")] - pub enum MaybeReady { - Ready(Option), - Blocking(JoinHandle>>), - } + cfg_dns! { + #[doc(hidden)] + #[derive(Debug)] + pub enum MaybeReady { + Ready(Option), + Blocking(JoinHandle>>), + } - #[doc(hidden)] - #[derive(Debug)] - #[cfg(feature = "dns")] - pub enum OneOrMore { - One(option::IntoIter), - More(vec::IntoIter), - } + #[doc(hidden)] + #[derive(Debug)] + pub enum OneOrMore { + One(option::IntoIter), + More(vec::IntoIter), + } - #[cfg(feature = "dns")] - impl Future for MaybeReady { - type Output = io::Result; + impl Future for MaybeReady { + type Output = io::Result; - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match *self { - MaybeReady::Ready(ref mut i) => { - let iter = OneOrMore::One(i.take().into_iter()); - Poll::Ready(Ok(iter)) - } - MaybeReady::Blocking(ref mut rx) => { - let res = ready!(Pin::new(rx).poll(cx))?.map(OneOrMore::More); + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match *self { + MaybeReady::Ready(ref mut i) => { + let iter = OneOrMore::One(i.take().into_iter()); + Poll::Ready(Ok(iter)) + } + MaybeReady::Blocking(ref mut rx) => { + let res = ready!(Pin::new(rx).poll(cx))?.map(OneOrMore::More); - Poll::Ready(res) + Poll::Ready(res) + } } } } - } - #[cfg(feature = "dns")] - impl Iterator for OneOrMore { - type Item = SocketAddr; + impl Iterator for OneOrMore { + type Item = SocketAddr; - fn next(&mut self) -> Option { - match self { - OneOrMore::One(i) => i.next(), - OneOrMore::More(i) => i.next(), + fn next(&mut self) -> Option { + match self { + OneOrMore::One(i) => i.next(), + OneOrMore::More(i) => i.next(), + } } - } - fn size_hint(&self) -> (usize, Option) { - match self { - OneOrMore::One(i) => i.size_hint(), - OneOrMore::More(i) => i.size_hint(), + fn size_hint(&self) -> (usize, Option) { + match self { + OneOrMore::One(i) => i.size_hint(), + OneOrMore::More(i) => i.size_hint(), + } } } } diff --git a/tokio/src/net/mod.rs b/tokio/src/net/mod.rs index 2ebf773fe..50021e37b 100644 --- a/tokio/src/net/mod.rs +++ b/tokio/src/net/mod.rs @@ -24,24 +24,22 @@ mod addr; pub use addr::ToSocketAddrs; -pub mod driver; +cfg_io_driver! { + pub mod driver; + pub mod util; +} -pub mod util; +cfg_tcp! { + pub mod tcp; + pub use tcp::{TcpListener, TcpStream}; +} -#[cfg(feature = "tcp")] -pub mod tcp; +cfg_udp! { + pub mod udp; + pub use udp::UdpSocket; +} -#[cfg(feature = "tcp")] -pub use self::tcp::{TcpListener, TcpStream}; - -#[cfg(feature = "udp")] -pub mod udp; - -#[cfg(feature = "udp")] -pub use self::udp::UdpSocket; - -#[cfg(all(unix, feature = "uds"))] -pub mod unix; - -#[cfg(all(unix, feature = "uds"))] -pub use self::unix::{UnixDatagram, UnixListener, UnixStream}; +cfg_uds! { + pub mod unix; + pub use unix::{UnixDatagram, UnixListener, UnixStream}; +} diff --git a/tokio/src/prelude.rs b/tokio/src/prelude.rs index 8675c5778..6e62c519d 100644 --- a/tokio/src/prelude.rs +++ b/tokio/src/prelude.rs @@ -12,6 +12,8 @@ //! The prelude may grow over time as additional items see ubiquitous use. pub use crate::io::{AsyncBufRead, AsyncRead, AsyncWrite}; -#[cfg(feature = "io-util")] -#[doc(no_inline)] -pub use crate::io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _}; + +cfg_io_util! { + #[doc(no_inline)] + pub use crate::io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _}; +} diff --git a/tokio/src/process/mod.rs b/tokio/src/process/mod.rs index 3013a65a7..ee43a9fb0 100644 --- a/tokio/src/process/mod.rs +++ b/tokio/src/process/mod.rs @@ -116,7 +116,7 @@ mod imp; mod kill; -use crate::io::{AsyncRead, AsyncReadExt, AsyncWrite}; +use crate::io::{AsyncRead, AsyncWrite}; use crate::process::kill::Kill; use std::ffi::OsStr; @@ -771,7 +771,7 @@ impl Child { async fn read_to_end(io: Option) -> io::Result> { let mut vec = Vec::new(); if let Some(mut io) = io { - AsyncReadExt::read_to_end(&mut io, &mut vec).await?; + crate::io::util::read_to_end(&mut io, &mut vec).await?; } Ok(vec) } diff --git a/tokio/src/runtime/blocking.rs b/tokio/src/runtime/blocking.rs index 39d103bd5..8408e78b7 100644 --- a/tokio/src/runtime/blocking.rs +++ b/tokio/src/runtime/blocking.rs @@ -3,10 +3,7 @@ //! shells. This isolates the complexity of dealing with conditional //! compilation. -pub(crate) use self::variant::*; - -#[cfg(feature = "blocking")] -mod variant { +cfg_blocking_impl! { pub(crate) use crate::blocking::BlockingPool; pub(crate) use crate::blocking::Spawner; @@ -17,8 +14,7 @@ mod variant { } } -#[cfg(not(feature = "blocking"))] -mod variant { +cfg_not_blocking_impl! { use crate::runtime::Builder; #[derive(Debug, Clone)] diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index 72cbc9c48..793146f70 100644 --- a/tokio/src/runtime/builder.rs +++ b/tokio/src/runtime/builder.rs @@ -62,7 +62,7 @@ enum Kind { Shell, #[cfg(feature = "rt-core")] Basic, - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] ThreadPool, } @@ -132,7 +132,7 @@ impl Builder { } /// Use a multi-threaded scheduler for executing tasks. - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] pub fn threaded_scheduler(&mut self) -> &mut Self { self.kind = Kind::ThreadPool; self @@ -255,7 +255,7 @@ impl Builder { Kind::Shell => self.build_shell_runtime(), #[cfg(feature = "rt-core")] Kind::Basic => self.build_basic_runtime(), - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] Kind::ThreadPool => self.build_threaded_runtime(), } } @@ -287,121 +287,127 @@ impl Builder { blocking_pool, }) } +} - #[cfg(feature = "rt-core")] - fn build_basic_runtime(&mut self) -> io::Result { - use crate::runtime::{BasicScheduler, Kind}; +cfg_rt_core! { + impl Builder { + fn build_basic_runtime(&mut self) -> io::Result { + use crate::runtime::{BasicScheduler, Kind}; - let clock = time::create_clock(); + let clock = time::create_clock(); - // Create I/O driver - let (io_driver, handle) = io::create_driver()?; - let io_handles = vec![handle]; - - let (driver, handle) = time::create_driver(io_driver, clock.clone()); - let time_handles = vec![handle]; - - // 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); - let spawner = scheduler.spawner(); - - // Blocking pool - let blocking_pool = blocking::create_blocking_pool(self); - let blocking_spawner = blocking_pool.spawner().clone(); - - Ok(Runtime { - kind: Kind::Basic(scheduler), - handle: Handle { - kind: handle::Kind::Basic(spawner), - io_handles, - time_handles, - clock, - blocking_spawner, - }, - blocking_pool, - }) - } - - #[cfg(feature = "rt-full")] - fn build_threaded_runtime(&mut self) -> io::Result { - use crate::runtime::{Kind, ThreadPool}; - use std::sync::Mutex; - - let clock = time::create_clock(); - - let mut io_handles = Vec::new(); - let mut time_handles = Vec::new(); - let mut drivers = Vec::new(); - - for _ in 0..self.num_threads { - // Create I/O driver and handle + // Create I/O driver let (io_driver, handle) = io::create_driver()?; - io_handles.push(handle); + let io_handles = vec![handle]; - // Create a new timer. - let (time_driver, handle) = time::create_driver(io_driver, clock.clone()); - time_handles.push(handle); - drivers.push(Mutex::new(Some(time_driver))); - } + let (driver, handle) = time::create_driver(io_driver, clock.clone()); + let time_handles = vec![handle]; - // Create the blocking pool - let blocking_pool = blocking::create_blocking_pool(self); - let blocking_spawner = blocking_pool.spawner().clone(); + // 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); + let spawner = scheduler.spawner(); - let scheduler = { - let clock = clock.clone(); - let io_handles = io_handles.clone(); - let time_handles = time_handles.clone(); + // Blocking pool + let blocking_pool = blocking::create_blocking_pool(self); + let blocking_spawner = blocking_pool.spawner().clone(); - let after_start = self.after_start.clone(); - let before_stop = self.before_stop.clone(); - - let around_worker = Arc::new(Box::new(move |index, next: &mut dyn FnMut()| { - // Configure the I/O driver - let _io = io::set_default(&io_handles[index]); - - // Configure time - time::with_default(&time_handles[index], &clock, || { - // Call the start callback - if let Some(after_start) = after_start.as_ref() { - after_start(); - } - - // Run the worker - next(); - - // Call the after call back - if let Some(before_stop) = before_stop.as_ref() { - before_stop(); - } - }) + Ok(Runtime { + kind: Kind::Basic(scheduler), + handle: Handle { + kind: handle::Kind::Basic(spawner), + io_handles, + time_handles, + clock, + blocking_spawner, + }, + blocking_pool, }) - as Box); + } + } +} - ThreadPool::new( - self.num_threads, - blocking_pool.spawner().clone(), - around_worker, - move |index| drivers[index].lock().unwrap().take().unwrap(), - ) - }; +cfg_rt_threaded! { + impl Builder { + fn build_threaded_runtime(&mut self) -> io::Result { + use crate::runtime::{Kind, ThreadPool}; + use std::sync::Mutex; - let spawner = scheduler.spawner().clone(); + let clock = time::create_clock(); - Ok(Runtime { - kind: Kind::ThreadPool(scheduler), - handle: Handle { - kind: handle::Kind::ThreadPool(spawner), - io_handles, - time_handles, - clock, - blocking_spawner, - }, - blocking_pool, - }) + let mut io_handles = Vec::new(); + let mut time_handles = Vec::new(); + let mut drivers = Vec::new(); + + for _ in 0..self.num_threads { + // Create I/O driver and handle + let (io_driver, handle) = io::create_driver()?; + io_handles.push(handle); + + // Create a new timer. + let (time_driver, handle) = time::create_driver(io_driver, clock.clone()); + time_handles.push(handle); + drivers.push(Mutex::new(Some(time_driver))); + } + + // Create the blocking pool + let blocking_pool = blocking::create_blocking_pool(self); + let blocking_spawner = blocking_pool.spawner().clone(); + + let scheduler = { + let clock = clock.clone(); + let io_handles = io_handles.clone(); + let time_handles = time_handles.clone(); + + let after_start = self.after_start.clone(); + let before_stop = self.before_stop.clone(); + + let around_worker = Arc::new(Box::new(move |index, next: &mut dyn FnMut()| { + // Configure the I/O driver + let _io = io::set_default(&io_handles[index]); + + // Configure time + time::with_default(&time_handles[index], &clock, || { + // Call the start callback + if let Some(after_start) = after_start.as_ref() { + after_start(); + } + + // Run the worker + next(); + + // Call the after call back + if let Some(before_stop) = before_stop.as_ref() { + before_stop(); + } + }) + }) + as Box); + + ThreadPool::new( + self.num_threads, + blocking_pool.spawner().clone(), + around_worker, + move |index| drivers[index].lock().unwrap().take().unwrap(), + ) + }; + + let spawner = scheduler.spawner().clone(); + + Ok(Runtime { + kind: Kind::ThreadPool(scheduler), + handle: Handle { + kind: handle::Kind::ThreadPool(spawner), + io_handles, + time_handles, + clock, + blocking_spawner, + }, + blocking_pool, + }) + } } } diff --git a/tokio/src/runtime/enter.rs b/tokio/src/runtime/enter.rs index 9caea0cf9..a0925cb49 100644 --- a/tokio/src/runtime/enter.rs +++ b/tokio/src/runtime/enter.rs @@ -1,7 +1,5 @@ use std::cell::{Cell, RefCell}; use std::fmt; -#[cfg(feature = "rt-full")] -use std::future::Future; use std::marker::PhantomData; thread_local!(static ENTERED: Cell = Cell::new(false)); @@ -48,58 +46,62 @@ pub(crate) fn try_enter() -> Option { // // This is hidden for a reason. Do not use without fully understanding // executors. Misuing can easily cause your program to deadlock. -#[cfg(feature = "rt-full")] -pub(crate) fn exit R, R>(f: F) -> R { - // Reset in case the closure panics - struct Reset; - impl Drop for Reset { - fn drop(&mut self) { - ENTERED.with(|c| { - c.set(true); - }); +cfg_rt_threaded! { + #[cfg(feature = "blocking")] + pub(crate) fn exit R, R>(f: F) -> R { + // Reset in case the closure panics + struct Reset; + impl Drop for Reset { + fn drop(&mut self) { + ENTERED.with(|c| { + c.set(true); + }); + } } + + ENTERED.with(|c| { + debug_assert!(c.get()); + c.set(false); + }); + + let reset = Reset; + let ret = f(); + ::std::mem::forget(reset); + + ENTERED.with(|c| { + assert!(!c.get(), "closure claimed permanent executor"); + c.set(true); + }); + + ret } - ENTERED.with(|c| { - debug_assert!(c.get()); - c.set(false); - }); + impl Enter { + /// Blocks the thread on the specified future, returning the value with + /// which that future completes. + pub(crate) fn block_on(&mut self, mut f: F) -> F::Output + where + F: std::future::Future, + { + use crate::runtime::park::{CachedParkThread, Park}; + use std::pin::Pin; + use std::task::Context; + use std::task::Poll::Ready; - let reset = Reset; - let ret = f(); - ::std::mem::forget(reset); + let mut park = CachedParkThread::new(); + let waker = park.unpark().into_waker(); + let mut cx = Context::from_waker(&waker); - ENTERED.with(|c| { - assert!(!c.get(), "closure claimed permanent executor"); - c.set(true); - }); + // `block_on` takes ownership of `f`. Once it is pinned here, the original `f` binding can + // no longer be accessed, making the pinning safe. + let mut f = unsafe { Pin::new_unchecked(&mut f) }; - ret -} - -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::runtime::park::{CachedParkThread, Park}; - use std::pin::Pin; - use std::task::Context; - use std::task::Poll::Ready; - - let mut park = CachedParkThread::new(); - let waker = park.unpark().into_waker(); - let mut cx = Context::from_waker(&waker); - - // `block_on` takes ownership of `f`. Once it is pinned here, the original `f` binding can - // no longer be accessed, making the pinning safe. - let mut f = unsafe { Pin::new_unchecked(&mut f) }; - - loop { - if let Ready(v) = f.as_mut().poll(&mut cx) { - return v; + loop { + if let Ready(v) = f.as_mut().poll(&mut cx) { + return v; + } + park.park().unwrap(); } - park.park().unwrap(); } } } diff --git a/tokio/src/runtime/global.rs b/tokio/src/runtime/global.rs index 146b678c6..d8b7ed9b9 100644 --- a/tokio/src/runtime/global.rs +++ b/tokio/src/runtime/global.rs @@ -1,9 +1,6 @@ use crate::runtime::basic_scheduler; use crate::task::JoinHandle; -#[cfg(feature = "rt-full")] -use crate::runtime::thread_pool; - use std::cell::Cell; use std::future::Future; @@ -16,7 +13,7 @@ enum State { Basic(*const basic_scheduler::SchedulerPriv), // default executor is a thread pool instance. - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] ThreadPool(*const thread_pool::Spawner), } @@ -34,7 +31,7 @@ where T::Output: Send + 'static, { EXECUTOR.with(|current_executor| match current_executor.get() { - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] State::ThreadPool(thread_pool_ptr) => { let thread_pool = unsafe { &*thread_pool_ptr }; thread_pool.spawn(future) @@ -75,12 +72,15 @@ pub(super) fn basic_scheduler_is_current(basic_scheduler: &basic_scheduler::Sche }) } -#[cfg(feature = "rt-full")] -pub(super) fn with_thread_pool(thread_pool: &thread_pool::Spawner, f: F) -> R -where - F: FnOnce() -> R, -{ - with_state(State::ThreadPool(thread_pool as *const _), f) +cfg_rt_threaded! { + use crate::runtime::thread_pool; + + pub(super) fn with_thread_pool(thread_pool: &thread_pool::Spawner, f: F) -> R + where + F: FnOnce() -> R, + { + with_state(State::ThreadPool(thread_pool as *const _), f) + } } fn with_state(state: State, f: F) -> R diff --git a/tokio/src/runtime/handle.rs b/tokio/src/runtime/handle.rs index 533477781..f8119e655 100644 --- a/tokio/src/runtime/handle.rs +++ b/tokio/src/runtime/handle.rs @@ -1,13 +1,15 @@ -#[cfg(feature = "rt-core")] -use crate::runtime::basic_scheduler; -#[cfg(feature = "rt-full")] -use crate::runtime::thread_pool; use crate::runtime::{blocking, io, time}; -#[cfg(feature = "rt-core")] -use crate::task::JoinHandle; -#[cfg(feature = "rt-core")] -use std::future::Future; +cfg_rt_core! { + use crate::runtime::basic_scheduler; + use crate::task::JoinHandle; + + use std::future::Future; +} + +cfg_rt_threaded! { + use crate::runtime::thread_pool; +} /// Handle to the runtime #[derive(Debug, Clone)] @@ -31,57 +33,11 @@ pub(super) enum Kind { Shell, #[cfg(feature = "rt-core")] Basic(basic_scheduler::Spawner), - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] ThreadPool(thread_pool::Spawner), } impl Handle { - /// Spawn a future onto the Tokio runtime. - /// - /// This spawns the given future onto the runtime's executor, usually a - /// thread pool. The thread pool is then responsible for polling the future - /// until it completes. - /// - /// See [module level][mod] documentation for more details. - /// - /// [mod]: index.html - /// - /// # Examples - /// - /// ``` - /// use tokio::runtime::Runtime; - /// - /// # fn dox() { - /// // Create the runtime - /// let rt = Runtime::new().unwrap(); - /// let handle = rt.handle(); - /// - /// // Spawn a future onto the runtime - /// handle.spawn(async { - /// println!("now running on a worker thread"); - /// }); - /// # } - /// ``` - /// - /// # Panics - /// - /// 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-core")] - pub fn spawn(&self, future: F) -> JoinHandle - where - F: Future + Send + 'static, - F::Output: Send + 'static, - { - match &self.kind { - Kind::Shell => panic!("spawning not enabled for runtime"), - #[cfg(feature = "rt-core")] - Kind::Basic(spawner) => spawner.spawn(future), - #[cfg(feature = "rt-full")] - Kind::ThreadPool(spawner) => spawner.spawn(future), - } - } - /// Enter the runtime context pub fn enter(&self, f: F) -> R where @@ -94,9 +50,58 @@ impl Handle { Kind::Shell => f(), #[cfg(feature = "rt-core")] Kind::Basic(spawner) => spawner.enter(f), - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] Kind::ThreadPool(spawner) => spawner.enter(f), }) }) } } + +cfg_rt_core! { + impl Handle { + /// Spawn a future onto the Tokio runtime. + /// + /// This spawns the given future onto the runtime's executor, usually a + /// thread pool. The thread pool is then responsible for polling the future + /// until it completes. + /// + /// See [module level][mod] documentation for more details. + /// + /// [mod]: index.html + /// + /// # Examples + /// + /// ``` + /// use tokio::runtime::Runtime; + /// + /// # fn dox() { + /// // Create the runtime + /// let rt = Runtime::new().unwrap(); + /// let handle = rt.handle(); + /// + /// // Spawn a future onto the runtime + /// handle.spawn(async { + /// println!("now running on a worker thread"); + /// }); + /// # } + /// ``` + /// + /// # Panics + /// + /// This function panics if the spawn fails. Failure occurs if the executor + /// is currently at capacity and is unable to spawn a new future. + pub fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + match &self.kind { + Kind::Shell => panic!("spawning not enabled for runtime"), + #[cfg(feature = "rt-core")] + Kind::Basic(spawner) => spawner.spawn(future), + #[cfg(feature = "rt-threaded")] + Kind::ThreadPool(spawner) => spawner.spawn(future), + } + } + } +} diff --git a/tokio/src/runtime/io.rs b/tokio/src/runtime/io.rs index 35009a67f..564e51a65 100644 --- a/tokio/src/runtime/io.rs +++ b/tokio/src/runtime/io.rs @@ -3,13 +3,10 @@ //! shells. This isolates the complexity of dealing with conditional //! compilation. -pub(crate) use self::variant::*; - /// Re-exported for convenience. pub(crate) use std::io::Result; -#[cfg(feature = "io-driver")] -mod variant { +cfg_io_driver! { use crate::net::driver; use std::io; @@ -38,8 +35,7 @@ mod variant { } } -#[cfg(not(feature = "io-driver"))] -mod variant { +cfg_not_io_driver! { use crate::runtime::park::ParkThread; use std::io; diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index 8f66b725d..a022f955f 100644 --- a/tokio/src/runtime/mod.rs +++ b/tokio/src/runtime/mod.rs @@ -132,10 +132,10 @@ #[macro_use] mod tests; -#[cfg(feature = "rt-core")] -mod basic_scheduler; -#[cfg(feature = "rt-core")] -use self::basic_scheduler::BasicScheduler; +cfg_rt_core! { + mod basic_scheduler; + use basic_scheduler::BasicScheduler; +} mod blocking; use blocking::BlockingPool; @@ -146,10 +146,10 @@ pub use self::builder::Builder; pub(crate) mod enter; use self::enter::enter; -#[cfg(feature = "rt-core")] -mod global; -#[cfg(feature = "rt-core")] -pub(crate) use self::global::spawn; +cfg_rt_core! { + mod global; + pub(crate) use global::spawn; +} mod handle; pub use self::handle::Handle; @@ -164,13 +164,14 @@ use self::shell::Shell; mod time; -#[cfg(feature = "rt-full")] -pub(crate) mod thread_pool; -#[cfg(feature = "rt-full")] -use self::thread_pool::ThreadPool; +cfg_rt_threaded! { + pub(crate) mod thread_pool; + use self::thread_pool::ThreadPool; +} -#[cfg(feature = "rt-core")] -use crate::task::JoinHandle; +cfg_rt_core! { + use crate::task::JoinHandle; +} use std::future::Future; @@ -223,7 +224,7 @@ enum Kind { Basic(BasicScheduler), /// Execute tasks across multiple threads. - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] ThreadPool(ThreadPool), } @@ -254,10 +255,10 @@ impl Runtime { /// /// [mod]: index.html pub fn new() -> io::Result { - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] let ret = Builder::new().threaded_scheduler().build(); - #[cfg(all(not(feature = "rt-full"), feature = "rt-core"))] + #[cfg(all(not(feature = "rt-threaded"), feature = "rt-core"))] let ret = Builder::new().basic_scheduler().build(); #[cfg(not(feature = "rt-core"))] @@ -304,7 +305,7 @@ impl Runtime { { match &self.kind { Kind::Shell(_) => panic!("task execution disabled"), - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] Kind::ThreadPool(exec) => exec.spawn(future), Kind::Basic(exec) => exec.spawn(future), } @@ -330,7 +331,7 @@ impl Runtime { Kind::Shell(exec) => exec.block_on(future), #[cfg(feature = "rt-core")] Kind::Basic(exec) => exec.block_on(future), - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] Kind::ThreadPool(exec) => exec.block_on(future), }) } diff --git a/tokio/src/runtime/park/mod.rs b/tokio/src/runtime/park/mod.rs index 122128323..bae96d9dd 100644 --- a/tokio/src/runtime/park/mod.rs +++ b/tokio/src/runtime/park/mod.rs @@ -45,7 +45,7 @@ //! [mio]: https://docs.rs/mio/0.6/mio/struct.Poll.html mod thread; -#[cfg(feature = "rt-full")] +#[cfg(feature = "rt-threaded")] pub(crate) use self::thread::CachedParkThread; #[cfg(not(feature = "io-driver"))] pub(crate) use self::thread::ParkThread; diff --git a/tokio/src/runtime/park/thread.rs b/tokio/src/runtime/park/thread.rs index 7d9bf4651..71bd5b92a 100644 --- a/tokio/src/runtime/park/thread.rs +++ b/tokio/src/runtime/park/thread.rs @@ -168,7 +168,7 @@ impl CachedParkThread { /// /// This type cannot be moved to other threads, so it should be created on /// the thread that the caller intends to park. - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] pub(crate) fn new() -> CachedParkThread { CachedParkThread { _anchor: PhantomData, @@ -217,7 +217,7 @@ impl Unpark for UnparkThread { } } -#[cfg(feature = "rt-full")] +#[cfg(feature = "rt-threaded")] mod waker { use super::{Inner, UnparkThread}; use crate::loom::sync::Arc; diff --git a/tokio/src/runtime/thread_pool/mod.rs b/tokio/src/runtime/thread_pool/mod.rs index b45d37072..4b23c3b90 100644 --- a/tokio/src/runtime/thread_pool/mod.rs +++ b/tokio/src/runtime/thread_pool/mod.rs @@ -22,7 +22,9 @@ mod shutdown; mod worker; -pub(crate) use worker::block_in_place; +cfg_blocking! { + pub(crate) use worker::block_in_place; +} /// Unit tests #[cfg(test)] diff --git a/tokio/src/runtime/thread_pool/worker.rs b/tokio/src/runtime/thread_pool/worker.rs index 2de2101e2..cf6b66d84 100644 --- a/tokio/src/runtime/thread_pool/worker.rs +++ b/tokio/src/runtime/thread_pool/worker.rs @@ -15,24 +15,26 @@ thread_local! { static ON_BLOCK: Cell> = Cell::new(None) } -pub(crate) fn block_in_place(f: F) -> R -where - F: FnOnce() -> R, -{ - // Make the current worker give away its Worker to another thread so that we can safely block - // this one without preventing progress on other futures the worker owns. - ON_BLOCK.with(|ob| { - let allow_blocking = ob - .get() - .expect("can only call blocking when on Tokio runtime"); +cfg_blocking! { + pub(crate) fn block_in_place(f: F) -> R + where + F: FnOnce() -> R, + { + // Make the current worker give away its Worker to another thread so that we can safely block + // this one without preventing progress on other futures the worker owns. + ON_BLOCK.with(|ob| { + let allow_blocking = ob + .get() + .expect("can only call blocking when on Tokio runtime"); - // This is safe, because ON_BLOCK was set from an &mut dyn FnMut in the worker that wraps - // the worker's operation, and is unset just prior to when the FnMut is dropped. - let allow_blocking = unsafe { &*allow_blocking }; + // This is safe, because ON_BLOCK was set from an &mut dyn FnMut in the worker that wraps + // the worker's operation, and is unset just prior to when the FnMut is dropped. + let allow_blocking = unsafe { &*allow_blocking }; - allow_blocking(); - f() - }) + allow_blocking(); + f() + }) + } } pub(crate) struct Worker { diff --git a/tokio/src/runtime/time.rs b/tokio/src/runtime/time.rs index c04c2c966..b72971f86 100644 --- a/tokio/src/runtime/time.rs +++ b/tokio/src/runtime/time.rs @@ -3,10 +3,7 @@ //! shells. This isolates the complexity of dealing with conditional //! compilation. -pub(crate) use self::variant::*; - -#[cfg(feature = "time")] -mod variant { +cfg_time! { use crate::runtime::io; use crate::time::{self, driver}; @@ -35,8 +32,7 @@ mod variant { } } -#[cfg(not(feature = "time"))] -mod variant { +cfg_not_time! { use crate::runtime::io; pub(crate) type Clock = (); diff --git a/tokio/src/sync/mod.rs b/tokio/src/sync/mod.rs index cf0038166..40566c0c0 100644 --- a/tokio/src/sync/mod.rs +++ b/tokio/src/sync/mod.rs @@ -13,44 +13,41 @@ //! - [watch](watch/index.html), a single-producer, multi-consumer channel that //! only stores the **most recently** sent value. -macro_rules! debug { - ($($t:tt)*) => { - if false { - println!($($t)*); - } +cfg_sync! { + mod barrier; + pub use barrier::{Barrier, BarrierWaitResult}; + + pub mod mpsc; + + mod mutex; + pub use mutex::{Mutex, MutexGuard}; + + pub mod oneshot; + + pub(crate) mod semaphore; + + mod task; + pub(crate) use task::AtomicWaker; + + pub mod watch; +} + +cfg_not_sync! { + cfg_atomic_waker! { + mod task; + pub(crate) use task::AtomicWaker; + } + + cfg_rt_threaded! { + pub(crate) mod oneshot; + } + + cfg_signal! { + pub(crate) mod mpsc; + pub(crate) mod semaphore; } } -macro_rules! if_loom { - ($($t:tt)*) => {{ - #[cfg(loom)] - const LOOM: bool = true; - #[cfg(not(loom))] - const LOOM: bool = false; - - if LOOM { - $($t)* - } - }} -} - -mod barrier; -pub use barrier::{Barrier, BarrierWaitResult}; - -pub mod mpsc; - -mod mutex; -pub use mutex::{Mutex, MutexGuard}; - -pub mod oneshot; - -pub mod semaphore; - -mod task; -pub(crate) use task::AtomicWaker; - -pub mod watch; - /// Unit tests #[cfg(test)] mod tests; diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs index 523dde75b..d635e1380 100644 --- a/tokio/src/sync/mpsc/bounded.rs +++ b/tokio/src/sync/mpsc/bounded.rs @@ -161,12 +161,13 @@ impl Receiver { impl Unpin for Receiver {} -#[cfg(feature = "stream")] -impl futures_core::Stream for Receiver { - type Item = T; +cfg_stream! { + impl futures_core::Stream for Receiver { + type Item = T; - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.poll_recv(cx) + fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.poll_recv(cx) + } } } diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs index 03f353391..4030e3801 100644 --- a/tokio/src/sync/mpsc/chan.rs +++ b/tokio/src/sync/mpsc/chan.rs @@ -299,12 +299,6 @@ where // second time here. try_recv!(); - debug!( - "recv; rx_closed = {:?}; is_idle = {:?}", - rx_fields.rx_closed, - self.inner.semaphore.is_idle() - ); - if rx_fields.rx_closed && self.inner.semaphore.is_idle() { Ready(None) } else { diff --git a/tokio/src/sync/mpsc/list.rs b/tokio/src/sync/mpsc/list.rs index eecc4da33..dc9564032 100644 --- a/tokio/src/sync/mpsc/list.rs +++ b/tokio/src/sync/mpsc/list.rs @@ -169,7 +169,6 @@ impl Tx { } pub(crate) unsafe fn reclaim_block(&self, mut block: NonNull>) { - debug!("+ reclaim_block({:p})", block); // The block has been removed from the linked list and ownership // is reclaimed. // @@ -206,7 +205,6 @@ impl Tx { } if !reused { - debug!(" + block freed {:p}", block); let _ = Box::from_raw(block.as_ptr()); } } @@ -226,7 +224,6 @@ impl Rx { pub(crate) fn pop(&mut self, tx: &Tx) -> Option> { // Advance `head`, if needed if !self.try_advancing_head() { - debug!("+ !self.try_advancing_head() -> false"); return None; } @@ -276,8 +273,6 @@ impl Rx { } fn reclaim_blocks(&mut self, tx: &Tx) { - debug!("+ reclaim_blocks()"); - while self.free_head != self.head { unsafe { // Get a handle to the block that will be freed and update @@ -316,7 +311,6 @@ impl Rx { /// Effectively `Drop` all the blocks. Should only be called once, when /// the list is dropping. pub(super) unsafe fn free_blocks(&mut self) { - debug!("+ free_blocks()"); debug_assert_ne!(self.free_head, NonNull::dangling()); let mut cur = Some(self.free_head); @@ -331,7 +325,6 @@ impl Rx { while let Some(block) = cur { cur = block.as_ref().load_next(Relaxed); - debug!(" + free: block = {:p}", block); drop(Box::from_raw(block.as_ptr())); } } diff --git a/tokio/src/sync/mpsc/mod.rs b/tokio/src/sync/mpsc/mod.rs index 7927dde6a..60ae60cde 100644 --- a/tokio/src/sync/mpsc/mod.rs +++ b/tokio/src/sync/mpsc/mod.rs @@ -1,3 +1,5 @@ +#![cfg_attr(not(feature = "sync"), allow(dead_code, unreachable_pub))] + //! A multi-producer, single-consumer queue for sending values across //! asynchronous tasks. //! diff --git a/tokio/src/sync/oneshot.rs b/tokio/src/sync/oneshot.rs index 7b84f3194..ed3801c81 100644 --- a/tokio/src/sync/oneshot.rs +++ b/tokio/src/sync/oneshot.rs @@ -1,3 +1,5 @@ +#![cfg_attr(not(feature = "sync"), allow(dead_code, unreachable_pub))] + //! A channel for sending a single message between asynchronous tasks. use crate::loom::cell::CausalCell; diff --git a/tokio/src/sync/semaphore.rs b/tokio/src/sync/semaphore.rs index b4a093f85..dab73a099 100644 --- a/tokio/src/sync/semaphore.rs +++ b/tokio/src/sync/semaphore.rs @@ -1,3 +1,5 @@ +#![cfg_attr(not(feature = "sync"), allow(dead_code, unreachable_pub))] + //! Thread-safe, asynchronous counting semaphore. //! //! A `Semaphore` instance holds a set of permits. Permits are used to @@ -24,7 +26,7 @@ use std::task::{Context, Poll}; use std::usize; /// Futures-aware semaphore. -pub struct Semaphore { +pub(crate) struct Semaphore { /// Tracks both the waiter queue tail pointer and the number of remaining /// permits. state: AtomicUsize, @@ -51,18 +53,18 @@ pub struct Semaphore { /// is the user's responsibility to ensure that `Permit::release` is called /// before dropping the permit. #[derive(Debug)] -pub struct Permit { +pub(crate) struct Permit { waiter: Option>, state: PermitState, } /// Error returned by `Permit::poll_acquire`. #[derive(Debug)] -pub struct AcquireError(()); +pub(crate) struct AcquireError(()); /// Error returned by `Permit::try_acquire`. #[derive(Debug)] -pub struct TryAcquireError { +pub(crate) struct TryAcquireError { kind: ErrorKind, } @@ -150,7 +152,7 @@ impl Semaphore { /// # Panics /// /// Panics if `permits` is zero. - pub fn new(permits: usize) -> Semaphore { + pub(crate) fn new(permits: usize) -> Semaphore { let stub = Box::new(WaiterNode::new()); let ptr = NonNull::new(&*stub as *const _ as *mut _).unwrap(); @@ -168,7 +170,7 @@ impl Semaphore { } /// Returns the current number of available permits - pub fn available_permits(&self) -> usize { + pub(crate) fn available_permits(&self) -> usize { let curr = SemState::load(&self.state, Acquire); curr.available_permits() } @@ -181,8 +183,6 @@ impl Semaphore { // Load the current state let mut curr = SemState::load(&self.state, Acquire); - debug!(" + poll_permit; sem-state = {:?}", curr); - // Tracks a *mut WaiterNode representing an Arc clone. // // This avoids having to bump the ref count unless required. @@ -210,8 +210,6 @@ impl Semaphore { } if !next.acquire_permit(&self.stub) { - debug!(" + poll_permit -- no permits"); - debug_assert!(curr.waiter().is_some()); if maybe_strong.is_none() { @@ -223,10 +221,7 @@ impl Semaphore { waiter.register(cx); - debug!(" + poll_permit -- to_queued_waiting"); - if !waiter.to_queued_waiting() { - debug!(" + poll_permit; waiter already queued"); // The node is alrady queued, there is no further work // to do. return Pending; @@ -243,14 +238,11 @@ impl Semaphore { next.set_waiter(maybe_strong.unwrap()); } - debug!(" + poll_permit -- pre-CAS; next = {:?}", next); - debug_assert_ne!(curr.0, 0); debug_assert_ne!(next.0, 0); match next.compare_exchange(&self.state, curr, AcqRel, Acquire) { Ok(_) => { - debug!(" + poll_permit -- CAS ok"); match curr.waiter() { Some(prev_waiter) => { let waiter = maybe_strong.unwrap(); @@ -260,13 +252,9 @@ impl Semaphore { prev_waiter.as_ref().next.store(waiter.as_ptr(), Release); } - debug!(" + poll_permit -- waiter pushed"); - return Pending; } None => { - debug!(" + poll_permit -- permit acquired"); - undo_strong!(); return Ready(Ok(())); @@ -282,15 +270,11 @@ impl Semaphore { /// Close the semaphore. This prevents the semaphore from issuing new /// permits and notifies all pending waiters. - pub fn close(&self) { - debug!("+ Semaphore::close"); - + pub(crate) fn close(&self) { // Acquire the `rx_lock`, setting the "closed" flag on the lock. let prev = self.rx_lock.fetch_or(1, AcqRel); - debug!(" + close -- rx_lock.fetch_add(1)"); if prev != 0 { - debug!("+ close -- locked; prev = {}", prev); // Another thread has the lock and will be responsible for notifying // pending waiters. return; @@ -300,9 +284,7 @@ impl Semaphore { } /// Add `n` new permits to the semaphore. - pub fn add_permits(&self, n: usize) { - debug!(" + add_permits; n = {}", n); - + pub(crate) fn add_permits(&self, n: usize) { if n == 0 { return; } @@ -310,10 +292,8 @@ impl Semaphore { // TODO: Handle overflow. A panic is not sufficient, the process must // abort. let prev = self.rx_lock.fetch_add(n << 1, AcqRel); - debug!(" + add_permits; rx_lock.fetch_add(n << 1); n = {}", n); if prev != 0 { - debug!(" + add_permits -- locked; prev = {}", prev); // Another thread has the lock and will be responsible for notifying // pending waiters. return; @@ -324,11 +304,6 @@ impl Semaphore { fn add_permits_locked(&self, mut rem: usize, mut closed: bool) { while rem > 0 || closed { - debug!( - " + add_permits_locked -- iter; rem = {}; closed = {:?}", - rem, closed - ); - if closed { SemState::fetch_set_closed(&self.state, AcqRel); } @@ -340,28 +315,16 @@ impl Semaphore { let actual = if closed { let actual = self.rx_lock.fetch_sub(n | 1, AcqRel); - debug!( - " + add_permits_locked; rx_lock.fetch_sub(n | 1); n = {}; actual={}", - n, actual - ); - closed = false; actual } else { let actual = self.rx_lock.fetch_sub(n, AcqRel); - debug!( - " + add_permits_locked; rx_lock.fetch_sub(n); n = {}; actual={}", - n, actual - ); - closed = actual & 1 == 1; actual }; rem = (actual >> 1) - rem; } - - debug!(" + add_permits; done"); } /// Release a specific amount of permits to the semaphore @@ -377,11 +340,8 @@ impl Semaphore { } }; - debug!(" + release_n -- notify"); - if waiter.notify(closed) { n = n.saturating_sub(1); - debug!(" + release_n -- dec"); } } } @@ -392,8 +352,6 @@ impl Semaphore { /// there are no more waiters to pop, `rem` is used to set the available /// permits. fn pop(&self, rem: usize, closed: bool) -> Option> { - debug!(" + pop; rem = {}", rem); - 'outer: loop { unsafe { let mut head = self.head.with(|head| *head); @@ -402,8 +360,6 @@ impl Semaphore { let stub = self.stub(); if head == stub { - debug!(" + pop; head == stub"); - let next = match NonNull::new(next_ptr) { Some(next) => next, None => { @@ -429,7 +385,6 @@ impl Semaphore { loop { if curr.has_waiter(&self.stub) { // Inconsistent - debug!(" + pop; inconsistent 1"); thread::yield_now(); continue 'outer; } @@ -456,8 +411,6 @@ impl Semaphore { } }; - debug!(" + pop; got next waiter"); - self.head.with_mut(|head| *head = next); head = next; next_ptr = next.as_ref().next.load(Acquire); @@ -476,7 +429,6 @@ impl Semaphore { if tail != head { // Inconsistent - debug!(" + pop; inconsistent 2"); thread::yield_now(); continue 'outer; } @@ -492,7 +444,6 @@ impl Semaphore { } // Inconsistent state, loop - debug!(" + pop; inconsistent 3"); thread::yield_now(); } } @@ -549,16 +500,7 @@ impl Permit { /// Create a new `Permit`. /// /// The permit begins in the "unacquired" state. - /// - /// # Examples - /// - /// ``` - /// use tokio::sync::semaphore::Permit; - /// - /// let permit = Permit::new(); - /// assert!(!permit.is_acquired()); - /// ``` - pub fn new() -> Permit { + pub(crate) fn new() -> Permit { Permit { waiter: None, state: PermitState::Idle, @@ -566,13 +508,13 @@ impl Permit { } /// Returns true if the permit has been acquired - pub fn is_acquired(&self) -> bool { + pub(crate) fn is_acquired(&self) -> bool { self.state == PermitState::Acquired } /// Try to acquire the permit. If no permits are available, the current task /// is notified once a new permit becomes available. - pub fn poll_acquire( + pub(crate) fn poll_acquire( &mut self, cx: &mut Context<'_>, semaphore: &Semaphore, @@ -607,7 +549,7 @@ impl Permit { } /// Try to acquire the permit. - pub fn try_acquire(&mut self, semaphore: &Semaphore) -> Result<(), TryAcquireError> { + pub(crate) fn try_acquire(&mut self, semaphore: &Semaphore) -> Result<(), TryAcquireError> { match self.state { PermitState::Idle => {} PermitState::Waiting => { @@ -635,7 +577,7 @@ impl Permit { } /// Release a permit back to the semaphore - pub fn release(&mut self, semaphore: &Semaphore) { + pub(crate) fn release(&mut self, semaphore: &Semaphore) { if self.forget2() { semaphore.add_permits(1); } @@ -648,7 +590,7 @@ impl Permit { /// /// Repeatedly calling `forget` without associated calls to `add_permit` /// will result in the semaphore losing all permits. - pub fn forget(&mut self) { + pub(crate) fn forget(&mut self) { self.forget2(); } @@ -711,7 +653,7 @@ impl TryAcquireError { } /// Returns true if the error was caused by a closed semaphore. - pub fn is_closed(&self) -> bool { + pub(crate) fn is_closed(&self) -> bool { match self.kind { ErrorKind::Closed => true, _ => false, @@ -720,7 +662,7 @@ impl TryAcquireError { /// Returns true if the error was caused by calling `try_acquire` on a /// semaphore with no available permits. - pub fn is_no_permits(&self) -> bool { + pub(crate) fn is_no_permits(&self) -> bool { match self.kind { ErrorKind::NoPermits => true, _ => false, @@ -857,14 +799,10 @@ impl WaiterNode { match next.compare_exchange(&self.state, curr, AcqRel, Acquire) { Ok(_) => match curr { QueuedWaiting => { - debug!(" + notify -- task notified"); self.waker.wake(); return true; } - other => { - debug!(" + notify -- not notified; state = {:?}", other); - return false; - } + _ => return false, }, Err(actual) => curr = actual, } @@ -1021,7 +959,6 @@ impl SemState { /// Load the state from an AtomicUsize. fn load(cell: &AtomicUsize, ordering: Ordering) -> SemState { let value = cell.load(ordering); - debug!(" + SemState::load; value = {}", value); SemState(value) } @@ -1044,13 +981,6 @@ impl SemState { let res = cell.compare_exchange(prev.to_usize(), self.to_usize(), success, failure); - debug!( - " + SemState::compare_exchange; prev = {}; next = {}; result = {:?}", - prev.to_usize(), - self.to_usize(), - res - ); - res.map(SemState).map_err(SemState) } diff --git a/tokio/src/sync/task/atomic_waker.rs b/tokio/src/sync/task/atomic_waker.rs index eaad17c1a..49a0ac048 100644 --- a/tokio/src/sync/task/atomic_waker.rs +++ b/tokio/src/sync/task/atomic_waker.rs @@ -170,10 +170,8 @@ impl AtomicWaker { where W: WakerRef, { - debug!(" + register_task"); match self.state.compare_and_swap(WAITING, REGISTERING, Acquire) { WAITING => { - debug!(" + WAITING"); unsafe { // Locked acquired, update the waker cell self.waker.with_mut(|t| *t = Some(waker.into_waker())); @@ -214,7 +212,6 @@ impl AtomicWaker { } } WAKING => { - debug!(" + WAKING"); // Currently in the process of waking the task, i.e., // `wake` is currently being called on the old waker. // So, we call wake on the new waker. @@ -240,7 +237,6 @@ impl AtomicWaker { /// /// If `register` has not been called yet, then this does nothing. pub(crate) fn wake(&self) { - debug!(" + wake"); if let Some(waker) = self.take_waker() { waker.wake(); } @@ -249,24 +245,20 @@ impl AtomicWaker { /// Attempts to take the `Waker` value out of the `AtomicWaker` with the /// intention that the caller will wake the task later. pub(crate) fn take_waker(&self) -> Option { - debug!(" + take_waker"); // AcqRel ordering is used in order to acquire the value of the `waker` // cell as well as to establish a `release` ordering with whatever // memory the `AtomicWaker` is associated with. match self.state.fetch_or(WAKING, AcqRel) { WAITING => { - debug!(" + WAITING"); // The waking lock has been acquired. let waker = unsafe { self.waker.with_mut(|t| (*t).take()) }; // Release the lock self.state.fetch_and(!WAKING, Release); - debug!(" + Done taking"); waker } state => { - debug!(" + state = {:?}", state); // There is a concurrent thread currently updating the // associated waker. // diff --git a/tokio/src/sync/tests/loom_list.rs b/tokio/src/sync/tests/loom_list.rs index 4f7746d58..4067f865c 100644 --- a/tokio/src/sync/tests/loom_list.rs +++ b/tokio/src/sync/tests/loom_list.rs @@ -21,17 +21,14 @@ fn smoke() { for i in 0..NUM_MSG { tx.push((th, i)); } - debug!(" + tx thread done"); }); } let mut next = vec![0; NUM_TX]; loop { - debug!(" + rx.pop()"); match rx.pop(&tx) { Some(Value((th, v))) => { - debug!(" + pop() -> Some(Value({}))", v); assert_eq!(v, next[th]); next[th] += 1; @@ -43,7 +40,6 @@ fn smoke() { panic!(); } None => { - debug!(" + pop() -> None"); thread::yield_now(); } } diff --git a/tokio/src/sync/tests/mod.rs b/tokio/src/sync/tests/mod.rs index 8da739f93..06d18e9a9 100644 --- a/tokio/src/sync/tests/mod.rs +++ b/tokio/src/sync/tests/mod.rs @@ -1,17 +1,12 @@ -#[cfg(not(loom))] -mod atomic_waker; +cfg_not_loom! { + mod atomic_waker; + mod semaphore; +} -#[cfg(loom)] -mod loom_atomic_waker; - -#[cfg(loom)] -mod loom_list; - -#[cfg(loom)] -mod loom_mpsc; - -#[cfg(loom)] -mod loom_oneshot; - -#[cfg(loom)] -mod loom_semaphore; +cfg_loom! { + mod loom_atomic_waker; + mod loom_list; + mod loom_mpsc; + mod loom_oneshot; + mod loom_semaphore; +} diff --git a/tokio/tests/sync_semaphore.rs b/tokio/src/sync/tests/semaphore.rs similarity index 97% rename from tokio/tests/sync_semaphore.rs rename to tokio/src/sync/tests/semaphore.rs index 93660532d..86dd7da58 100644 --- a/tokio/tests/sync_semaphore.rs +++ b/tokio/src/sync/tests/semaphore.rs @@ -1,6 +1,4 @@ -#![warn(rust_2018_idioms)] - -use tokio::sync::semaphore::{Permit, Semaphore}; +use crate::sync::semaphore::{Permit, Semaphore}; use tokio_test::task; use tokio_test::{assert_pending, assert_ready_err, assert_ready_ok}; diff --git a/tokio/src/task/blocking.rs b/tokio/src/task/blocking.rs index e0c12d4a2..4445bb729 100644 --- a/tokio/src/task/blocking.rs +++ b/tokio/src/task/blocking.rs @@ -1,62 +1,65 @@ use crate::blocking; use crate::task::JoinHandle; -/// Run the provided blocking function without blocking the executor. -/// -/// In general, issuing a blocking call or performing a lot of compute in a -/// future without yielding is not okay, as it may prevent the executor from -/// driving other futures forward. If you run a closure through this method, -/// the current executor thread will relegate all its executor duties to another -/// (possibly new) thread, and only then poll the task. Note that this requires -/// additional synchronization. -/// -/// # Examples -/// -/// ``` -/// use tokio::task; -/// -/// # async fn docs() { -/// task::block_in_place(move || { -/// // do some compute-heavy work or call synchronous code -/// }); -/// # } -/// ``` -#[cfg(feature = "rt-full")] -pub fn block_in_place(f: F) -> R -where - F: FnOnce() -> R, -{ - use crate::runtime::{enter, thread_pool}; +cfg_rt_threaded! { + /// Run the provided blocking function without blocking the executor. + /// + /// In general, issuing a blocking call or performing a lot of compute in a + /// future without yielding is not okay, as it may prevent the executor from + /// driving other futures forward. If you run a closure through this method, + /// the current executor thread will relegate all its executor duties to another + /// (possibly new) thread, and only then poll the task. Note that this requires + /// additional synchronization. + /// + /// # Examples + /// + /// ``` + /// use tokio::task; + /// + /// # async fn docs() { + /// task::block_in_place(move || { + /// // do some compute-heavy work or call synchronous code + /// }); + /// # } + /// ``` + pub fn block_in_place(f: F) -> R + where + F: FnOnce() -> R, + { + use crate::runtime::{enter, thread_pool}; - enter::exit(|| thread_pool::block_in_place(f)) + enter::exit(|| thread_pool::block_in_place(f)) + } } -/// Run the provided closure on a thread where blocking is acceptable. -/// -/// In general, issuing a blocking call or performing a lot of compute in a future without -/// yielding is not okay, as it may prevent the executor from driving other futures forward. -/// A closure that is run through this method will instead be run on a dedicated thread pool for -/// such blocking tasks without holding up the main futures executor. -/// -/// # Examples -/// -/// ``` -/// use tokio::task; -/// -/// # async fn docs() -> Result<(), Box>{ -/// let res = task::spawn_blocking(move || { -/// // do some compute-heavy work or call synchronous code -/// "done computing" -/// }).await?; -/// -/// assert_eq!(res, "done computing"); -/// # Ok(()) -/// # } -/// ``` -pub fn spawn_blocking(f: F) -> JoinHandle -where - F: FnOnce() -> R + Send + 'static, - R: Send + 'static, -{ - blocking::spawn_blocking(f) +cfg_blocking! { + /// Run the provided closure on a thread where blocking is acceptable. + /// + /// In general, issuing a blocking call or performing a lot of compute in a future without + /// yielding is not okay, as it may prevent the executor from driving other futures forward. + /// A closure that is run through this method will instead be run on a dedicated thread pool for + /// such blocking tasks without holding up the main futures executor. + /// + /// # Examples + /// + /// ``` + /// use tokio::task; + /// + /// # async fn docs() -> Result<(), Box>{ + /// let res = task::spawn_blocking(move || { + /// // do some compute-heavy work or call synchronous code + /// "done computing" + /// }).await?; + /// + /// assert_eq!(res, "done computing"); + /// # Ok(()) + /// # } + /// ``` + pub fn spawn_blocking(f: F) -> JoinHandle + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + blocking::spawn_blocking(f) + } } diff --git a/tokio/src/task/mod.rs b/tokio/src/task/mod.rs index d2025330c..7b27ff1b7 100644 --- a/tokio/src/task/mod.rs +++ b/tokio/src/task/mod.rs @@ -1,11 +1,13 @@ //! Asynchronous green-threads. -#[cfg(feature = "blocking")] -mod blocking; -#[cfg(feature = "rt-full")] -pub use blocking::block_in_place; -#[cfg(feature = "blocking")] -pub use blocking::spawn_blocking; +cfg_blocking! { + mod blocking; + pub use blocking::spawn_blocking; + + cfg_rt_threaded! { + pub use blocking::block_in_place; + } +} mod core; use self::core::Cell; @@ -17,10 +19,11 @@ pub use self::error::JoinError; mod harness; use self::harness::Harness; -mod join; -#[cfg(feature = "rt-core")] -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::join::JoinHandle; +cfg_rt_core! { + mod join; + #[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 + pub use self::join::JoinHandle; +} mod list; pub(crate) use self::list::OwnedList; @@ -28,10 +31,10 @@ pub(crate) use self::list::OwnedList; mod raw; use self::raw::RawTask; -#[cfg(feature = "rt-core")] -mod spawn; -#[cfg(feature = "rt-core")] -pub use spawn::spawn; +cfg_rt_core! { + mod spawn; + pub use spawn::spawn; +} mod stack; pub(crate) use self::stack::TransferStack; @@ -81,16 +84,17 @@ pub(crate) trait Schedule: Send + Sync + Sized + 'static { fn schedule(&self, task: Task); } -/// Create a new task without an associated join handle -#[cfg(feature = "rt-full")] -pub(crate) fn background(task: T) -> Task -where - T: Future + Send + 'static, - S: Schedule, -{ - Task { - raw: RawTask::new_background::<_, S>(task), - _p: PhantomData, +cfg_rt_threaded! { + /// Create a new task without an associated join handle + pub(crate) fn background(task: T) -> Task + where + T: Future + Send + 'static, + S: Schedule, + { + Task { + raw: RawTask::new_background::<_, S>(task), + _p: PhantomData, + } } } diff --git a/tokio/src/task/raw.rs b/tokio/src/task/raw.rs index 899e5aa36..c1879344a 100644 --- a/tokio/src/task/raw.rs +++ b/tokio/src/task/raw.rs @@ -54,16 +54,19 @@ pub(super) fn vtable() -> &'static Vtable { } } -impl RawTask { - #[cfg(feature = "rt-full")] - pub(super) fn new_background(task: T) -> RawTask - where - T: Future + Send + 'static, - S: Schedule, - { - RawTask::new::<_, S>(task, State::new_background()) +cfg_rt_threaded! { + impl RawTask { + pub(super) fn new_background(task: T) -> RawTask + where + T: Future + Send + 'static, + S: Schedule, + { + RawTask::new::<_, S>(task, State::new_background()) + } } +} +impl RawTask { pub(super) fn new_joinable(task: T) -> RawTask where T: Future + Send + 'static, diff --git a/tokio/src/task/state.rs b/tokio/src/task/state.rs index e10f0601c..a9284ac9e 100644 --- a/tokio/src/task/state.rs +++ b/tokio/src/task/state.rs @@ -58,7 +58,7 @@ const INITIAL_STATE: usize = NOTIFIED; /// unambiguous modification order. impl State { /// Starts with a ref count of 1 - #[cfg(feature = "rt-full")] + #[cfg(feature = "rt-threaded")] pub(super) fn new_background() -> State { State { val: AtomicUsize::new(INITIAL_STATE), diff --git a/tokio/src/tests/mod.rs b/tokio/src/tests/mod.rs index 0f934f57d..b326561b6 100644 --- a/tokio/src/tests/mod.rs +++ b/tokio/src/tests/mod.rs @@ -1,25 +1,3 @@ -#[macro_export] -/// Assert option is some -macro_rules! assert_some { - ($e:expr) => {{ - match $e { - Some(v) => v, - _ => panic!("expected some, was none"), - } - }}; -} - -#[macro_export] -/// Assert option is none -macro_rules! assert_none { - ($e:expr) => {{ - match $e { - Some(v) => panic!("expected none, was {:?}", v), - _ => {} - } - }}; -} - #[cfg(not(loom))] pub(crate) mod backoff; diff --git a/tokio/src/time/clock.rs b/tokio/src/time/clock.rs index 7727f4048..e44f4925b 100644 --- a/tokio/src/time/clock.rs +++ b/tokio/src/time/clock.rs @@ -4,15 +4,7 @@ //! `test-util` feature flag is enabled, the values returned for `now()` are //! configurable. -#[cfg(feature = "test-util")] -pub(crate) use self::variant::now; -pub(crate) use self::variant::Clock; -#[cfg(feature = "test-util")] -#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 -pub use self::variant::{advance, pause, resume}; - -#[cfg(not(feature = "test-util"))] -mod variant { +cfg_not_test_util! { use crate::time::Instant; #[derive(Debug, Clone)] @@ -40,8 +32,7 @@ mod variant { } } -#[cfg(feature = "test-util")] -mod variant { +cfg_test_util! { use crate::time::{Duration, Instant}; use std::cell::Cell; diff --git a/tokio/tests/fs_file_mocked.rs b/tokio/tests/fs_file_mocked.rs index 2b05d8c45..12463cfc3 100644 --- a/tokio/tests/fs_file_mocked.rs +++ b/tokio/tests/fs_file_mocked.rs @@ -9,6 +9,16 @@ macro_rules! ready { }; } +#[macro_export] +macro_rules! cfg_fs { + ($($item:item)*) => { $($item)* } +} + +#[macro_export] +macro_rules! cfg_io_std { + ($($item:item)*) => { $($item)* } +} + use futures::future; // Load source