itsscb/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2025-09-25 12:00:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: refine feature flags (#1785)

Browse Source 
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.
This commit is contained in:
Carl Lerche 2019-11-18 07:00:55 -08:00 committed by GitHub
parent 13b6e9939e
commit 0d38936b35
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
49 changed files with 1092 additions and 915 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tokio-macros/Cargo.toml
View File

@ -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

33
tokio/Cargo.toml
View File

@ -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"]

12
tokio/src/blocking/mod.rs
View File

@ -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;
}

22
tokio/src/blocking/pool.rs
View File

@ -116,16 +116,20 @@ impl fmt::Debug for BlockingPool {
// ===== impl Spawner =====
impl Spawner {
#[cfg(feature = "rt-full")]
pub(crate) fn spawn_background<F>(&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<F>(&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

30
tokio/src/io/blocking.rs
View File

@ -34,13 +34,14 @@ enum State<T> {
Busy(sys::Blocking<(io::Result<usize>, Buf, T)>),
}
impl<T> Blocking<T> {
#[cfg(feature = "io")]
pub(crate) fn new(inner: T) -> Blocking<T> {
Blocking {
inner: Some(inner),
state: State::Idle(Some(Buf::with_capacity(0))),
need_flush: false,
cfg_io_std! {
impl<T> Blocking<T> {
pub(crate) fn new(inner: T) -> Blocking<T> {
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
}
}
}

79
tokio/src/io/mod.rs
View File

@ -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;
}
}

110
tokio/src/io/util/mod.rs
View File

@ -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;
}
}

81
tokio/src/lib.rs
View File

@ -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<T: Unpin>() {}

8
tokio/src/loom/std/mod.rs
View File

@ -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
}

19
tokio/src/macros/assert.rs Normal file
View File

@ -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),
_ => {}
}
}};
}

217
tokio/src/macros/cfg.rs Normal file
View File

@ -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 )*
}
}

12
tokio/src/macros/loom.rs Normal file
View File

@ -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)*
}
}}
}

17
tokio/src/macros/mod.rs Normal file
View File

@ -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;

8
tokio/src/macros/ready.rs Normal file
View File

@ -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,
}
};
}

4
tokio/src/macros/thread_local.rs Normal file
View File

@ -0,0 +1,4 @@
#[cfg(all(loom, test))]
macro_rules! thread_local {
($($tts:tt)+) => { loom::thread_local!{ $($tts)+ } }
}

270
tokio/src/net/addr.rs
View File

@ -16,6 +16,22 @@ pub trait ToSocketAddrs: sealed::ToSocketAddrsPriv {}
type ReadyFuture<T> = future::Ready<io::Result<T>>;
// ===== impl &impl ToSocketAddrs =====
impl<T: ToSocketAddrs + ?Sized> ToSocketAddrs for &T {}
impl<T> 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<SocketAddr, _> = 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::<Ipv4Addr>() {
let addr = SocketAddrV4::new(addr, port);
let addr = SocketAddr::V4(addr);
return MaybeReady::Ready(Some(addr));
}
if let Ok(addr) = host.parse::<Ipv6Addr>() {
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 = <str as sealed::ToSocketAddrsPriv>::Iter;
type Future = <str as sealed::ToSocketAddrsPriv>::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<SocketAddr, _> = 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<T: ToSocketAddrs + ?Sized> ToSocketAddrs for &T {}
impl ToSocketAddrs for (&str, u16) {}
impl<T> 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::<Ipv4Addr>() {
let addr = SocketAddrV4::new(addr, port);
let addr = SocketAddr::V4(addr);
return MaybeReady::Ready(Some(addr));
}
if let Ok(addr) = host.parse::<Ipv6Addr>() {
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 = <str as sealed::ToSocketAddrsPriv>::Iter;
type Future = <str as sealed::ToSocketAddrsPriv>::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<SocketAddr>),
Blocking(JoinHandle<io::Result<vec::IntoIter<SocketAddr>>>),
}
cfg_dns! {
#[doc(hidden)]
#[derive(Debug)]
pub enum MaybeReady {
Ready(Option<SocketAddr>),
Blocking(JoinHandle<io::Result<vec::IntoIter<SocketAddr>>>),
}
#[doc(hidden)]
#[derive(Debug)]
#[cfg(feature = "dns")]
pub enum OneOrMore {
One(option::IntoIter<SocketAddr>),
More(vec::IntoIter<SocketAddr>),
}
#[doc(hidden)]
#[derive(Debug)]
pub enum OneOrMore {
One(option::IntoIter<SocketAddr>),
More(vec::IntoIter<SocketAddr>),
}
#[cfg(feature = "dns")]
impl Future for MaybeReady {
type Output = io::Result<OneOrMore>;
impl Future for MaybeReady {
type Output = io::Result<OneOrMore>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
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<Self::Output> {
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<Self::Item> {
match self {
OneOrMore::One(i) => i.next(),
OneOrMore::More(i) => i.next(),
fn next(&mut self) -> Option<Self::Item> {
match self {
OneOrMore::One(i) => i.next(),
OneOrMore::More(i) => i.next(),
}
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
match self {
OneOrMore::One(i) => i.size_hint(),
OneOrMore::More(i) => i.size_hint(),
fn size_hint(&self) -> (usize, Option<usize>) {
match self {
OneOrMore::One(i) => i.size_hint(),
OneOrMore::More(i) => i.size_hint(),
}
}
}
}

34
tokio/src/net/mod.rs
View File

@ -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};
}

8
tokio/src/prelude.rs
View File

@ -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 _};
}

4
tokio/src/process/mod.rs
View File

@ -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<A: AsyncRead + Unpin>(io: Option<A>) -> io::Result<Vec<u8>> {
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)
}

8
tokio/src/runtime/blocking.rs
View File

@ -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)]

218
tokio/src/runtime/builder.rs
View File

@ -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<Runtime> {
use crate::runtime::{BasicScheduler, Kind};
cfg_rt_core! {
impl Builder {
fn build_basic_runtime(&mut self) -> io::Result<Runtime> {
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<Runtime> {
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<dyn Fn(usize, &mut dyn FnMut()) + Send + Sync>);
}
}
}
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<Runtime> {
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<dyn Fn(usize, &mut dyn FnMut()) + Send + Sync>);
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,
})
}
}
}

96
tokio/src/runtime/enter.rs
View File

@ -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<bool> = Cell::new(false));
@ -48,58 +46,62 @@ pub(crate) fn try_enter() -> Option<Enter> {
//
// 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<F: FnOnce() -> 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<F: FnOnce() -> 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<F>(&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<F: Future>(&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();
}
}
}

22
tokio/src/runtime/global.rs
View File

@ -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<F, R>(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<F, R>(thread_pool: &thread_pool::Spawner, f: F) -> R
where
F: FnOnce() -> R,
{
with_state(State::ThreadPool(thread_pool as *const _), f)
}
}
fn with_state<F, R>(state: State, f: F) -> R

117
tokio/src/runtime/handle.rs
View File

@ -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<F>(&self, future: F) -> JoinHandle<F::Output>
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<F, R>(&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<F>(&self, future: F) -> JoinHandle<F::Output>
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),
}
}
}
}

8
tokio/src/runtime/io.rs
View File

@ -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;

39
tokio/src/runtime/mod.rs
View File

@ -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<time::Driver>),
/// 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<Self> {
#[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),
})
}

2
tokio/src/runtime/park/mod.rs
View File

@ -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;

4
tokio/src/runtime/park/thread.rs
View File

@ -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;

4
tokio/src/runtime/thread_pool/mod.rs
View File

@ -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)]

34
tokio/src/runtime/thread_pool/worker.rs
View File

@ -15,24 +15,26 @@ thread_local! {
static ON_BLOCK: Cell<Option<*const dyn Fn()>> = Cell::new(None)
}
pub(crate) fn block_in_place<F, R>(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, R>(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<P: Park + 'static> {

8
tokio/src/runtime/time.rs
View File

@ -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 = ();

67
tokio/src/sync/mod.rs
View File

@ -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;

11
tokio/src/sync/mpsc/bounded.rs
View File

@ -161,12 +161,13 @@ impl<T> Receiver<T> {
impl<T> Unpin for Receiver<T> {}
#[cfg(feature = "stream")]
impl<T> futures_core::Stream for Receiver<T> {
type Item = T;
cfg_stream! {
impl<T> futures_core::Stream for Receiver<T> {
type Item = T;
fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.poll_recv(cx)
fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.poll_recv(cx)
}
}
}

6
tokio/src/sync/mpsc/chan.rs
View File

@ -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 {

7
tokio/src/sync/mpsc/list.rs
View File

@ -169,7 +169,6 @@ impl<T> Tx<T> {
}
pub(crate) unsafe fn reclaim_block(&self, mut block: NonNull<Block<T>>) {
debug!("+ reclaim_block({:p})", block);
// The block has been removed from the linked list and ownership
// is reclaimed.
//
@ -206,7 +205,6 @@ impl<T> Tx<T> {
}
if !reused {
debug!(" + block freed {:p}", block);
let _ = Box::from_raw(block.as_ptr());
}
}
@ -226,7 +224,6 @@ impl<T> Rx<T> {
pub(crate) fn pop(&mut self, tx: &Tx<T>) -> Option<block::Read<T>> {
// Advance `head`, if needed
if !self.try_advancing_head() {
debug!("+ !self.try_advancing_head() -> false");
return None;
}
@ -276,8 +273,6 @@ impl<T> Rx<T> {
}
fn reclaim_blocks(&mut self, tx: &Tx<T>) {
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<T> Rx<T> {
/// 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<T> Rx<T> {
while let Some(block) = cur {
cur = block.as_ref().load_next(Relaxed);
debug!(" + free: block = {:p}", block);
drop(Box::from_raw(block.as_ptr()));
}
}

2
tokio/src/sync/mpsc/mod.rs
View File

@ -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.
//!

2
tokio/src/sync/oneshot.rs
View File

@ -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;

108
tokio/src/sync/semaphore.rs
View File

@ -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<Arc<WaiterNode>>,
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<Arc<WaiterNode>> {
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)
}

8
tokio/src/sync/task/atomic_waker.rs
View File

@ -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<Waker> {
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.
//

4
tokio/src/sync/tests/loom_list.rs
View File

@ -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();
}
}

27
tokio/src/sync/tests/mod.rs
View File

@ -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;
}

4
tokio/tests/sync_semaphore.rs → tokio/src/sync/tests/semaphore.rs
View File

@ -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};

113
tokio/src/task/blocking.rs
View File

@ -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, R>(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, R>(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<dyn std::error::Error>>{
/// 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, R>(f: F) -> JoinHandle<R>
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<dyn std::error::Error>>{
/// 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, R>(f: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
blocking::spawn_blocking(f)
}
}

52
tokio/src/task/mod.rs
View File

@ -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<Self>);
}
/// Create a new task without an associated join handle
#[cfg(feature = "rt-full")]
pub(crate) fn background<T, S>(task: T) -> Task<S>
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<T, S>(task: T) -> Task<S>
where
T: Future + Send + 'static,
S: Schedule,
{
Task {
raw: RawTask::new_background::<_, S>(task),
_p: PhantomData,
}
}
}

19
tokio/src/task/raw.rs
View File

@ -54,16 +54,19 @@ pub(super) fn vtable<T: Future, S: Schedule>() -> &'static Vtable {
}
}
impl RawTask {
#[cfg(feature = "rt-full")]
pub(super) fn new_background<T, S>(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<T, S>(task: T) -> RawTask
where
T: Future + Send + 'static,
S: Schedule,
{
RawTask::new::<_, S>(task, State::new_background())
}
}
}
impl RawTask {
pub(super) fn new_joinable<T, S>(task: T) -> RawTask
where
T: Future + Send + 'static,

2
tokio/src/task/state.rs
View File

@ -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),

22
tokio/src/tests/mod.rs
View File

@ -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;

13
tokio/src/time/clock.rs
View File

@ -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;

10
tokio/tests/fs_file_mocked.rs
View File

@ -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