rt: Remove threaded_scheduler() and basic_scheduler() (#2876)

Co-authored-by: Alice Ryhl <[email protected]>
Co-authored-by: Carl Lerche <[email protected]>
This commit is contained in:
Lucio Franco
2020-10-12 13:44:54 -04:00
committed by GitHub
co-authored by Alice Ryhl Carl Lerche
parent 0893841f31
commit 8880222036
90 changed files with 1480 additions and 1698 deletions
+12 -25
View File
@@ -4,6 +4,13 @@ use tokio::sync::mpsc;
type Medium = [usize; 64];
type Large = [Medium; 64];
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap()
}
fn create_1_medium(b: &mut Bencher) {
b.iter(|| {
black_box(&mpsc::channel::<Medium>(1));
@@ -43,11 +50,7 @@ fn send_large(b: &mut Bencher) {
}
fn contention_bounded(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
.build()
.unwrap();
let rt = rt();
b.iter(|| {
rt.block_on(async move {
@@ -70,11 +73,7 @@ fn contention_bounded(b: &mut Bencher) {
}
fn contention_bounded_full(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
.build()
.unwrap();
let rt = rt();
b.iter(|| {
rt.block_on(async move {
@@ -97,11 +96,7 @@ fn contention_bounded_full(b: &mut Bencher) {
}
fn contention_unbounded(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
.build()
.unwrap();
let rt = rt();
b.iter(|| {
rt.block_on(async move {
@@ -124,11 +119,7 @@ fn contention_unbounded(b: &mut Bencher) {
}
fn uncontented_bounded(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
.build()
.unwrap();
let rt = rt();
b.iter(|| {
rt.block_on(async move {
@@ -146,11 +137,7 @@ fn uncontented_bounded(b: &mut Bencher) {
}
fn uncontented_unbounded(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
.build()
.unwrap();
let rt = rt();
b.iter(|| {
rt.block_on(async move {
+2 -3
View File
@@ -139,9 +139,8 @@ fn chained_spawn(b: &mut Bencher) {
}
fn rt() -> Runtime {
runtime::Builder::new()
.threaded_scheduler()
.core_threads(4)
runtime::Builder::new_multi_thread()
.worker_threads(4)
.enable_all()
.build()
.unwrap()
+2 -2
View File
@@ -45,9 +45,9 @@ fn many_signals(bench: &mut Bencher) {
let num_signals = 10;
let (tx, mut rx) = mpsc::channel(num_signals);
let rt = runtime::Builder::new()
let rt = runtime::Builder::new_multi_thread()
// Intentionally single threaded to measure delays in propagating wakes
.basic_scheduler()
.worker_threads(0)
.enable_all()
.build()
.unwrap();
+4 -10
View File
@@ -10,8 +10,7 @@ async fn work() -> usize {
}
fn basic_scheduler_local_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new()
.basic_scheduler()
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
runtime.block_on(async {
@@ -23,8 +22,7 @@ fn basic_scheduler_local_spawn(bench: &mut Bencher) {
}
fn threaded_scheduler_local_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new()
.threaded_scheduler()
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
runtime.block_on(async {
@@ -36,8 +34,7 @@ fn threaded_scheduler_local_spawn(bench: &mut Bencher) {
}
fn basic_scheduler_remote_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new()
.basic_scheduler()
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
@@ -48,10 +45,7 @@ fn basic_scheduler_remote_spawn(bench: &mut Bencher) {
}
fn threaded_scheduler_remote_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
bench.iter(|| {
let h = runtime.spawn(work());
+8 -13
View File
@@ -3,9 +3,8 @@ use std::sync::Arc;
use tokio::{sync::RwLock, task};
fn read_uncontended(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap();
@@ -22,9 +21,8 @@ fn read_uncontended(b: &mut Bencher) {
}
fn read_concurrent_uncontended_multi(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap();
@@ -51,8 +49,7 @@ fn read_concurrent_uncontended_multi(b: &mut Bencher) {
}
fn read_concurrent_uncontended(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.basic_scheduler()
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
@@ -78,9 +75,8 @@ fn read_concurrent_uncontended(b: &mut Bencher) {
}
fn read_concurrent_contended_multi(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap();
@@ -108,8 +104,7 @@ fn read_concurrent_contended_multi(b: &mut Bencher) {
}
fn read_concurrent_contended(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.basic_scheduler()
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
+9 -13
View File
@@ -3,9 +3,8 @@ use std::sync::Arc;
use tokio::{sync::Semaphore, task};
fn uncontended(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap();
@@ -27,9 +26,8 @@ async fn task(s: Arc<Semaphore>) {
}
fn uncontended_concurrent_multi(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap();
@@ -51,8 +49,8 @@ fn uncontended_concurrent_multi(b: &mut Bencher) {
}
fn uncontended_concurrent_single(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.basic_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(0)
.build()
.unwrap();
@@ -73,9 +71,8 @@ fn uncontended_concurrent_single(b: &mut Bencher) {
}
fn contended_concurrent_multi(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.core_threads(6)
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(6)
.build()
.unwrap();
@@ -97,8 +94,7 @@ fn contended_concurrent_multi(b: &mut Bencher) {
}
fn contended_concurrent_single(b: &mut Bencher) {
let rt = tokio::runtime::Builder::new()
.basic_scheduler()
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
+1
View File
@@ -7,6 +7,7 @@ publish = false
[features]
full = ["tokio/full"]
rt-core = ["tokio/rt-core", "tokio/macros"]
[dependencies]
tokio = { path = "../tokio", optional = true }
@@ -0,0 +1,6 @@
use tests_build::tokio;
#[tokio::main]
async fn my_fn() {}
fn main() {}
@@ -0,0 +1,7 @@
error: The default runtime flavor is `multi_thread`, but the `rt-threaded` feature is disabled.
--> $DIR/macros_core_no_default.rs:3:1
|
3 | #[tokio::main]
| ^^^^^^^^^^^^^^
|
= note: this error originates in an attribute macro (in Nightly builds, run with -Z macro-backtrace for more info)
@@ -4,7 +4,7 @@ error: the async keyword is missing from the function declaration
4 | fn main_is_not_async() {}
| ^^
error: Unknown attribute foo is specified; expected `basic_scheduler` or `threaded_scheduler`
error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`
--> $DIR/macros_invalid_input.rs:6:15
|
6 | #[tokio::main(foo)]
@@ -28,7 +28,7 @@ error: the test function cannot accept arguments
16 | async fn test_fn_has_args(_x: u8) {}
| ^^^^^^
error: Unknown attribute foo is specified; expected `basic_scheduler` or `threaded_scheduler`
error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`
--> $DIR/macros_invalid_input.rs:18:15
|
18 | #[tokio::test(foo)]
+4 -1
View File
@@ -1,9 +1,12 @@
#[test]
fn compile_fail() {
fn compile_fail_full() {
let t = trybuild::TestCases::new();
#[cfg(feature = "full")]
t.compile_fail("tests/fail/macros_invalid_input.rs");
#[cfg(all(feature = "rt-core", not(feature = "full")))]
t.compile_fail("tests/fail/macros_core_no_default.rs");
drop(t);
}
+9 -12
View File
@@ -1,4 +1,4 @@
#![cfg(feature = "macros")]
#![cfg(all(feature = "macros", feature = "rt-core"))]
#[tokio::main]
async fn basic_main() -> usize {
@@ -10,18 +10,15 @@ async fn generic_fun<T: Default>() -> T {
T::default()
}
#[cfg(feature = "rt-core")]
mod spawn {
#[tokio::main]
async fn spawning() -> usize {
let join = tokio::spawn(async { 1 });
join.await.unwrap()
}
#[tokio::main]
async fn spawning() -> usize {
let join = tokio::spawn(async { 1 });
join.await.unwrap()
}
#[test]
fn main_with_spawn() {
assert_eq!(1, spawning());
}
#[test]
fn main_with_spawn() {
assert_eq!(1, spawning());
}
#[test]
-32
View File
@@ -1,32 +0,0 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "sync")]
use tokio::runtime;
use tokio::sync::oneshot;
use std::sync::mpsc;
use std::thread;
#[test]
fn basic_shell_rt() {
let (feed_tx, feed_rx) = mpsc::channel::<oneshot::Sender<()>>();
let th = thread::spawn(move || {
for tx in feed_rx.iter() {
tx.send(()).unwrap();
}
});
for _ in 0..1_000 {
let rt = runtime::Builder::new().build().unwrap();
let (tx, rx) = oneshot::channel();
feed_tx.send(tx).unwrap();
rt.block_on(rx).unwrap();
}
drop(feed_tx);
th.join().unwrap();
}
+173 -238
View File
@@ -1,13 +1,139 @@
use proc_macro::TokenStream;
use proc_macro2::Span;
use quote::quote;
use std::num::NonZeroUsize;
use syn::spanned::Spanned;
#[derive(Clone, Copy, PartialEq)]
enum Runtime {
Basic,
enum RuntimeFlavor {
CurrentThread,
Threaded,
}
impl RuntimeFlavor {
fn from_str(s: &str) -> Result<RuntimeFlavor, String> {
match s {
"current_thread" => Ok(RuntimeFlavor::CurrentThread),
"multi_thread" => Ok(RuntimeFlavor::Threaded),
"single_thread" => Err("The single threaded runtime flavor is called `current_thread`.".to_string()),
"basic_scheduler" => Err("The `basic_scheduler` runtime flavor has been renamed to `current_thread`.".to_string()),
"threaded_scheduler" => Err("The `threaded_scheduler` runtime flavor has been renamed to `multi_thread`.".to_string()),
_ => Err(format!("No such runtime flavor `{}`. The runtime flavors are `current_thread` and `multi_thread`.", s)),
}
}
}
struct FinalConfig {
flavor: RuntimeFlavor,
worker_threads: Option<usize>,
}
struct Configuration {
rt_threaded_available: bool,
default_flavor: RuntimeFlavor,
flavor: Option<RuntimeFlavor>,
worker_threads: Option<(usize, Span)>,
}
impl Configuration {
fn new(is_test: bool, rt_threaded: bool) -> Self {
Configuration {
rt_threaded_available: rt_threaded,
default_flavor: match is_test {
true => RuntimeFlavor::CurrentThread,
false => RuntimeFlavor::Threaded,
},
flavor: None,
worker_threads: None,
}
}
fn set_flavor(&mut self, runtime: syn::Lit, span: Span) -> Result<(), syn::Error> {
if self.flavor.is_some() {
return Err(syn::Error::new(span, "`flavor` set multiple times."));
}
let runtime_str = parse_string(runtime, span, "flavor")?;
let runtime =
RuntimeFlavor::from_str(&runtime_str).map_err(|err| syn::Error::new(span, err))?;
self.flavor = Some(runtime);
Ok(())
}
fn set_worker_threads(
&mut self,
worker_threads: syn::Lit,
span: Span,
) -> Result<(), syn::Error> {
if self.worker_threads.is_some() {
return Err(syn::Error::new(
span,
"`worker_threads` set multiple times.",
));
}
let worker_threads = parse_int(worker_threads, span, "worker_threads")?;
if worker_threads == 0 {
return Err(syn::Error::new(span, "`worker_threads` may not be 0."));
}
self.worker_threads = Some((worker_threads, span));
Ok(())
}
fn build(&self) -> Result<FinalConfig, syn::Error> {
let flavor = self.flavor.unwrap_or(self.default_flavor);
use RuntimeFlavor::*;
match (flavor, self.worker_threads) {
(CurrentThread, Some((_, worker_threads_span))) => Err(syn::Error::new(
worker_threads_span,
"The `worker_threads` option requires the `multi_thread` runtime flavor.",
)),
(CurrentThread, None) => Ok(FinalConfig {
flavor,
worker_threads: None,
}),
(Threaded, worker_threads) if self.rt_threaded_available => Ok(FinalConfig {
flavor,
worker_threads: worker_threads.map(|(val, _span)| val),
}),
(Threaded, _) => {
let msg = if self.flavor.is_none() {
"The default runtime flavor is `multi_thread`, but the `rt-threaded` feature is disabled."
} else {
"The runtime flavor `multi_thread` requires the `rt-threaded` feature."
};
Err(syn::Error::new(Span::call_site(), msg))
}
}
}
}
fn parse_int(int: syn::Lit, span: Span, field: &str) -> Result<usize, syn::Error> {
match int {
syn::Lit::Int(lit) => match lit.base10_parse::<usize>() {
Ok(value) => Ok(value),
Err(e) => Err(syn::Error::new(
span,
format!("Failed to parse {} as integer: {}", field, e),
)),
},
_ => Err(syn::Error::new(
span,
format!("Failed to parse {} as integer.", field),
)),
}
}
fn parse_string(int: syn::Lit, span: Span, field: &str) -> Result<String, syn::Error> {
match int {
syn::Lit::Str(s) => Ok(s.value()),
syn::Lit::Verbatim(s) => Ok(s.to_string()),
_ => Err(syn::Error::new(
span,
format!("Failed to parse {} as string.", field),
)),
}
}
fn parse_knobs(
mut input: syn::ItemFn,
args: syn::AttributeArgs,
@@ -26,9 +152,12 @@ fn parse_knobs(
sig.asyncness = None;
let mut runtime = None;
let mut core_threads = None;
let mut max_threads = None;
let macro_name = if is_test {
"tokio::test"
} else {
"tokio::main"
};
let mut config = Configuration::new(is_test, rt_threaded);
for arg in args {
match arg {
@@ -39,65 +168,18 @@ fn parse_knobs(
return Err(syn::Error::new_spanned(namevalue, msg));
}
match ident.unwrap().to_string().to_lowercase().as_str() {
"core_threads" => {
if rt_threaded {
match &namevalue.lit {
syn::Lit::Int(expr) => {
let num = expr.base10_parse::<NonZeroUsize>().unwrap();
if num.get() > 1 {
runtime = Some(Runtime::Threaded);
} else {
runtime = Some(Runtime::Basic);
}
if let Some(v) = max_threads {
if v < num {
return Err(syn::Error::new_spanned(
namevalue,
"max_threads cannot be less than core_threads",
));
}
}
core_threads = Some(num);
}
_ => {
return Err(syn::Error::new_spanned(
namevalue,
"core_threads argument must be an int",
))
}
}
} else {
return Err(syn::Error::new_spanned(
namevalue,
"core_threads can only be set with rt-threaded feature flag enabled",
));
}
"worker_threads" => {
config.set_worker_threads(namevalue.lit.clone(), namevalue.span())?;
}
"flavor" => {
config.set_flavor(namevalue.lit.clone(), namevalue.span())?;
}
"core_threads" => {
let msg = "Attribute `core_threads` is renamed to `worker_threads`";
return Err(syn::Error::new_spanned(namevalue, msg));
}
"max_threads" => match &namevalue.lit {
syn::Lit::Int(expr) => {
let num = expr.base10_parse::<NonZeroUsize>().unwrap();
if let Some(v) = core_threads {
if num < v {
return Err(syn::Error::new_spanned(
namevalue,
"max_threads cannot be less than core_threads",
));
}
}
max_threads = Some(num);
}
_ => {
return Err(syn::Error::new_spanned(
namevalue,
"max_threads argument must be an int",
))
}
},
name => {
let msg = format!("Unknown attribute pair {} is specified; expected one of: `core_threads`, `max_threads`", name);
let msg = format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `max_threads`", name);
return Err(syn::Error::new_spanned(namevalue, msg));
}
}
@@ -108,16 +190,22 @@ fn parse_knobs(
let msg = "Must have specified ident";
return Err(syn::Error::new_spanned(path, msg));
}
match ident.unwrap().to_string().to_lowercase().as_str() {
"threaded_scheduler" => {
runtime = Some(runtime.unwrap_or_else(|| Runtime::Threaded))
}
"basic_scheduler" => runtime = Some(runtime.unwrap_or_else(|| Runtime::Basic)),
let name = ident.unwrap().to_string().to_lowercase();
let msg = match name.as_str() {
"threaded_scheduler" | "multi_thread" => {
format!("Set the runtime flavor with #[{}(flavor = \"multi_thread\")].", macro_name)
},
"basic_scheduler" | "current_thread" | "single_threaded" => {
format!("Set the runtime flavor with #[{}(flavor = \"current_thread\")].", macro_name)
},
"flavor" | "worker_threads" => {
format!("The `{}` attribute requires an argument.", name)
},
name => {
let msg = format!("Unknown attribute {} is specified; expected `basic_scheduler` or `threaded_scheduler`", name);
return Err(syn::Error::new_spanned(path, msg));
}
}
format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`", name)
},
};
return Err(syn::Error::new_spanned(path, msg));
}
other => {
return Err(syn::Error::new_spanned(
@@ -128,15 +216,18 @@ fn parse_knobs(
}
}
let mut rt = quote! { tokio::runtime::Builder::new().basic_scheduler() };
if rt_threaded && (runtime == Some(Runtime::Threaded) || (runtime.is_none() && !is_test)) {
rt = quote! { #rt.threaded_scheduler() };
}
if let Some(v) = core_threads.map(|v| v.get()) {
rt = quote! { #rt.core_threads(#v) };
}
if let Some(v) = max_threads.map(|v| v.get()) {
rt = quote! { #rt.max_threads(#v) };
let config = config.build()?;
let mut rt = match config.flavor {
RuntimeFlavor::CurrentThread => quote! {
tokio::runtime::Builder::new_current_thread()
},
RuntimeFlavor::Threaded => quote! {
tokio::runtime::Builder::new_multi_thread()
},
};
if let Some(v) = config.worker_threads {
rt = quote! { #rt.worker_threads(#v) };
}
let header = {
@@ -171,7 +262,7 @@ pub(crate) fn main(args: TokenStream, item: TokenStream, rt_threaded: bool) -> T
if input.sig.ident == "main" && !input.sig.inputs.is_empty() {
let msg = "the main function cannot accept arguments";
return syn::Error::new_spanned(&input.sig.inputs, msg)
return syn::Error::new_spanned(&input.sig.ident, msg)
.to_compile_error()
.into();
}
@@ -201,159 +292,3 @@ pub(crate) fn test(args: TokenStream, item: TokenStream, rt_threaded: bool) -> T
parse_knobs(input, args, true, rt_threaded).unwrap_or_else(|e| e.to_compile_error().into())
}
pub(crate) mod old {
use proc_macro::TokenStream;
use quote::quote;
enum Runtime {
Basic,
Threaded,
Auto,
}
#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub(crate) fn main(args: TokenStream, item: TokenStream) -> TokenStream {
let mut input = syn::parse_macro_input!(item as syn::ItemFn);
let args = syn::parse_macro_input!(args as syn::AttributeArgs);
let sig = &mut input.sig;
let name = &sig.ident;
let inputs = &sig.inputs;
let body = &input.block;
let attrs = &input.attrs;
let vis = input.vis;
if sig.asyncness.is_none() {
let msg = "the async keyword is missing from the function declaration";
return syn::Error::new_spanned(sig.fn_token, msg)
.to_compile_error()
.into();
} else if name == "main" && !inputs.is_empty() {
let msg = "the main function cannot accept arguments";
return syn::Error::new_spanned(&sig.inputs, msg)
.to_compile_error()
.into();
}
sig.asyncness = None;
let mut runtime = Runtime::Auto;
for arg in args {
if let syn::NestedMeta::Meta(syn::Meta::Path(path)) = arg {
let ident = path.get_ident();
if ident.is_none() {
let msg = "Must have specified ident";
return syn::Error::new_spanned(path, msg).to_compile_error().into();
}
match ident.unwrap().to_string().to_lowercase().as_str() {
"threaded_scheduler" => runtime = Runtime::Threaded,
"basic_scheduler" => runtime = Runtime::Basic,
name => {
let msg = format!("Unknown attribute {} is specified; expected `basic_scheduler` or `threaded_scheduler`", name);
return syn::Error::new_spanned(path, msg).to_compile_error().into();
}
}
}
}
let result = match runtime {
Runtime::Threaded | Runtime::Auto => quote! {
#(#attrs)*
#vis #sig {
tokio::runtime::Runtime::new().unwrap().block_on(async { #body })
}
},
Runtime::Basic => quote! {
#(#attrs)*
#vis #sig {
tokio::runtime::Builder::new()
.basic_scheduler()
.enable_all()
.build()
.unwrap()
.block_on(async { #body })
}
},
};
result.into()
}
pub(crate) fn test(args: TokenStream, item: TokenStream) -> TokenStream {
let input = syn::parse_macro_input!(item as syn::ItemFn);
let args = syn::parse_macro_input!(args as syn::AttributeArgs);
let ret = &input.sig.output;
let name = &input.sig.ident;
let body = &input.block;
let attrs = &input.attrs;
let vis = input.vis;
for attr in attrs {
if attr.path.is_ident("test") {
let msg = "second test attribute is supplied";
return syn::Error::new_spanned(&attr, msg)
.to_compile_error()
.into();
}
}
if input.sig.asyncness.is_none() {
let msg = "the async keyword is missing from the function declaration";
return syn::Error::new_spanned(&input.sig.fn_token, msg)
.to_compile_error()
.into();
} else if !input.sig.inputs.is_empty() {
let msg = "the test function cannot accept arguments";
return syn::Error::new_spanned(&input.sig.inputs, msg)
.to_compile_error()
.into();
}
let mut runtime = Runtime::Auto;
for arg in args {
if let syn::NestedMeta::Meta(syn::Meta::Path(path)) = arg {
let ident = path.get_ident();
if ident.is_none() {
let msg = "Must have specified ident";
return syn::Error::new_spanned(path, msg).to_compile_error().into();
}
match ident.unwrap().to_string().to_lowercase().as_str() {
"threaded_scheduler" => runtime = Runtime::Threaded,
"basic_scheduler" => runtime = Runtime::Basic,
name => {
let msg = format!("Unknown attribute {} is specified; expected `basic_scheduler` or `threaded_scheduler`", name);
return syn::Error::new_spanned(path, msg).to_compile_error().into();
}
}
}
}
let result = match runtime {
Runtime::Threaded => quote! {
#[::core::prelude::v1::test]
#(#attrs)*
#vis fn #name() #ret {
tokio::runtime::Runtime::new().unwrap().block_on(async { #body })
}
},
Runtime::Basic | Runtime::Auto => quote! {
#[::core::prelude::v1::test]
#(#attrs)*
#vis fn #name() #ret {
tokio::runtime::Builder::new()
.basic_scheduler()
.enable_all()
.build()
.unwrap()
.block_on(async { #body })
}
},
};
result.into()
}
}
+77 -163
View File
@@ -24,23 +24,38 @@ mod select;
use proc_macro::TokenStream;
/// Marks async function to be executed by selected runtime. This macro helps set up a `Runtime`
/// without requiring the user to use [Runtime](../tokio/runtime/struct.Runtime.html) or
/// Marks async function to be executed by the selected runtime. This macro helps
/// set up a `Runtime` without requiring the user to use
/// [Runtime](../tokio/runtime/struct.Runtime.html) or
/// [Builder](../tokio/runtime/struct.Builder.html) directly.
///
/// Note: This macro is designed to be simplistic and targets applications that do not require
/// a complex setup. If provided functionality is not sufficient, user may be interested in
/// using [Builder](../tokio/runtime/struct.Builder.html), which provides a more powerful
/// interface.
/// Note: This macro is designed to be simplistic and targets applications that
/// do not require a complex setup. If the provided functionality is not
/// sufficient, you may be interested in using
/// [Builder](../tokio/runtime/struct.Builder.html), which provides a more
/// powerful interface.
///
/// ## Options:
/// # Multi-threaded runtime
///
/// If you want to set the number of worker threads used for asynchronous code, use the
/// `core_threads` option.
/// To use the multi-threaded runtime, the macro can be configured using
///
/// - `core_threads=n` - Sets core threads to `n` (requires `rt-threaded` feature).
/// - `max_threads=n` - Sets max threads to `n` (requires `rt-core` or `rt-threaded` feature).
/// - `basic_scheduler` - Use the basic schduler (requires `rt-core`).
/// ```
/// #[tokio::main(flavor = "multi_thread", worker_threads = 10)]
/// # async fn main() {}
/// ```
///
/// The `worker_threads` option configures the number of worker threads, and
/// defaults to the number of cpus on the system. This is the default flavor.
///
/// # Current thread runtime
///
/// To use the single-threaded runtime known as the `current_thread` runtime,
/// the macro can be configured using
///
/// ```
/// #[tokio::main(flavor = "current_thread")]
/// # async fn main() {}
/// ```
///
/// ## Function arguments:
///
@@ -48,7 +63,7 @@ use proc_macro::TokenStream;
///
/// ## Usage
///
/// ### Using default
/// ### Using the multi-thread runtime
///
/// ```rust
/// #[tokio::main]
@@ -61,8 +76,7 @@ use proc_macro::TokenStream;
///
/// ```rust
/// fn main() {
/// tokio::runtime::Builder::new()
/// .threaded_scheduler()
/// tokio::runtime::Builder::new_multi_thread()
/// .enable_all()
/// .build()
/// .unwrap()
@@ -72,12 +86,12 @@ use proc_macro::TokenStream;
/// }
/// ```
///
/// ### Using basic scheduler
/// ### Using current thread runtime
///
/// The basic scheduler is single-threaded.
///
/// ```rust
/// #[tokio::main(basic_scheduler)]
/// #[tokio::main(flavor = "current_thread")]
/// async fn main() {
/// println!("Hello world");
/// }
@@ -87,8 +101,7 @@ use proc_macro::TokenStream;
///
/// ```rust
/// fn main() {
/// tokio::runtime::Builder::new()
/// .basic_scheduler()
/// tokio::runtime::Builder::new_current_thread()
/// .enable_all()
/// .build()
/// .unwrap()
@@ -98,10 +111,10 @@ use proc_macro::TokenStream;
/// }
/// ```
///
/// ### Set number of core threads
/// ### Set number of worker threads
///
/// ```rust
/// #[tokio::main(core_threads = 2)]
/// #[tokio::main(worker_threads = 2)]
/// async fn main() {
/// println!("Hello world");
/// }
@@ -111,9 +124,8 @@ use proc_macro::TokenStream;
///
/// ```rust
/// fn main() {
/// tokio::runtime::Builder::new()
/// .threaded_scheduler()
/// .core_threads(2)
/// tokio::runtime::Builder::new_multi_thread()
/// .worker_threads(2)
/// .enable_all()
/// .build()
/// .unwrap()
@@ -125,103 +137,20 @@ use proc_macro::TokenStream;
///
/// ### NOTE:
///
/// If you rename the tokio crate in your dependencies this macro
/// will not work. If you must rename the 0.2 version of tokio because
/// you're also using the 0.1 version of tokio, you _must_ make the
/// tokio 0.2 crate available as `tokio` in the module where this
/// macro is expanded.
#[proc_macro_attribute]
#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub fn main_threaded(args: TokenStream, item: TokenStream) -> TokenStream {
entry::main(args, item, true)
}
/// Marks async function to be executed by selected runtime. This macro helps set up a `Runtime`
/// without requiring the user to use [Runtime](../tokio/runtime/struct.Runtime.html) or
/// [Builder](../tokio/runtime/struct.Builder.html) directly.
///
/// Note: This macro is designed to be simplistic and targets applications that do not require
/// a complex setup. If provided functionality is not sufficient, user may be interested in
/// using [Builder](../tokio/runtime/struct.Builder.html), which provides a more powerful
/// interface.
///
/// ## Options:
///
/// - `basic_scheduler` - All tasks are executed on the current thread.
/// - `threaded_scheduler` - Uses the multi-threaded scheduler. Used by default (requires `rt-threaded` feature).
///
/// ## Function arguments:
///
/// Arguments are allowed for any functions aside from `main` which is special
///
/// ## Usage
///
/// ### Using default
///
/// ```rust
/// #[tokio::main]
/// async fn main() {
/// println!("Hello world");
/// }
/// ```
///
/// Equivalent code not using `#[tokio::main]`
///
/// ```rust
/// fn main() {
/// tokio::runtime::Runtime::new()
/// .unwrap()
/// .block_on(async {
/// println!("Hello world");
/// })
/// }
/// ```
///
/// ### Select runtime
///
/// ```rust
/// #[tokio::main(basic_scheduler)]
/// async fn main() {
/// println!("Hello world");
/// }
/// ```
///
/// Equivalent code not using `#[tokio::main]`
///
/// ```rust
/// fn main() {
/// tokio::runtime::Builder::new()
/// .basic_scheduler()
/// .enable_all()
/// .build()
/// .unwrap()
/// .block_on(async {
/// println!("Hello world");
/// })
/// }
/// ```
///
/// ### NOTE:
///
/// If you rename the tokio crate in your dependencies this macro
/// will not work. If you must rename the 0.2 version of tokio because
/// you're also using the 0.1 version of tokio, you _must_ make the
/// tokio 0.2 crate available as `tokio` in the module where this
/// macro is expanded.
/// If you rename the tokio crate in your dependencies this macro will not work.
/// If you must rename the 0.2 version of tokio because you're also using the
/// 0.1 version of tokio, you _must_ make the tokio 0.2 crate available as
/// `tokio` in the module where this macro is expanded.
#[proc_macro_attribute]
#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
entry::old::main(args, item)
entry::main(args, item, true)
}
/// Marks async function to be executed by selected runtime. This macro helps set up a `Runtime`
/// without requiring the user to use [Runtime](../tokio/runtime/struct.Runtime.html) or
/// [Builder](../tokio/runtime/struct.builder.html) directly.
///
/// ## Options:
///
/// - `max_threads=n` - Sets max threads to `n`.
///
/// ## Function arguments:
///
/// Arguments are allowed for any functions aside from `main` which is special
@@ -231,7 +160,7 @@ pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
/// ### Using default
///
/// ```rust
/// #[tokio::main]
/// #[tokio::main(flavor = "current_thread")]
/// async fn main() {
/// println!("Hello world");
/// }
@@ -241,8 +170,7 @@ pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
///
/// ```rust
/// fn main() {
/// tokio::runtime::Builder::new()
/// .basic_scheduler()
/// tokio::runtime::Builder::new_current_thread()
/// .enable_all()
/// .build()
/// .unwrap()
@@ -261,23 +189,18 @@ pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
/// macro is expanded.
#[proc_macro_attribute]
#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub fn main_basic(args: TokenStream, item: TokenStream) -> TokenStream {
pub fn main_rt_core(args: TokenStream, item: TokenStream) -> TokenStream {
entry::main(args, item, false)
}
/// Marks async function to be executed by runtime, suitable to test environment
///
/// ## Options:
///
/// - `core_threads=n` - Sets core threads to `n` (requires `rt-threaded` feature).
/// - `max_threads=n` - Sets max threads to `n` (requires `rt-core` or `rt-threaded` feature).
///
/// ## Usage
///
/// ### Select runtime
/// ### Multi-thread runtime
///
/// ```no_run
/// #[tokio::test(core_threads = 1)]
/// #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
/// async fn my_test() {
/// assert!(true);
/// }
@@ -285,44 +208,7 @@ pub fn main_basic(args: TokenStream, item: TokenStream) -> TokenStream {
///
/// ### Using default
///
/// ```no_run
/// #[tokio::test]
/// async fn my_test() {
/// assert!(true);
/// }
/// ```
///
/// ### NOTE:
///
/// If you rename the tokio crate in your dependencies this macro
/// will not work. If you must rename the 0.2 version of tokio because
/// you're also using the 0.1 version of tokio, you _must_ make the
/// tokio 0.2 crate available as `tokio` in the module where this
/// macro is expanded.
#[proc_macro_attribute]
pub fn test_threaded(args: TokenStream, item: TokenStream) -> TokenStream {
entry::test(args, item, true)
}
/// Marks async function to be executed by runtime, suitable to test environment
///
/// ## Options:
///
/// - `basic_scheduler` - All tasks are executed on the current thread. Used by default.
/// - `threaded_scheduler` - Use multi-threaded scheduler (requires `rt-threaded` feature).
///
/// ## Usage
///
/// ### Select runtime
///
/// ```no_run
/// #[tokio::test(threaded_scheduler)]
/// async fn my_test() {
/// assert!(true);
/// }
/// ```
///
/// ### Using default
/// The default test runtime is multi-threaded.
///
/// ```no_run
/// #[tokio::test]
@@ -340,7 +226,7 @@ pub fn test_threaded(args: TokenStream, item: TokenStream) -> TokenStream {
/// macro is expanded.
#[proc_macro_attribute]
pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
entry::old::test(args, item)
entry::test(args, item, true)
}
/// Marks async function to be executed by runtime, suitable to test environment
@@ -366,10 +252,38 @@ pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
/// tokio 0.2 crate available as `tokio` in the module where this
/// macro is expanded.
#[proc_macro_attribute]
pub fn test_basic(args: TokenStream, item: TokenStream) -> TokenStream {
pub fn test_rt_core(args: TokenStream, item: TokenStream) -> TokenStream {
entry::test(args, item, false)
}
/// Always fails with the error message below.
/// ```text
/// The #[tokio::main] macro requires rt-core or rt-threaded.
/// ```
#[proc_macro_attribute]
pub fn main_fail(_args: TokenStream, _item: TokenStream) -> TokenStream {
syn::Error::new(
proc_macro2::Span::call_site(),
"The #[tokio::main] macro requires rt-core or rt-threaded.",
)
.to_compile_error()
.into()
}
/// Always fails with the error message below.
/// ```text
/// The #[tokio::test] macro requires rt-core or rt-threaded.
/// ```
#[proc_macro_attribute]
pub fn test_fail(_args: TokenStream, _item: TokenStream) -> TokenStream {
syn::Error::new(
proc_macro2::Span::call_site(),
"The #[tokio::test] macro requires rt-core or rt-threaded.",
)
.to_compile_error()
.into()
}
/// Implementation detail of the `select!` macro. This macro is **not** intended
/// to be used as part of the public API and is permitted to change.
#[proc_macro]
+1 -2
View File
@@ -28,8 +28,7 @@ pub mod task;
pub fn block_on<F: std::future::Future>(future: F) -> F::Output {
use tokio::runtime;
let rt = runtime::Builder::new()
.basic_scheduler()
let rt = runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
+1
View File
@@ -31,6 +31,7 @@ compat = ["futures-io",]
codec = ["tokio/stream"]
time = ["tokio/time","slab"]
io = []
rt-core = ["tokio/rt-core"]
[dependencies]
tokio = { version = "0.3.0", path = "../tokio" }
+10
View File
@@ -39,3 +39,13 @@ macro_rules! cfg_io {
)*
}
}
macro_rules! cfg_rt_core {
($($item:item)*) => {
$(
#[cfg(feature = "rt-core")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-core")))]
$item
)*
}
}
+1 -1
View File
@@ -33,7 +33,7 @@ use std::io;
/// # }
/// # }
/// #
/// # #[tokio::main(core_threads = 1)]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() -> Result<(), std::io::Error> {
/// let my_async_read = File::open("filename.txt").await?;
/// let my_stream_of_bytes = FramedRead::new(my_async_read, BytesCodec::new());
+6 -6
View File
@@ -48,14 +48,14 @@ pub trait RuntimeExt {
/// use tokio_util::context::RuntimeExt;
/// use tokio::time::{sleep, Duration};
///
/// let rt = tokio::runtime::Builder::new()
/// .threaded_scheduler()
/// let rt = tokio::runtime::Builder::new_multi_thread()
/// .enable_all()
/// .build().unwrap();
/// .build()
/// .unwrap();
///
/// let rt2 = tokio::runtime::Builder::new()
/// .threaded_scheduler()
/// .build().unwrap();
/// let rt2 = tokio::runtime::Builder::new_multi_thread()
/// .build()
/// .unwrap();
///
/// let fut = sleep(Duration::from_millis(2));
///
+3 -1
View File
@@ -47,7 +47,9 @@ cfg_io! {
pub mod io;
}
pub mod context;
cfg_rt_core! {
pub mod context;
}
pub mod sync;
+5 -6
View File
@@ -1,3 +1,4 @@
#![cfg(feature = "rt-core")]
#![warn(rust_2018_idioms)]
use tokio::runtime::Builder;
@@ -6,15 +7,13 @@ use tokio_util::context::RuntimeExt;
#[test]
fn tokio_context_with_another_runtime() {
let rt1 = Builder::new()
.threaded_scheduler()
.core_threads(1)
let rt1 = Builder::new_multi_thread()
.worker_threads(1)
// no timer!
.build()
.unwrap();
let rt2 = Builder::new()
.threaded_scheduler()
.core_threads(1)
let rt2 = Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.unwrap();
+3 -5
View File
@@ -30,7 +30,6 @@ default = []
# enable everything
full = [
"blocking",
"dns",
"fs",
"io-util",
@@ -47,12 +46,11 @@ full = [
"time",
]
blocking = ["rt-core"]
dns = ["rt-core"]
fs = ["rt-core", "io-util"]
dns = []
fs = []
io-util = ["memchr"]
# stdin, stdout, stderr
io-std = ["rt-core"]
io-std = []
macros = ["tokio-macros"]
net = ["dns", "tcp", "udp", "uds"]
process = [
+48
View File
@@ -0,0 +1,48 @@
cfg_rt_core! {
pub(crate) use crate::runtime::spawn_blocking;
pub(crate) use crate::task::JoinHandle;
}
cfg_not_rt_core! {
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
pub(crate) fn spawn_blocking<F, R>(_f: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
assert_send_sync::<JoinHandle<std::cell::Cell<()>>>();
panic!("requires the `rt-core` Tokio feature flag")
}
pub(crate) struct JoinHandle<R> {
_p: std::marker::PhantomData<R>,
}
unsafe impl<T: Send> Send for JoinHandle<T> {}
unsafe impl<T: Send> Sync for JoinHandle<T> {}
impl<R> Future for JoinHandle<R> {
type Output = Result<R, std::io::Error>;
fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
unreachable!()
}
}
impl<T> fmt::Debug for JoinHandle<T>
where
T: fmt::Debug,
{
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("JoinHandle").finish()
}
}
fn assert_send_sync<T: Send + Sync>() {
}
}
+8 -9
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "full"), allow(dead_code))]
//! Opt-in yield points for improved cooperative scheduling.
//!
//! A single call to [`poll`] on a top-level task may potentially do a lot of
@@ -96,14 +98,6 @@ pub(crate) fn budget<R>(f: impl FnOnce() -> R) -> R {
with_budget(Budget::initial(), f)
}
cfg_rt_threaded! {
/// Set the current task's budget
#[cfg(feature = "blocking")]
pub(crate) fn set(budget: Budget) {
CURRENT.with(|cell| cell.set(budget))
}
}
#[inline(always)]
fn with_budget<R>(budget: Budget, f: impl FnOnce() -> R) -> R {
struct ResetGuard<'a> {
@@ -129,13 +123,18 @@ fn with_budget<R>(budget: Budget, f: impl FnOnce() -> R) -> R {
}
cfg_rt_threaded! {
/// Set the current task's budget
pub(crate) fn set(budget: Budget) {
CURRENT.with(|cell| cell.set(budget))
}
#[inline(always)]
pub(crate) fn has_budget_remaining() -> bool {
CURRENT.with(|cell| cell.get().has_remaining())
}
}
cfg_blocking_impl! {
cfg_rt_core! {
/// Forcibly remove the budgeting constraints early.
///
/// Returns the remaining budget
+2 -2
View File
@@ -107,6 +107,6 @@ mod sys {
pub(crate) use std::fs::File;
// TODO: don't rename
pub(crate) use crate::runtime::spawn_blocking as run;
pub(crate) use crate::task::JoinHandle as Blocking;
pub(crate) use crate::blocking::spawn_blocking as run;
pub(crate) use crate::blocking::JoinHandle as Blocking;
}
+15
View File
@@ -0,0 +1,15 @@
use std::future::Future;
cfg_rt_core! {
pub(crate) fn block_on<F: Future>(f: F) -> F::Output {
let mut e = crate::runtime::enter::enter(false);
e.block_on(f).unwrap()
}
}
cfg_not_rt_core! {
pub(crate) fn block_on<F: Future>(f: F) -> F::Output {
let mut park = crate::park::thread::CachedParkThread::new();
park.block_on(f).unwrap()
}
}
+16 -7
View File
@@ -1,15 +1,24 @@
#![allow(unused_imports, dead_code)]
#![cfg_attr(not(feature = "macros"), allow(unreachable_pub))]
//! Asynchronous values.
mod maybe_done;
pub use maybe_done::{maybe_done, MaybeDone};
#[cfg(any(feature = "macros", feature = "process"))]
pub(crate) mod maybe_done;
mod poll_fn;
pub use poll_fn::poll_fn;
mod ready;
pub(crate) use ready::{ok, Ready};
cfg_not_loom! {
mod ready;
pub(crate) use ready::{ok, Ready};
}
mod try_join;
pub(crate) use try_join::try_join3;
cfg_process! {
mod try_join;
pub(crate) use try_join::try_join3;
}
cfg_sync! {
mod block_on;
pub(crate) use block_on::block_on;
}
+2
View File
@@ -1,3 +1,5 @@
#![allow(dead_code)]
//! Definition of the `PollFn` adapter combinator
use std::fmt;
+1 -1
View File
@@ -1,4 +1,4 @@
use crate::future::{maybe_done, MaybeDone};
use crate::future::maybe_done::{maybe_done, MaybeDone};
use pin_project_lite::pin_project;
use std::future::Future;
+30 -10
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "rt-core"), allow(dead_code))]
mod ready;
use ready::Ready;
@@ -5,7 +7,6 @@ mod scheduled_io;
pub(crate) use scheduled_io::ScheduledIo; // pub(crate) for tests
use crate::park::{Park, Unpark};
use crate::runtime::context;
use crate::util::bit;
use crate::util::slab::{self, Slab};
@@ -218,17 +219,36 @@ impl fmt::Debug for Driver {
// ===== impl Handle =====
impl Handle {
/// Returns a handle to the current reactor
///
/// # Panics
///
/// This function panics if there is no current reactor set.
pub(super) fn current() -> Self {
context::io_handle()
.expect("there is no reactor running, must be called from the context of Tokio runtime")
cfg_rt_core! {
impl Handle {
/// Returns a handle to the current reactor
///
/// # Panics
///
/// This function panics if there is no current reactor set and `rt-core` feature
/// flag is not enabled.
pub(super) fn current() -> Self {
crate::runtime::context::io_handle()
.expect("there is no reactor running, must be called from the context of Tokio runtime")
}
}
}
cfg_not_rt_core! {
impl Handle {
/// Returns a handle to the current reactor
///
/// # Panics
///
/// This function panics if there is no current reactor set, or if the `rt-core`
/// feature flag is not enabled.
pub(super) fn current() -> Self {
panic!("there is no reactor running, must be called from the context of Tokio runtime with `rt-core` enabled.")
}
}
}
impl Handle {
/// Forces a reactor blocked in a call to `turn` to wakeup, or otherwise
/// makes the next call to `turn` return immediately.
///
+2 -2
View File
@@ -250,7 +250,7 @@ cfg_io_blocking! {
/// Types in this module can be mocked out in tests.
mod sys {
// TODO: don't rename
pub(crate) use crate::runtime::spawn_blocking as run;
pub(crate) use crate::task::JoinHandle as Blocking;
pub(crate) use crate::blocking::spawn_blocking as run;
pub(crate) use crate::blocking::JoinHandle as Blocking;
}
}
+2 -4
View File
@@ -183,8 +183,7 @@ mod tests {
let fut = async move {
wr.write_all(data.as_bytes()).await.unwrap();
};
crate::runtime::Builder::new()
.basic_scheduler()
crate::runtime::Builder::new_current_thread()
.build()
.unwrap()
.block_on(fut);
@@ -200,8 +199,7 @@ mod tests {
data.extend(std::iter::repeat(0b1010_1010).take(MAX_BUF - checked_count + 1));
let mut writer = LoggingMockWriter::new();
let mut splitter = super::SplitByUtf8BoundaryIfWindows::new(&mut writer);
crate::runtime::Builder::new()
.basic_scheduler()
crate::runtime::Builder::new_current_thread()
.build()
.unwrap()
.block_on(async {
+17 -11
View File
@@ -348,8 +348,7 @@ cfg_fs! {
pub mod fs;
}
#[doc(hidden)]
pub mod future;
mod future;
pub mod io;
pub mod net;
@@ -363,7 +362,14 @@ cfg_process! {
pub mod process;
}
pub mod runtime;
#[cfg(any(feature = "dns", feature = "fs", feature = "io-std"))]
mod blocking;
cfg_rt_core! {
pub mod runtime;
}
#[cfg(all(not(feature = "rt-core"), feature = "rt-util"))]
mod runtime;
pub(crate) mod coop;
@@ -389,8 +395,8 @@ cfg_not_sync! {
mod sync;
}
pub mod task;
cfg_rt_core! {
pub mod task;
pub use task::spawn;
}
@@ -414,24 +420,24 @@ cfg_macros! {
#[cfg(not(test))] // Work around for rust-lang/rust#62127
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
pub use tokio_macros::main_threaded as main;
pub use tokio_macros::main;
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
pub use tokio_macros::test_threaded as test;
pub use tokio_macros::test;
}
cfg_not_rt_threaded! {
#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub use tokio_macros::main_basic as main;
pub use tokio_macros::test_basic as test;
pub use tokio_macros::main_rt_core as main;
pub use tokio_macros::test_rt_core as test;
}
}
// Maintains old behavior
// Always fail if rt-core is not enabled.
cfg_not_rt_core! {
#[cfg(not(test))]
pub use tokio_macros::main;
pub use tokio_macros::test;
pub use tokio_macros::main_fail as main;
pub use tokio_macros::test_fail as test;
}
}
+2
View File
@@ -1,6 +1,8 @@
//! This module abstracts over `loom` and `std::sync` depending on whether we
//! are running tests or not.
#![allow(unused)]
#[cfg(not(all(test, loom)))]
mod std;
#[cfg(not(all(test, loom)))]
+12 -78
View File
@@ -1,70 +1,10 @@
#![allow(unused_macros)]
macro_rules! cfg_resource_drivers {
($($item:item)*) => {
$(
#[cfg(any(
feature = "process",
all(unix, feature = "signal"),
all(not(loom), feature = "tcp"),
feature = "time",
all(not(loom), feature = "udp"),
all(not(loom), feature = "uds"),
))]
$item
)*
}
}
macro_rules! cfg_blocking {
($($item:item)*) => {
$(
#[cfg(feature = "blocking")]
#[cfg_attr(docsrs, doc(cfg(feature = "blocking")))]
$item
)*
}
}
/// Enables 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
)*
}
}
/// Enables blocking API internals
macro_rules! cfg_blocking_impl_or_task {
($($item:item)*) => {
$(
#[cfg(any(
feature = "blocking",
feature = "fs",
feature = "dns",
feature = "io-std",
feature = "rt-threaded",
feature = "task",
))]
$item
)*
}
}
/// Enables enter::block_on
macro_rules! cfg_block_on {
($($item:item)*) => {
$(
#[cfg(any(
feature = "blocking",
feature = "fs",
feature = "dns",
feature = "io-std",
@@ -75,29 +15,13 @@ macro_rules! cfg_block_on {
}
}
/// Enables 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
)*
}
}
/// Enables internal `AtomicWaker` impl
macro_rules! cfg_atomic_waker_impl {
($($item:item)*) => {
$(
#[cfg(any(
feature = "process",
all(feature = "rt-core", feature = "rt-util"),
feature = "rt-util",
feature = "signal",
feature = "tcp",
feature = "time",
@@ -324,6 +248,16 @@ macro_rules! cfg_rt_core {
}
}
macro_rules! cfg_task {
($($item:item)*) => {
$(
#[cfg(any(feature = "rt-core", feature = "rt-util"))]
#[cfg_attr(docsrs, doc(cfg(any(feature = "rt-core", feature = "rt-util"))))]
$item
)*
}
}
macro_rules! doc_rt_core {
($($item:item)*) => {
$(
@@ -451,12 +385,12 @@ macro_rules! cfg_coop {
($($item:item)*) => {
$(
#[cfg(any(
feature = "blocking",
feature = "dns",
feature = "fs",
feature = "io-std",
feature = "process",
feature = "rt-core",
feature = "rt-util",
feature = "signal",
feature = "sync",
feature = "stream",
+1 -1
View File
@@ -16,7 +16,7 @@ mod ready;
mod thread_local;
#[macro_use]
#[cfg(feature = "rt-core")]
#[cfg(any(feature = "rt-core", feature = "rt-util"))]
pub(crate) mod scoped_tls;
cfg_macros! {
+2 -1
View File
@@ -1,5 +1,6 @@
cfg_macros! {
pub use crate::future::{maybe_done, poll_fn};
pub use crate::future::poll_fn;
pub use crate::future::maybe_done::maybe_done;
pub use crate::util::thread_rng_n;
}
+23 -22
View File
@@ -153,22 +153,22 @@ cfg_dns! {
type Future = sealed::MaybeReady;
fn to_socket_addrs(&self, _: sealed::Internal) -> Self::Future {
use crate::runtime::spawn_blocking;
use crate::blocking::spawn_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));
return MaybeReady(sealed::State::Ready(Some(addr)));
}
// Run DNS lookup on the blocking pool
let s = self.to_owned();
MaybeReady::Blocking(spawn_blocking(move || {
MaybeReady(sealed::State::Blocking(spawn_blocking(move || {
std::net::ToSocketAddrs::to_socket_addrs(&s)
}))
})))
}
}
@@ -181,7 +181,7 @@ cfg_dns! {
type Future = sealed::MaybeReady;
fn to_socket_addrs(&self, _: sealed::Internal) -> Self::Future {
use crate::runtime::spawn_blocking;
use crate::blocking::spawn_blocking;
use sealed::MaybeReady;
let (host, port) = *self;
@@ -191,21 +191,21 @@ cfg_dns! {
let addr = SocketAddrV4::new(addr, port);
let addr = SocketAddr::V4(addr);
return MaybeReady::Ready(Some(addr));
return MaybeReady(sealed::State::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));
return MaybeReady(sealed::State::Ready(Some(addr)));
}
let host = host.to_owned();
MaybeReady::Blocking(spawn_blocking(move || {
MaybeReady(sealed::State::Blocking(spawn_blocking(move || {
std::net::ToSocketAddrs::to_socket_addrs(&(&host[..], port))
}))
})))
}
}
@@ -245,15 +245,6 @@ pub(crate) mod sealed {
use std::io;
use std::net::SocketAddr;
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 {
type Iter: Iterator<Item = SocketAddr> + Send + 'static;
@@ -266,9 +257,19 @@ pub(crate) mod sealed {
pub struct Internal;
cfg_dns! {
use crate::blocking::JoinHandle;
use std::option;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::vec;
#[doc(hidden)]
#[derive(Debug)]
pub enum MaybeReady {
pub struct MaybeReady(pub(super) State);
#[derive(Debug)]
pub(super) enum State {
Ready(Option<SocketAddr>),
Blocking(JoinHandle<io::Result<vec::IntoIter<SocketAddr>>>),
}
@@ -284,12 +285,12 @@ pub(crate) mod sealed {
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) => {
match self.0 {
State::Ready(ref mut i) => {
let iter = OneOrMore::One(i.take().into_iter());
Poll::Ready(Ok(iter))
}
MaybeReady::Blocking(ref mut rx) => {
State::Blocking(ref mut rx) => {
let res = ready!(Pin::new(rx).poll(cx))?.map(OneOrMore::More);
Poll::Ready(res)
+2
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "full"), allow(dead_code))]
use crate::park::{Park, Unpark};
use std::fmt;
+4 -7
View File
@@ -34,15 +34,12 @@
//! * `park_timeout` does the same as `park` but allows specifying a maximum
//! time to block the thread for.
cfg_resource_drivers! {
mod either;
pub(crate) use self::either::Either;
cfg_rt_core! {
pub(crate) mod either;
}
mod thread;
pub(crate) use self::thread::ParkThread;
pub(crate) use self::thread::{CachedParkThread, ParkError};
#[cfg(any(feature = "rt-core", feature = "rt-util", feature = "sync"))]
pub(crate) mod thread;
use std::sync::Arc;
use std::time::Duration;
+23 -2
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "full"), allow(dead_code))]
use crate::loom::sync::atomic::AtomicUsize;
use crate::loom::sync::{Arc, Condvar, Mutex};
use crate::park::{Park, Unpark};
@@ -212,10 +214,10 @@ impl Unpark for UnparkThread {
}
}
use std::future::Future;
use std::marker::PhantomData;
use std::rc::Rc;
use std::mem;
use std::rc::Rc;
use std::task::{RawWaker, RawWakerVTable, Waker};
/// Blocks the current thread using a condition variable.
@@ -246,6 +248,25 @@ impl CachedParkThread {
{
CURRENT_PARKER.try_with(|inner| f(inner)).map_err(|_| ())
}
pub(crate) fn block_on<F: Future>(&mut self, f: F) -> Result<F::Output, ParkError> {
use std::task::Context;
use std::task::Poll::Ready;
// `get_unpark()` should not return a Result
let waker = self.get_unpark()?.into_waker();
let mut cx = Context::from_waker(&waker);
pin!(f);
loop {
if let Ready(v) = crate::coop::budget(|| f.as_mut().poll(&mut cx)) {
return Ok(v);
}
self.park()?;
}
}
}
impl Park for CachedParkThread {
+2
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "rt-core"), allow(dead_code))]
//! Process driver
use crate::park::Park;
+1 -1
View File
@@ -2,7 +2,7 @@ use crate::future::poll_fn;
use crate::loom::sync::Mutex;
use crate::park::{Park, Unpark};
use crate::runtime::task::{self, JoinHandle, Schedule, Task};
use crate::sync::Notify;
use crate::sync::notify::Notify;
use crate::util::linked_list::{Link, LinkedList};
use crate::util::{waker_ref, Wake, WakerRef};
+10 -10
View File
@@ -3,21 +3,20 @@
//! shells. This isolates the complexity of dealing with conditional
//! compilation.
cfg_blocking_impl! {
mod pool;
pub(crate) use pool::{spawn_blocking, try_spawn_blocking, BlockingPool, Spawner};
mod pool;
pub(crate) use pool::{spawn_blocking, BlockingPool, Spawner};
mod schedule;
mod shutdown;
pub(crate) mod task;
mod schedule;
mod shutdown;
pub(crate) mod task;
use crate::runtime::Builder;
use crate::runtime::Builder;
pub(crate) fn create_blocking_pool(builder: &Builder, thread_cap: usize) -> BlockingPool {
BlockingPool::new(builder, thread_cap)
}
pub(crate) fn create_blocking_pool(builder: &Builder, thread_cap: usize) -> BlockingPool {
BlockingPool::new(builder, thread_cap)
}
/*
cfg_not_blocking_impl! {
use crate::runtime::Builder;
use std::time::Duration;
@@ -40,3 +39,4 @@ cfg_not_blocking_impl! {
}
}
}
*/
-3
View File
@@ -94,10 +94,7 @@ where
impl BlockingPool {
pub(crate) fn new(builder: &Builder, thread_cap: usize) -> BlockingPool {
let (shutdown_tx, shutdown_rx) = shutdown::channel();
#[cfg(feature = "blocking")]
let keep_alive = builder.keep_alive.unwrap_or(KEEP_ALIVE);
#[cfg(not(feature = "blocking"))]
let keep_alive = KEEP_ALIVE;
BlockingPool {
spawner: Spawner {
+104 -160
View File
@@ -1,9 +1,8 @@
use crate::runtime::handle::Handle;
use crate::runtime::shell::Shell;
use crate::runtime::{blocking, driver, io, Callback, Runtime, Spawner};
use crate::runtime::{blocking, driver, Callback, Runtime, Spawner};
use std::fmt;
#[cfg(feature = "blocking")]
use std::io;
use std::time::Duration;
/// Builds Tokio Runtime with custom configuration values.
@@ -26,9 +25,8 @@ use std::time::Duration;
///
/// fn main() {
/// // build runtime
/// let runtime = Builder::new()
/// .threaded_scheduler()
/// .core_threads(4)
/// let runtime = Builder::new_multi_thread()
/// .worker_threads(4)
/// .thread_name("my-custom-name")
/// .thread_stack_size(3 * 1024 * 1024)
/// .build()
@@ -38,7 +36,7 @@ use std::time::Duration;
/// }
/// ```
pub struct Builder {
/// The task execution model to use.
/// Runtime type
kind: Kind,
/// Whether or not to enable the I/O driver
@@ -50,7 +48,7 @@ pub struct Builder {
/// The number of worker threads, used by Runtime.
///
/// Only used when not using the current-thread executor.
core_threads: Option<usize>,
worker_threads: Option<usize>,
/// Cap on thread usage.
max_threads: usize,
@@ -67,32 +65,37 @@ pub struct Builder {
/// To run before each worker thread stops
pub(super) before_stop: Option<Callback>,
#[cfg(feature = "blocking")]
#[cfg_attr(docsrs, doc(cfg(feature = "blocking")))]
/// Customizable keep alive timeout for BlockingPool
pub(super) keep_alive: Option<Duration>,
}
pub(crate) type ThreadNameFn = std::sync::Arc<dyn Fn() -> String + Send + Sync + 'static>;
#[derive(Debug, Clone, Copy)]
enum Kind {
Shell,
#[cfg(feature = "rt-core")]
Basic,
pub(crate) enum Kind {
CurrentThread,
#[cfg(feature = "rt-threaded")]
ThreadPool,
MultiThread,
}
impl Builder {
/// TODO
pub fn new_current_thread() -> Builder {
Builder::new(Kind::CurrentThread)
}
/// TODO
#[cfg(feature = "rt-threaded")]
pub fn new_multi_thread() -> Builder {
Builder::new(Kind::MultiThread)
}
/// Returns a new runtime builder initialized with default configuration
/// values.
///
/// Configuration methods can be chained on the return value.
pub fn new() -> Builder {
pub(crate) fn new(kind: Kind) -> Builder {
Builder {
// No task execution by default
kind: Kind::Shell,
kind,
// I/O defaults to "off"
enable_io: false,
@@ -101,7 +104,7 @@ impl Builder {
enable_time: false,
// Default to lazy auto-detection (one thread per CPU core)
core_threads: None,
worker_threads: None,
max_threads: 512,
@@ -115,7 +118,6 @@ impl Builder {
after_start: None,
before_stop: None,
#[cfg(feature = "blocking")]
keep_alive: None,
}
}
@@ -131,8 +133,7 @@ impl Builder {
/// ```
/// use tokio::runtime;
///
/// let rt = runtime::Builder::new()
/// .threaded_scheduler()
/// let rt = runtime::Builder::new_multi_thread()
/// .enable_all()
/// .build()
/// .unwrap();
@@ -152,51 +153,63 @@ impl Builder {
self
}
#[deprecated(note = "In future will be replaced by core_threads method")]
/// Sets the maximum number of worker threads for the `Runtime`'s thread pool.
/// Sets the number of worker threads the `Runtime` will use.
///
/// This must be a number between 1 and 32,768 though it is advised to keep
/// this value on the smaller side.
/// This should be a number between 0 and 32,768 though it is advised to
/// keep this value on the smaller side.
///
/// The default value is the number of cores available to the system.
pub fn num_threads(&mut self, val: usize) -> &mut Self {
self.core_threads = Some(val);
self
}
/// Sets the core number of worker threads for the `Runtime`'s thread pool.
///
/// This should be a number between 1 and 32,768 though it is advised to keep
/// this value on the smaller side.
/// # Default
///
/// The default value is the number of cores available to the system.
///
/// These threads will be always active and running.
/// # Panic
///
/// When using the `current_thread` runtime this method will panic, since
/// those variants do not allow setting worker thread counts.
///
///
/// # Examples
///
/// ## Multi threaded runtime with 4 threads
///
/// ```
/// use tokio::runtime;
///
/// let rt = runtime::Builder::new()
/// .threaded_scheduler()
/// .core_threads(4)
/// // This will spawn a work-stealing runtime with 4 worker threads.
/// let rt = runtime::Builder::new_multi_thread()
/// .worker_threads(4)
/// .build()
/// .unwrap();
///
/// rt.spawn(async move {});
/// ```
pub fn core_threads(&mut self, val: usize) -> &mut Self {
assert_ne!(val, 0, "Core threads cannot be zero");
self.core_threads = Some(val);
///
/// ## Current thread runtime (will only run on the current thread via `Runtime::block_on`)
///
/// ```
/// use tokio::runtime;
///
/// // Create a runtime that _must_ be driven from a call
/// // to `Runtime::block_on`.
/// let rt = runtime::Builder::new_current_thread()
/// .build()
/// .unwrap();
///
/// // This will run the runtime and future on the current thread
/// rt.block_on(async move {});
/// ```
pub fn worker_threads(&mut self, val: usize) -> &mut Self {
self.worker_threads = Some(val);
self
}
/// Specifies limit for threads, spawned by the Runtime.
///
/// This is number of threads to be used by Runtime, including `core_threads`
/// Having `max_threads` less than `core_threads` results in invalid configuration
/// Having `max_threads` less than `worker_threads` results in invalid configuration
/// when building multi-threaded `Runtime`, which would cause a panic.
///
/// Similarly to the `core_threads`, this number should be between 1 and 32,768.
/// Similarly to the `worker_threads`, this number should be between 0 and 32,768.
///
/// The default value is 512.
///
@@ -205,7 +218,6 @@ impl Builder {
/// Otherwise as `core_threads` are always active, it limits additional threads (e.g. for
/// blocking annotations) as `max_threads - core_threads`.
pub fn max_threads(&mut self, val: usize) -> &mut Self {
assert_ne!(val, 0, "Thread limit cannot be zero");
self.max_threads = val;
self
}
@@ -220,7 +232,7 @@ impl Builder {
/// # use tokio::runtime;
///
/// # pub fn main() {
/// let rt = runtime::Builder::new()
/// let rt = runtime::Builder::new_multi_thread()
/// .thread_name("my-pool")
/// .build();
/// # }
@@ -242,7 +254,7 @@ impl Builder {
/// # use std::sync::atomic::{AtomicUsize, Ordering};
///
/// # pub fn main() {
/// let rt = runtime::Builder::new()
/// let rt = runtime::Builder::new_multi_thread()
/// .thread_name_fn(|| {
/// static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
/// let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
@@ -273,8 +285,7 @@ impl Builder {
/// # use tokio::runtime;
///
/// # pub fn main() {
/// let rt = runtime::Builder::new()
/// .threaded_scheduler()
/// let rt = runtime::Builder::new_multi_thread()
/// .thread_stack_size(32 * 1024)
/// .build();
/// # }
@@ -295,8 +306,7 @@ impl Builder {
/// # use tokio::runtime;
///
/// # pub fn main() {
/// let runtime = runtime::Builder::new()
/// .threaded_scheduler()
/// let runtime = runtime::Builder::new_multi_thread()
/// .on_thread_start(|| {
/// println!("thread started");
/// })
@@ -322,8 +332,7 @@ impl Builder {
/// # use tokio::runtime;
///
/// # pub fn main() {
/// let runtime = runtime::Builder::new()
/// .threaded_scheduler()
/// let runtime = runtime::Builder::new_multi_thread()
/// .on_thread_stop(|| {
/// println!("thread stopping");
/// })
@@ -341,26 +350,24 @@ impl Builder {
/// Creates the configured `Runtime`.
///
/// The returned `ThreadPool` instance is ready to spawn tasks.
/// The returned `Runtime` instance is ready to spawn tasks.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Builder;
///
/// let rt = Builder::new().build().unwrap();
/// let rt = Builder::new_multi_thread().build().unwrap();
///
/// rt.block_on(async {
/// println!("Hello from the Tokio runtime");
/// });
/// ```
pub fn build(&mut self) -> io::Result<Runtime> {
match self.kind {
Kind::Shell => self.build_shell_runtime(),
#[cfg(feature = "rt-core")]
Kind::Basic => self.build_basic_runtime(),
match &self.kind {
Kind::CurrentThread => self.build_basic_runtime(),
#[cfg(feature = "rt-threaded")]
Kind::ThreadPool => self.build_threaded_runtime(),
Kind::MultiThread => self.build_threaded_runtime(),
}
}
@@ -371,32 +378,6 @@ impl Builder {
}
}
fn build_shell_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::Kind;
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
let spawner = Spawner::Shell;
let blocking_pool = blocking::create_blocking_pool(self, self.max_threads);
let blocking_spawner = blocking_pool.spawner().clone();
Ok(Runtime {
kind: Kind::Shell(Shell::new(driver)),
handle: Handle {
spawner,
io_handle: resources.io_handle,
time_handle: resources.time_handle,
signal_handle: resources.signal_handle,
clock: resources.clock,
blocking_spawner,
},
blocking_pool,
})
}
#[cfg(feature = "blocking")]
#[cfg_attr(docsrs, doc(cfg(feature = "blocking")))]
/// Sets a custom timeout for a thread in the blocking pool.
///
/// By default, the timeout for a thread is set to 10 seconds. This can
@@ -409,7 +390,7 @@ impl Builder {
/// # use std::time::Duration;
///
/// # pub fn main() {
/// let rt = runtime::Builder::new()
/// let rt = runtime::Builder::new_multi_thread()
/// .thread_keep_alive(Duration::from_millis(100))
/// .build();
/// # }
@@ -418,6 +399,36 @@ impl Builder {
self.keep_alive = Some(duration);
self
}
fn build_basic_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::{BasicScheduler, Kind};
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
// And now put a single-threaded scheduler on top of the timer. When
// there are no futures ready to do something, it'll let the timer or
// the reactor to generate some new stimuli for the futures to continue
// in their life.
let scheduler = BasicScheduler::new(driver);
let spawner = Spawner::Basic(scheduler.spawner().clone());
// Blocking pool
let blocking_pool = blocking::create_blocking_pool(self, self.max_threads);
let blocking_spawner = blocking_pool.spawner().clone();
Ok(Runtime {
kind: Kind::CurrentThread(scheduler),
handle: Handle {
spawner,
io_handle: resources.io_handle,
time_handle: resources.time_handle,
signal_handle: resources.signal_handle,
clock: resources.clock,
blocking_spawner,
},
blocking_pool,
})
}
}
cfg_io_driver! {
@@ -432,7 +443,7 @@ cfg_io_driver! {
/// ```
/// use tokio::runtime;
///
/// let rt = runtime::Builder::new()
/// let rt = runtime::Builder::new_multi_thread()
/// .enable_io()
/// .build()
/// .unwrap();
@@ -455,7 +466,7 @@ cfg_time! {
/// ```
/// use tokio::runtime;
///
/// let rt = runtime::Builder::new()
/// let rt = runtime::Builder::new_multi_thread()
/// .enable_time()
/// .build()
/// .unwrap();
@@ -467,75 +478,15 @@ cfg_time! {
}
}
cfg_rt_core! {
impl Builder {
/// Sets runtime to use a simpler scheduler that runs all tasks on the current-thread.
///
/// The executor and all necessary drivers will all be run on the current
/// thread during [`block_on`] calls.
///
/// See also [the module level documentation][1], which has a section on scheduler
/// types.
///
/// [1]: index.html#runtime-configurations
/// [`block_on`]: Runtime::block_on
pub fn basic_scheduler(&mut self) -> &mut Self {
self.kind = Kind::Basic;
self
}
fn build_basic_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::{BasicScheduler, Kind};
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
// And now put a single-threaded scheduler on top of the timer. When
// there are no futures ready to do something, it'll let the timer or
// the reactor to generate some new stimuli for the futures to continue
// in their life.
let scheduler = BasicScheduler::new(driver);
let spawner = Spawner::Basic(scheduler.spawner().clone());
// Blocking pool
let blocking_pool = blocking::create_blocking_pool(self, self.max_threads);
let blocking_spawner = blocking_pool.spawner().clone();
Ok(Runtime {
kind: Kind::Basic(scheduler),
handle: Handle {
spawner,
io_handle: resources.io_handle,
time_handle: resources.time_handle,
signal_handle: resources.signal_handle,
clock: resources.clock,
blocking_spawner,
},
blocking_pool,
})
}
}
}
cfg_rt_threaded! {
impl Builder {
/// Sets runtime to use a multi-threaded scheduler for executing tasks.
///
/// See also [the module level documentation][1], which has a section on scheduler
/// types.
///
/// [1]: index.html#runtime-configurations
pub fn threaded_scheduler(&mut self) -> &mut Self {
self.kind = Kind::ThreadPool;
self
}
fn build_threaded_runtime(&mut self) -> io::Result<Runtime> {
use crate::loom::sys::num_cpus;
use crate::runtime::{Kind, ThreadPool};
use crate::runtime::park::Parker;
use std::cmp;
let core_threads = self.core_threads.unwrap_or_else(|| cmp::min(self.max_threads, num_cpus()));
let core_threads = self.worker_threads.unwrap_or_else(|| cmp::min(self.max_threads, num_cpus()));
assert!(core_threads <= self.max_threads, "Core threads number cannot be above max limit");
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
@@ -569,17 +520,10 @@ cfg_rt_threaded! {
}
}
impl Default for Builder {
fn default() -> Self {
Self::new()
}
}
impl fmt::Debug for Builder {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Builder")
.field("kind", &self.kind)
.field("core_threads", &self.core_threads)
.field("worker_threads", &self.worker_threads)
.field("max_threads", &self.max_threads)
.field(
"thread_name",
+2 -4
View File
@@ -7,10 +7,8 @@ thread_local! {
static CONTEXT: RefCell<Option<Handle>> = RefCell::new(None)
}
cfg_blocking_impl! {
pub(crate) fn current() -> Option<Handle> {
CONTEXT.with(|ctx| ctx.borrow().clone())
}
pub(crate) fn current() -> Option<Handle> {
CONTEXT.with(|ctx| ctx.borrow().clone())
}
cfg_io_driver! {
+11 -9
View File
@@ -1,16 +1,18 @@
//! Abstracts out the entire chain of runtime sub-drivers into common types.
use crate::park::{Park, ParkThread};
use crate::park::thread::ParkThread;
use crate::park::Park;
use std::io;
use std::time::Duration;
// ===== io driver =====
cfg_io_driver! {
type IoDriver = crate::park::Either<crate::io::driver::Driver, crate::park::ParkThread>;
type IoDriver = crate::park::either::Either<crate::io::driver::Driver, ParkThread>;
pub(crate) type IoHandle = Option<crate::io::driver::Handle>;
fn create_io_driver(enable: bool) -> io::Result<(IoDriver, IoHandle)> {
use crate::park::Either;
use crate::park::either::Either;
#[cfg(loom)]
assert!(!enable);
@@ -47,11 +49,11 @@ macro_rules! cfg_signal_internal_and_unix {
}
cfg_signal_internal_and_unix! {
type SignalDriver = crate::park::Either<crate::signal::unix::driver::Driver, IoDriver>;
type SignalDriver = crate::park::either::Either<crate::signal::unix::driver::Driver, IoDriver>;
pub(crate) type SignalHandle = Option<crate::signal::unix::driver::Handle>;
fn create_signal_driver(io_driver: IoDriver) -> io::Result<(SignalDriver, SignalHandle)> {
use crate::park::Either;
use crate::park::either::Either;
// Enable the signal driver if IO is also enabled
match io_driver {
@@ -77,10 +79,10 @@ cfg_not_signal_internal! {
// ===== process driver =====
cfg_process_driver! {
type ProcessDriver = crate::park::Either<crate::process::unix::driver::Driver, SignalDriver>;
type ProcessDriver = crate::park::either::Either<crate::process::unix::driver::Driver, SignalDriver>;
fn create_process_driver(signal_driver: SignalDriver) -> io::Result<ProcessDriver> {
use crate::park::Either;
use crate::park::either::Either;
// Enable the signal driver if IO is also enabled
match signal_driver {
@@ -104,7 +106,7 @@ cfg_not_process_driver! {
// ===== time driver =====
cfg_time! {
type TimeDriver = crate::park::Either<crate::time::driver::Driver<ProcessDriver>, ProcessDriver>;
type TimeDriver = crate::park::either::Either<crate::time::driver::Driver<ProcessDriver>, ProcessDriver>;
pub(crate) type Clock = crate::time::Clock;
pub(crate) type TimeHandle = Option<crate::time::driver::Handle>;
@@ -118,7 +120,7 @@ cfg_time! {
process_driver: ProcessDriver,
clock: Clock,
) -> (TimeDriver, TimeHandle) {
use crate::park::Either;
use crate::park::either::Either;
if enable {
let driver = crate::time::driver::Driver::new(process_driver, clock);
+104 -118
View File
@@ -4,8 +4,8 @@ use std::marker::PhantomData;
#[derive(Debug, Clone, Copy)]
pub(crate) enum EnterContext {
#[cfg_attr(not(feature = "rt-core"), allow(dead_code))]
Entered {
#[allow(dead_code)]
allow_blocking: bool,
},
NotEntered,
@@ -24,32 +24,38 @@ pub(crate) struct Enter {
_p: PhantomData<RefCell<()>>,
}
/// Marks the current thread as being within the dynamic extent of an
/// executor.
pub(crate) fn enter(allow_blocking: bool) -> Enter {
if let Some(enter) = try_enter(allow_blocking) {
return enter;
cfg_rt_core! {
use crate::park::thread::ParkError;
use std::time::Duration;
/// Marks the current thread as being within the dynamic extent of an
/// executor.
pub(crate) fn enter(allow_blocking: bool) -> Enter {
if let Some(enter) = try_enter(allow_blocking) {
return enter;
}
panic!(
"Cannot start a runtime from within a runtime. This happens \
because a function (like `block_on`) attempted to block the \
current thread while the thread is being used to drive \
asynchronous tasks."
);
}
panic!(
"Cannot start a runtime from within a runtime. This happens \
because a function (like `block_on`) attempted to block the \
current thread while the thread is being used to drive \
asynchronous tasks."
);
}
/// Tries to enter a runtime context, returns `None` if already in a runtime
/// context.
pub(crate) fn try_enter(allow_blocking: bool) -> Option<Enter> {
ENTERED.with(|c| {
if c.get().is_entered() {
None
} else {
c.set(EnterContext::Entered { allow_blocking });
Some(Enter { _p: PhantomData })
}
})
/// Tries to enter a runtime context, returns `None` if already in a runtime
/// context.
pub(crate) fn try_enter(allow_blocking: bool) -> Option<Enter> {
ENTERED.with(|c| {
if c.get().is_entered() {
None
} else {
c.set(EnterContext::Entered { allow_blocking });
Some(Enter { _p: PhantomData })
}
})
}
}
// Forces the current "entered" state to be cleared while the closure
@@ -59,113 +65,92 @@ pub(crate) fn try_enter(allow_blocking: bool) -> Option<Enter> {
//
// This is hidden for a reason. Do not use without fully understanding
// executors. Misuing can easily cause your program to deadlock.
#[cfg(all(feature = "rt-threaded", feature = "blocking"))]
pub(crate) fn exit<F: FnOnce() -> R, R>(f: F) -> R {
// Reset in case the closure panics
struct Reset(EnterContext);
impl Drop for Reset {
fn drop(&mut self) {
ENTERED.with(|c| {
assert!(!c.get().is_entered(), "closure claimed permanent executor");
c.set(self.0);
});
cfg_rt_threaded! {
pub(crate) fn exit<F: FnOnce() -> R, R>(f: F) -> R {
// Reset in case the closure panics
struct Reset(EnterContext);
impl Drop for Reset {
fn drop(&mut self) {
ENTERED.with(|c| {
assert!(!c.get().is_entered(), "closure claimed permanent executor");
c.set(self.0);
});
}
}
let was = ENTERED.with(|c| {
let e = c.get();
assert!(e.is_entered(), "asked to exit when not entered");
c.set(EnterContext::NotEntered);
e
});
let _reset = Reset(was);
// dropping _reset after f() will reset ENTERED
f()
}
let was = ENTERED.with(|c| {
let e = c.get();
assert!(e.is_entered(), "asked to exit when not entered");
c.set(EnterContext::NotEntered);
e
});
let _reset = Reset(was);
// dropping _reset after f() will reset ENTERED
f()
}
cfg_rt_core! {
cfg_rt_util! {
/// Disallow blocking in the current runtime context until the guard is dropped.
pub(crate) fn disallow_blocking() -> DisallowBlockingGuard {
let reset = ENTERED.with(|c| {
if let EnterContext::Entered {
allow_blocking: true,
} = c.get()
{
c.set(EnterContext::Entered {
allow_blocking: false,
});
true
} else {
false
}
});
DisallowBlockingGuard(reset)
}
cfg_rt_util! {
/// Disallow blocking in the current runtime context until the guard is dropped.
pub(crate) fn disallow_blocking() -> DisallowBlockingGuard {
let reset = ENTERED.with(|c| {
if let EnterContext::Entered {
allow_blocking: true,
} = c.get()
{
c.set(EnterContext::Entered {
allow_blocking: false,
});
true
} else {
false
}
});
DisallowBlockingGuard(reset)
}
pub(crate) struct DisallowBlockingGuard(bool);
impl Drop for DisallowBlockingGuard {
fn drop(&mut self) {
if self.0 {
// XXX: Do we want some kind of assertion here, or is "best effort" okay?
ENTERED.with(|c| {
if let EnterContext::Entered {
allow_blocking: false,
} = c.get()
{
c.set(EnterContext::Entered {
allow_blocking: true,
});
}
})
}
pub(crate) struct DisallowBlockingGuard(bool);
impl Drop for DisallowBlockingGuard {
fn drop(&mut self) {
if self.0 {
// XXX: Do we want some kind of assertion here, or is "best effort" okay?
ENTERED.with(|c| {
if let EnterContext::Entered {
allow_blocking: false,
} = c.get()
{
c.set(EnterContext::Entered {
allow_blocking: true,
});
}
})
}
}
}
}
cfg_rt_threaded! {
cfg_blocking! {
/// Returns true if in a runtime context.
pub(crate) fn context() -> EnterContext {
ENTERED.with(|c| c.get())
}
/// Returns true if in a runtime context.
pub(crate) fn context() -> EnterContext {
ENTERED.with(|c| c.get())
}
}
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, f: F) -> Result<F::Output, crate::park::ParkError>
where
F: std::future::Future,
{
use crate::park::{CachedParkThread, Park};
use std::task::Context;
use std::task::Poll::Ready;
let mut park = CachedParkThread::new();
let waker = park.get_unpark()?.into_waker();
let mut cx = Context::from_waker(&waker);
pin!(f);
loop {
if let Ready(v) = crate::coop::budget(|| f.as_mut().poll(&mut cx)) {
return Ok(v);
}
park.park()?;
}
}
}
cfg_blocking_impl! {
use crate::park::ParkError;
use std::time::Duration;
cfg_rt_core! {
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, f: F) -> Result<F::Output, ParkError>
where
F: std::future::Future,
{
use crate::park::thread::CachedParkThread;
let mut park = CachedParkThread::new();
park.block_on(f)
}
/// Blocks the thread on the specified future for **at most** `timeout`
///
/// If the future completes before `timeout`, the result is returned. If
@@ -174,7 +159,8 @@ cfg_blocking_impl! {
where
F: std::future::Future,
{
use crate::park::{CachedParkThread, Park};
use crate::park::Park;
use crate::park::thread::CachedParkThread;
use std::task::Context;
use std::task::Poll::Ready;
use std::time::Instant;
+305 -353
View File
@@ -1,8 +1,7 @@
//! The Tokio runtime.
//!
//! Unlike other Rust programs, asynchronous applications require
//! runtime support. In particular, the following runtime services are
//! necessary:
//! Unlike other Rust programs, asynchronous applications require runtime
//! support. In particular, the following runtime services are necessary:
//!
//! * An **I/O event loop**, called the driver, which drives I/O resources and
//! dispatches I/O events to tasks that depend on them.
@@ -10,14 +9,14 @@
//! * A **timer** for scheduling work to run after a set period of time.
//!
//! Tokio's [`Runtime`] bundles all of these services as a single type, allowing
//! them to be started, shut down, and configured together. However, often
//! it is not required to configure a [`Runtime`] manually, and user may just
//! use the [`tokio::main`] attribute macro, which creates a [`Runtime`] under
//! the hood.
//! them to be started, shut down, and configured together. However, often it is
//! not required to configure a [`Runtime`] manually, and user may just use the
//! [`tokio::main`] attribute macro, which creates a [`Runtime`] under the hood.
//!
//! # Usage
//!
//! When no fine tuning is required, the [`tokio::main`] attribute macro can be used.
//! When no fine tuning is required, the [`tokio::main`] attribute macro can be
//! used.
//!
//! ```no_run
//! use tokio::net::TcpListener;
@@ -111,47 +110,37 @@
//! applications. The [runtime builder] or `#[tokio::main]` attribute may be
//! used to select which scheduler to use.
//!
//! #### Basic Scheduler
//! #### Current-Thread Scheduler
//!
//! The basic scheduler provides a _single-threaded_ future executor. All tasks
//! will be created and executed on the current thread. The basic scheduler
//! requires the `rt-core` feature flag, and can be selected using the
//! [`Builder::basic_scheduler`] method:
//! The current-thread scheduler provides a _single-threaded_ future executor.
//! All tasks will be created and executed on the current thread. This requires
//! the `rt-core` feature flag.
//! ```
//! use tokio::runtime;
//!
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let basic_rt = runtime::Builder::new()
//! .basic_scheduler()
//! let basic_rt = runtime::Builder::new_current_thread()
//! .build()?;
//! # Ok(()) }
//! ```
//!
//! If the `rt-core` feature is enabled and `rt-threaded` is not,
//! [`Runtime::new`] will return a basic scheduler runtime by default.
//! #### Multi-Thread Scheduler
//!
//! #### Threaded Scheduler
//!
//! The threaded scheduler executes futures on a _thread pool_, using a
//! The multi-thread scheduler executes futures on a _thread pool_, using a
//! work-stealing strategy. By default, it will start a worker thread for each
//! CPU core available on the system. This tends to be the ideal configurations
//! for most applications. The threaded scheduler requires the `rt-threaded` feature
//! flag, and can be selected using the [`Builder::threaded_scheduler`] method:
//! for most applications. The multi-thread scheduler requires the `rt-threaded`
//! feature flag, and is selected by default:
//! ```
//! use tokio::runtime;
//!
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let threaded_rt = runtime::Builder::new()
//! .threaded_scheduler()
//! .build()?;
//! let threaded_rt = runtime::Runtime::new()?;
//! # Ok(()) }
//! ```
//!
//! If the `rt-threaded` feature flag is enabled, [`Runtime::new`] will return a
//! threaded scheduler runtime by default.
//!
//! Most applications should use the threaded scheduler, except in some niche
//! use-cases, such as when running only a single thread is required.
//! Most applications should use the multi-thread scheduler, except in some
//! niche use-cases, such as when running only a single thread is required.
//!
//! #### Resource drivers
//!
@@ -188,37 +177,31 @@
#[macro_use]
mod tests;
pub(crate) mod context;
pub(crate) mod enter;
pub(crate) mod task;
cfg_rt_core! {
mod basic_scheduler;
use basic_scheduler::BasicScheduler;
pub(crate) mod task;
}
mod blocking;
use blocking::BlockingPool;
pub(crate) use blocking::spawn_blocking;
mod blocking;
use blocking::BlockingPool;
mod builder;
pub use self::builder::Builder;
cfg_blocking_impl! {
#[allow(unused_imports)]
pub(crate) use blocking::{spawn_blocking, try_spawn_blocking};
}
pub(crate) mod context;
pub(crate) mod driver;
mod builder;
pub use self::builder::Builder;
use self::enter::enter;
pub(crate) mod driver;
mod handle;
use handle::Handle;
pub(crate) mod enter;
use self::enter::enter;
mod handle;
use handle::Handle;
mod io {
/// Re-exported for convenience.
pub(crate) use std::io::Result;
mod spawner;
use self::spawner::Spawner;
}
cfg_rt_threaded! {
@@ -226,12 +209,6 @@ cfg_rt_threaded! {
use park::Parker;
}
mod shell;
use self::shell::Shell;
mod spawner;
use self::spawner::Spawner;
cfg_rt_threaded! {
mod queue;
@@ -241,318 +218,293 @@ cfg_rt_threaded! {
cfg_rt_core! {
use crate::task::JoinHandle;
}
use std::future::Future;
use std::time::Duration;
use std::future::Future;
use std::time::Duration;
/// The Tokio runtime.
///
/// The runtime provides an I/O driver, task scheduler, [timer], and blocking
/// pool, necessary for running asynchronous tasks.
///
/// Instances of `Runtime` can be created using [`new`] or [`Builder`]. However,
/// most users will use the `#[tokio::main]` annotation on their entry point instead.
///
/// See [module level][mod] documentation for more details.
///
/// # Shutdown
///
/// Shutting down the runtime is done by dropping the value. The current thread
/// will block until the shut down operation has completed.
///
/// * Drain any scheduled work queues.
/// * Drop any futures that have not yet completed.
/// * Drop the reactor.
///
/// Once the reactor has dropped, any outstanding I/O resources bound to
/// that reactor will no longer function. Calling any method on them will
/// result in an error.
///
/// [timer]: crate::time
/// [mod]: index.html
/// [`new`]: method@Self::new
/// [`Builder`]: struct@Builder
/// [`tokio::run`]: fn@run
#[derive(Debug)]
pub struct Runtime {
/// Task executor
kind: Kind,
/// Handle to runtime, also contains driver handles
handle: Handle,
/// Blocking pool handle, used to signal shutdown
blocking_pool: BlockingPool,
}
/// The runtime executor is either a thread-pool or a current-thread executor.
#[derive(Debug)]
enum Kind {
/// Not able to execute concurrent tasks. This variant is mostly used to get
/// access to the driver handles.
Shell(Shell),
/// Execute all tasks on the current-thread.
#[cfg(feature = "rt-core")]
Basic(BasicScheduler<driver::Driver>),
/// Execute tasks across multiple threads.
#[cfg(feature = "rt-threaded")]
ThreadPool(ThreadPool),
}
/// After thread starts / before thread stops
type Callback = std::sync::Arc<dyn Fn() + Send + Sync>;
impl Runtime {
/// Create a new runtime instance with default configuration values.
/// The Tokio runtime.
///
/// This results in a scheduler, I/O driver, and time driver being
/// initialized. The type of scheduler used depends on what feature flags
/// are enabled: if the `rt-threaded` feature is enabled, the [threaded
/// scheduler] is used, while if only the `rt-core` feature is enabled, the
/// [basic scheduler] is used instead.
/// The runtime provides an I/O driver, task scheduler, [timer], and
/// blocking pool, necessary for running asynchronous tasks.
///
/// If the threaded scheduler is selected, it will not spawn
/// any worker threads until it needs to, i.e. tasks are scheduled to run.
///
/// Most applications will not need to call this function directly. Instead,
/// they will use the [`#[tokio::main]` attribute][main]. When more complex
/// configuration is necessary, the [runtime builder] may be used.
/// Instances of `Runtime` can be created using [`new`], or [`Builder`].
/// However, most users will use the `#[tokio::main]` annotation on their
/// entry point instead.
///
/// See [module level][mod] documentation for more details.
///
/// # Examples
/// # Shutdown
///
/// Creating a new `Runtime` with default configuration values.
/// Shutting down the runtime is done by dropping the value. The current
/// thread will block until the shut down operation has completed.
///
/// ```
/// use tokio::runtime::Runtime;
/// * Drain any scheduled work queues.
/// * Drop any futures that have not yet completed.
/// * Drop the reactor.
///
/// let rt = Runtime::new()
/// .unwrap();
///
/// // Use the runtime...
/// ```
/// Once the reactor has dropped, any outstanding I/O resources bound to
/// that reactor will no longer function. Calling any method on them will
/// result in an error.
///
/// [timer]: crate::time
/// [mod]: index.html
/// [main]: ../attr.main.html
/// [threaded scheduler]: index.html#threaded-scheduler
/// [basic scheduler]: index.html#basic-scheduler
/// [runtime builder]: crate::runtime::Builder
pub fn new() -> io::Result<Runtime> {
#[cfg(feature = "rt-threaded")]
let ret = Builder::new().threaded_scheduler().enable_all().build();
/// [`new`]: method@Self::new
/// [`Builder`]: struct@Builder
#[derive(Debug)]
pub struct Runtime {
/// Task executor
kind: Kind,
#[cfg(all(not(feature = "rt-threaded"), feature = "rt-core"))]
let ret = Builder::new().basic_scheduler().enable_all().build();
/// Handle to runtime, also contains driver handles
handle: Handle,
#[cfg(not(feature = "rt-core"))]
let ret = Builder::new().enable_all().build();
ret
/// Blocking pool handle, used to signal shutdown
blocking_pool: BlockingPool,
}
/// 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();
///
/// // Spawn a future onto the runtime
/// rt.spawn(async {
/// println!("now running on a worker thread");
/// });
/// # }
/// ```
///
/// # Panics
///
/// This function will not panic unless task execution is disabled on the
/// executor. This can only happen if the runtime was built using
/// [`Builder`] without picking either [`basic_scheduler`] or
/// [`threaded_scheduler`].
///
/// [`Builder`]: struct@Builder
/// [`threaded_scheduler`]: fn@Builder::threaded_scheduler
/// [`basic_scheduler`]: fn@Builder::basic_scheduler
#[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!("task execution disabled"),
#[cfg(feature = "rt-threaded")]
Kind::ThreadPool(exec) => exec.spawn(future),
Kind::Basic(exec) => exec.spawn(future),
/// The runtime executor is either a thread-pool or a current-thread executor.
#[derive(Debug)]
enum Kind {
/// Execute all tasks on the current-thread.
#[cfg(feature = "rt-core")]
CurrentThread(BasicScheduler<driver::Driver>),
/// Execute tasks across multiple threads.
#[cfg(feature = "rt-threaded")]
ThreadPool(ThreadPool),
}
/// After thread starts / before thread stops
type Callback = std::sync::Arc<dyn Fn() + Send + Sync>;
impl Runtime {
/// Create a new runtime instance with default configuration values.
///
/// This results in a scheduler, I/O driver, and time driver being
/// initialized. The type of scheduler used depends on what feature flags
/// are enabled: if the `rt-threaded` feature is enabled, the [threaded
/// scheduler] is used, while if only the `rt-core` feature is enabled, the
/// [basic scheduler] is used instead.
///
/// If the threaded scheduler is selected, it will not spawn
/// any worker threads until it needs to, i.e. tasks are scheduled to run.
///
/// Most applications will not need to call this function directly. Instead,
/// they will use the [`#[tokio::main]` attribute][main]. When more complex
/// configuration is necessary, the [runtime builder] may be used.
///
/// See [module level][mod] documentation for more details.
///
/// # Examples
///
/// Creating a new `Runtime` with default configuration values.
///
/// ```
/// use tokio::runtime::Runtime;
///
/// let rt = Runtime::new()
/// .unwrap();
///
/// // Use the runtime...
/// ```
///
/// [mod]: index.html
/// [main]: ../attr.main.html
/// [threaded scheduler]: index.html#threaded-scheduler
/// [basic scheduler]: index.html#basic-scheduler
/// [runtime builder]: crate::runtime::Builder
#[cfg(feature = "rt-threaded")]
pub fn new() -> std::io::Result<Runtime> {
Builder::new_multi_thread().enable_all().build()
}
/// 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();
///
/// // Spawn a future onto the runtime
/// rt.spawn(async {
/// println!("now running on a worker thread");
/// });
/// # }
/// ```
#[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 {
#[cfg(feature = "rt-threaded")]
Kind::ThreadPool(exec) => exec.spawn(future),
Kind::CurrentThread(exec) => exec.spawn(future),
}
}
/// Run a future to completion on the Tokio runtime. This is the runtime's
/// entry point.
///
/// This runs the given future on the runtime, blocking until it is
/// complete, and yielding its resolved result. Any tasks or timers which
/// the future spawns internally will be executed on the runtime.
///
/// When this runtime is configured with `core_threads = 0`, only the first call
/// to `block_on` will run the IO and timer drivers. Calls to other methods _before_ the first
/// `block_on` completes will just hook into the driver running on the thread
/// that first called `block_on`. This means that the driver may be passed
/// from thread to thread by the user between calls to `block_on`.
///
/// This method may not be called from an asynchronous context.
///
/// # Panics
///
/// This function panics if the provided future panics, or if called within an
/// asynchronous execution context.
///
/// # Examples
///
/// ```no_run
/// use tokio::runtime::Runtime;
///
/// // Create the runtime
/// let rt = Runtime::new().unwrap();
///
/// // Execute the future, blocking the current thread until completion
/// rt.block_on(async {
/// println!("hello");
/// });
/// ```
///
/// [handle]: fn@Handle::block_on
pub fn block_on<F: Future>(&self, future: F) -> F::Output {
self.handle.enter(|| match &self.kind {
#[cfg(feature = "rt-core")]
Kind::CurrentThread(exec) => exec.block_on(future),
#[cfg(feature = "rt-threaded")]
Kind::ThreadPool(exec) => exec.block_on(future),
})
}
/// Enter the runtime context. This allows you to construct types that must
/// have an executor available on creation such as [`Sleep`] or [`TcpStream`].
/// It will also allow you to call methods such as [`tokio::spawn`].
///
/// [`Sleep`]: struct@crate::time::Sleep
/// [`TcpStream`]: struct@crate::net::TcpStream
/// [`tokio::spawn`]: fn@crate::spawn
///
/// # Example
///
/// ```
/// use tokio::runtime::Runtime;
///
/// fn function_that_spawns(msg: String) {
/// // Had we not used `rt.enter` below, this would panic.
/// tokio::spawn(async move {
/// println!("{}", msg);
/// });
/// }
///
/// fn main() {
/// let rt = Runtime::new().unwrap();
///
/// let s = "Hello World!".to_string();
///
/// // By entering the context, we tie `tokio::spawn` to this executor.
/// rt.enter(|| function_that_spawns(s));
/// }
/// ```
pub fn enter<F, R>(&self, f: F) -> R
where
F: FnOnce() -> R,
{
self.handle.enter(f)
}
/// Shutdown the runtime, waiting for at most `duration` for all spawned
/// task to shutdown.
///
/// Usually, dropping a `Runtime` handle is sufficient as tasks are able to
/// shutdown in a timely fashion. However, dropping a `Runtime` will wait
/// indefinitely for all tasks to terminate, and there are cases where a long
/// blocking task has been spawned, which can block dropping `Runtime`.
///
/// In this case, calling `shutdown_timeout` with an explicit wait timeout
/// can work. The `shutdown_timeout` will signal all tasks to shutdown and
/// will wait for at most `duration` for all spawned tasks to terminate. If
/// `timeout` elapses before all tasks are dropped, the function returns and
/// outstanding tasks are potentially leaked.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Runtime;
/// use tokio::task;
///
/// use std::thread;
/// use std::time::Duration;
///
/// fn main() {
/// let runtime = Runtime::new().unwrap();
///
/// runtime.block_on(async move {
/// task::spawn_blocking(move || {
/// thread::sleep(Duration::from_secs(10_000));
/// });
/// });
///
/// runtime.shutdown_timeout(Duration::from_millis(100));
/// }
/// ```
pub fn shutdown_timeout(mut self, duration: Duration) {
// Wakeup and shutdown all the worker threads
self.handle.spawner.shutdown();
self.blocking_pool.shutdown(Some(duration));
}
/// Shutdown the runtime, without waiting for any spawned tasks to shutdown.
///
/// This can be useful if you want to drop a runtime from within another runtime.
/// Normally, dropping a runtime will block indefinitely for spawned blocking tasks
/// to complete, which would normally not be permitted within an asynchronous context.
/// By calling `shutdown_background()`, you can drop the runtime from such a context.
///
/// Note however, that because we do not wait for any blocking tasks to complete, this
/// may result in a resource leak (in that any blocking tasks are still running until they
/// return.
///
/// This function is equivalent to calling `shutdown_timeout(Duration::of_nanos(0))`.
///
/// ```
/// use tokio::runtime::Runtime;
///
/// fn main() {
/// let runtime = Runtime::new().unwrap();
///
/// runtime.block_on(async move {
/// let inner_runtime = Runtime::new().unwrap();
/// // ...
/// inner_runtime.shutdown_background();
/// });
/// }
/// ```
pub fn shutdown_background(self) {
self.shutdown_timeout(Duration::from_nanos(0))
}
}
/// Run a future to completion on the Tokio runtime. This is the runtime's
/// entry point.
///
/// This runs the given future on the runtime, blocking until it is
/// complete, and yielding its resolved result. Any tasks or timers which
/// the future spawns internally will be executed on the runtime.
///
/// When this runtime is configured with `core_threads = 0`, only the first call
/// to `block_on` will run the IO and timer drivers. Calls to other methods _before_ the first
/// `block_on` completes will just hook into the driver running on the thread
/// that first called `block_on`. This means that the driver may be passed
/// from thread to thread by the user between calls to `block_on`.
///
/// This method may not be called from an asynchronous context.
///
/// # Panics
///
/// This function panics if the provided future panics, or if called within an
/// asynchronous execution context.
///
/// # Examples
///
/// ```no_run
/// use tokio::runtime::Runtime;
///
/// // Create the runtime
/// let rt = Runtime::new().unwrap();
///
/// // Execute the future, blocking the current thread until completion
/// rt.block_on(async {
/// println!("hello");
/// });
/// ```
///
/// [handle]: fn@Handle::block_on
pub fn block_on<F: Future>(&self, future: F) -> F::Output {
self.handle.enter(|| match &self.kind {
Kind::Shell(exec) => exec.block_on(future),
#[cfg(feature = "rt-core")]
Kind::Basic(exec) => exec.block_on(future),
#[cfg(feature = "rt-threaded")]
Kind::ThreadPool(exec) => exec.block_on(future),
})
}
/// Enter the runtime context. This allows you to construct types that must
/// have an executor available on creation such as [`Sleep`] or [`TcpStream`].
/// It will also allow you to call methods such as [`tokio::spawn`].
///
/// [`Sleep`]: struct@crate::time::Sleep
/// [`TcpStream`]: struct@crate::net::TcpStream
/// [`tokio::spawn`]: fn@crate::spawn
///
/// # Example
///
/// ```
/// use tokio::runtime::Runtime;
///
/// fn function_that_spawns(msg: String) {
/// // Had we not used `rt.enter` below, this would panic.
/// tokio::spawn(async move {
/// println!("{}", msg);
/// });
/// }
///
/// fn main() {
/// let rt = Runtime::new().unwrap();
///
/// let s = "Hello World!".to_string();
///
/// // By entering the context, we tie `tokio::spawn` to this executor.
/// rt.enter(|| function_that_spawns(s));
/// }
/// ```
pub fn enter<F, R>(&self, f: F) -> R
where
F: FnOnce() -> R,
{
self.handle.enter(f)
}
/// Shutdown the runtime, waiting for at most `duration` for all spawned
/// task to shutdown.
///
/// Usually, dropping a `Runtime` handle is sufficient as tasks are able to
/// shutdown in a timely fashion. However, dropping a `Runtime` will wait
/// indefinitely for all tasks to terminate, and there are cases where a long
/// blocking task has been spawned, which can block dropping `Runtime`.
///
/// In this case, calling `shutdown_timeout` with an explicit wait timeout
/// can work. The `shutdown_timeout` will signal all tasks to shutdown and
/// will wait for at most `duration` for all spawned tasks to terminate. If
/// `timeout` elapses before all tasks are dropped, the function returns and
/// outstanding tasks are potentially leaked.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Runtime;
/// use tokio::task;
///
/// use std::thread;
/// use std::time::Duration;
///
/// fn main() {
/// let runtime = Runtime::new().unwrap();
///
/// runtime.block_on(async move {
/// task::spawn_blocking(move || {
/// thread::sleep(Duration::from_secs(10_000));
/// });
/// });
///
/// runtime.shutdown_timeout(Duration::from_millis(100));
/// }
/// ```
pub fn shutdown_timeout(mut self, duration: Duration) {
// Wakeup and shutdown all the worker threads
self.handle.spawner.shutdown();
self.blocking_pool.shutdown(Some(duration));
}
/// Shutdown the runtime, without waiting for any spawned tasks to shutdown.
///
/// This can be useful if you want to drop a runtime from within another runtime.
/// Normally, dropping a runtime will block indefinitely for spawned blocking tasks
/// to complete, which would normally not be permitted within an asynchronous context.
/// By calling `shutdown_background()`, you can drop the runtime from such a context.
///
/// Note however, that because we do not wait for any blocking tasks to complete, this
/// may result in a resource leak (in that any blocking tasks are still running until they
/// return.
///
/// This function is equivalent to calling `shutdown_timeout(Duration::of_nanos(0))`.
///
/// ```
/// use tokio::runtime::Runtime;
///
/// fn main() {
/// let runtime = Runtime::new().unwrap();
///
/// runtime.block_on(async move {
/// let inner_runtime = Runtime::new().unwrap();
/// // ...
/// inner_runtime.shutdown_background();
/// });
/// }
/// ```
pub fn shutdown_background(self) {
self.shutdown_timeout(Duration::from_nanos(0))
}
}
-2
View File
@@ -11,7 +11,6 @@ cfg_rt_threaded! {
#[derive(Debug, Clone)]
pub(crate) enum Spawner {
Shell,
#[cfg(feature = "rt-core")]
Basic(basic_scheduler::Spawner),
#[cfg(feature = "rt-threaded")]
@@ -37,7 +36,6 @@ cfg_rt_core! {
F::Output: Send + 'static,
{
match self {
Spawner::Shell => panic!("spawning not enabled for runtime"),
#[cfg(feature = "rt-core")]
Spawner::Basic(spawner) => spawner.spawn(future),
#[cfg(feature = "rt-threaded")]
+1 -1
View File
@@ -3,7 +3,7 @@ use std::fmt;
use std::io;
use std::sync::Mutex;
doc_rt_core! {
cfg_task! {
/// Task failed to execute to completion.
pub struct JoinError {
repr: Repr,
+1 -1
View File
@@ -6,7 +6,7 @@ use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
doc_rt_core! {
cfg_task! {
/// An owned permission to join on a task (await its termination).
///
/// This can be thought of as the equivalent of [`std::thread::JoinHandle`] for
+15 -13
View File
@@ -79,22 +79,24 @@ pub(crate) trait Schedule: Sync + Sized + 'static {
}
}
/// Create a new task with an associated join handle
pub(crate) fn joinable<T, S>(task: T) -> (Notified<S>, JoinHandle<T::Output>)
where
T: Future + Send + 'static,
S: Schedule,
{
let raw = RawTask::new::<_, S>(task);
cfg_rt_core! {
/// Create a new task with an associated join handle
pub(crate) fn joinable<T, S>(task: T) -> (Notified<S>, JoinHandle<T::Output>)
where
T: Future + Send + 'static,
S: Schedule,
{
let raw = RawTask::new::<_, S>(task);
let task = Task {
raw,
_p: PhantomData,
};
let task = Task {
raw,
_p: PhantomData,
};
let join = JoinHandle::new(raw);
let join = JoinHandle::new(raw);
(Notified(task), join)
(Notified(task), join)
}
}
cfg_rt_util! {
+2 -3
View File
@@ -23,9 +23,8 @@ fn blocking_shutdown() {
}
fn mk_runtime(num_threads: usize) -> Runtime {
runtime::Builder::new()
.threaded_scheduler()
.core_threads(num_threads)
runtime::Builder::new_multi_thread()
.worker_threads(num_threads)
.build()
.unwrap()
}
+2 -3
View File
@@ -296,9 +296,8 @@ mod group_d {
}
fn mk_pool(num_threads: usize) -> Runtime {
runtime::Builder::new()
.threaded_scheduler()
.core_threads(num_threads)
runtime::Builder::new_multi_thread()
.worker_threads(num_threads)
.build()
.unwrap()
}
@@ -22,7 +22,6 @@ impl<T> AtomicCell<T> {
from_raw(old)
}
#[cfg(feature = "blocking")]
pub(super) fn set(&self, val: Box<T>) {
let _ = self.swap(Some(val));
}
+1 -3
View File
@@ -9,9 +9,7 @@ use self::idle::Idle;
mod worker;
pub(crate) use worker::Launch;
cfg_blocking! {
pub(crate) use worker::block_in_place;
}
pub(crate) use worker::block_in_place;
use crate::loom::sync::Arc;
use crate::runtime::task::{self, JoinHandle};
+81 -84
View File
@@ -9,6 +9,7 @@ use crate::loom::rand::seed;
use crate::loom::sync::{Arc, Mutex};
use crate::park::{Park, Unpark};
use crate::runtime;
use crate::runtime::enter::EnterContext;
use crate::runtime::park::{Parker, Unparker};
use crate::runtime::thread_pool::{AtomicCell, Idle};
use crate::runtime::{queue, task};
@@ -172,100 +173,96 @@ pub(super) fn create(size: usize, park: Parker) -> (Arc<Shared>, Launch) {
(shared, launch)
}
cfg_blocking! {
use crate::runtime::enter::EnterContext;
pub(crate) fn block_in_place<F, R>(f: F) -> R
where
F: FnOnce() -> R,
{
// Try to steal the worker core back
struct Reset(coop::Budget);
pub(crate) fn block_in_place<F, R>(f: F) -> R
where
F: FnOnce() -> R,
{
// Try to steal the worker core back
struct Reset(coop::Budget);
impl Drop for Reset {
fn drop(&mut self) {
CURRENT.with(|maybe_cx| {
if let Some(cx) = maybe_cx {
let core = cx.worker.core.take();
let mut cx_core = cx.core.borrow_mut();
assert!(cx_core.is_none());
*cx_core = core;
impl Drop for Reset {
fn drop(&mut self) {
CURRENT.with(|maybe_cx| {
if let Some(cx) = maybe_cx {
let core = cx.worker.core.take();
let mut cx_core = cx.core.borrow_mut();
assert!(cx_core.is_none());
*cx_core = core;
// Reset the task budget as we are re-entering the
// runtime.
coop::set(self.0);
}
});
}
// Reset the task budget as we are re-entering the
// runtime.
coop::set(self.0);
}
});
}
}
let mut had_entered = false;
let mut had_entered = false;
CURRENT.with(|maybe_cx| {
match (crate::runtime::enter::context(), maybe_cx.is_some()) {
(EnterContext::Entered { .. }, true) => {
// We are on a thread pool runtime thread, so we just need to set up blocking.
CURRENT.with(|maybe_cx| {
match (crate::runtime::enter::context(), maybe_cx.is_some()) {
(EnterContext::Entered { .. }, true) => {
// We are on a thread pool runtime thread, so we just need to set up blocking.
had_entered = true;
}
(EnterContext::Entered { allow_blocking }, false) => {
// We are on an executor, but _not_ on the thread pool.
// That is _only_ okay if we are in a thread pool runtime's block_on method:
if allow_blocking {
had_entered = true;
}
(EnterContext::Entered { allow_blocking }, false) => {
// We are on an executor, but _not_ on the thread pool.
// That is _only_ okay if we are in a thread pool runtime's block_on method:
if allow_blocking {
had_entered = true;
return;
} else {
// This probably means we are on the basic_scheduler or in a LocalSet,
// where it is _not_ okay to block.
panic!("can call blocking only when running on the multi-threaded runtime");
}
}
(EnterContext::NotEntered, true) => {
// This is a nested call to block_in_place (we already exited).
// All the necessary setup has already been done.
return;
}
(EnterContext::NotEntered, false) => {
// We are outside of the tokio runtime, so blocking is fine.
// We can also skip all of the thread pool blocking setup steps.
return;
} else {
// This probably means we are on the basic_scheduler or in a LocalSet,
// where it is _not_ okay to block.
panic!("can call blocking only when running on the multi-threaded runtime");
}
}
let cx = maybe_cx.expect("no .is_some() == false cases above should lead here");
// Get the worker core. If none is set, then blocking is fine!
let core = match cx.core.borrow_mut().take() {
Some(core) => core,
None => return,
};
// The parker should be set here
assert!(core.park.is_some());
// In order to block, the core must be sent to another thread for
// execution.
//
// First, move the core back into the worker's shared core slot.
cx.worker.core.set(core);
// Next, clone the worker handle and send it to a new thread for
// processing.
//
// Once the blocking task is done executing, we will attempt to
// steal the core back.
let worker = cx.worker.clone();
runtime::spawn_blocking(move || run(worker));
});
if had_entered {
// Unset the current task's budget. Blocking sections are not
// constrained by task budgets.
let _reset = Reset(coop::stop());
crate::runtime::enter::exit(f)
} else {
f()
(EnterContext::NotEntered, true) => {
// This is a nested call to block_in_place (we already exited).
// All the necessary setup has already been done.
return;
}
(EnterContext::NotEntered, false) => {
// We are outside of the tokio runtime, so blocking is fine.
// We can also skip all of the thread pool blocking setup steps.
return;
}
}
let cx = maybe_cx.expect("no .is_some() == false cases above should lead here");
// Get the worker core. If none is set, then blocking is fine!
let core = match cx.core.borrow_mut().take() {
Some(core) => core,
None => return,
};
// The parker should be set here
assert!(core.park.is_some());
// In order to block, the core must be sent to another thread for
// execution.
//
// First, move the core back into the worker's shared core slot.
cx.worker.core.set(core);
// Next, clone the worker handle and send it to a new thread for
// processing.
//
// Once the blocking task is done executing, we will attempt to
// steal the core back.
let worker = cx.worker.clone();
runtime::spawn_blocking(move || run(worker));
});
if had_entered {
// Unset the current task's budget. Blocking sections are not
// constrained by task budgets.
let _reset = Reset(coop::stop());
crate::runtime::enter::exit(f)
} else {
f()
}
}
+1 -1
View File
@@ -306,7 +306,7 @@ mod tests {
}
fn rt() -> Runtime {
runtime::Builder::new().basic_scheduler().build().unwrap()
runtime::Builder::new_current_thread().build().unwrap()
}
async fn collect(mut rx: crate::sync::mpsc::Receiver<()>) -> Vec<()> {
+33 -12
View File
@@ -1,9 +1,10 @@
#![cfg_attr(not(feature = "rt-core"), allow(dead_code))]
//! Signal driver
use crate::io::driver::Driver as IoDriver;
use crate::io::PollEvented;
use crate::park::Park;
use crate::runtime::context;
use crate::signal::registry::globals;
use mio::net::UnixStream;
@@ -165,17 +166,6 @@ impl Park for Driver {
// ===== impl Handle =====
impl Handle {
/// Returns a handle to the current driver
///
/// # Panics
///
/// This function panics if there is no current signal driver set.
pub(super) fn current() -> Self {
context::signal_handle().expect(
"there is no signal driver running, must be called from the context of Tokio runtime",
)
}
pub(super) fn check_inner(&self) -> io::Result<()> {
if self.inner.strong_count() > 0 {
Ok(())
@@ -184,3 +174,34 @@ impl Handle {
}
}
}
cfg_rt_core! {
impl Handle {
/// Returns a handle to the current driver
///
/// # Panics
///
/// This function panics if there is no current signal driver set.
pub(super) fn current() -> Self {
crate::runtime::context::signal_handle().expect(
"there is no signal driver running, must be called from the context of Tokio runtime",
)
}
}
}
cfg_not_rt_core! {
impl Handle {
/// Returns a handle to the current driver
///
/// # Panics
///
/// This function panics if there is no current signal driver set.
pub(super) fn current() -> Self {
panic!(
"there is no signal driver running, must be called from the context of Tokio runtime or with\
`rt-core` enabled.",
)
}
}
}
+1 -2
View File
@@ -289,8 +289,7 @@ mod tests {
}
fn rt() -> Runtime {
crate::runtime::Builder::new()
.basic_scheduler()
crate::runtime::Builder::new_current_thread()
.build()
.unwrap()
}
+1 -1
View File
@@ -267,7 +267,7 @@ pub trait StreamExt: Stream {
/// # /*
/// #[tokio::main]
/// # */
/// # #[tokio::main(basic_scheduler)]
/// # #[tokio::main(flavor = "current_thread")]
/// async fn main() {
/// # time::pause();
/// let (tx1, rx1) = mpsc::channel(10);
+3 -3
View File
@@ -437,7 +437,7 @@ cfg_sync! {
mod mutex;
pub use mutex::{Mutex, MutexGuard, TryLockError, OwnedMutexGuard};
mod notify;
pub(crate) mod notify;
pub use notify::Notify;
pub mod oneshot;
@@ -464,8 +464,8 @@ cfg_not_sync! {
pub(crate) use mutex::Mutex;
}
mod notify;
pub(crate) use notify::Notify;
#[cfg(any(feature = "rt-core", feature = "signal", all(unix, feature = "process")))]
pub(crate) mod notify;
cfg_atomic_waker_impl! {
mod task;
+4 -4
View File
@@ -178,9 +178,9 @@ impl<T> Receiver<T> {
/// sync_code.join().unwrap()
/// }
/// ```
#[cfg(feature = "sync")]
pub fn blocking_recv(&mut self) -> Option<T> {
let mut enter_handle = crate::runtime::enter::enter(false);
enter_handle.block_on(self.recv()).unwrap()
crate::future::block_on(self.recv())
}
/// Attempts to return a pending value on this receiver without blocking.
@@ -518,9 +518,9 @@ impl<T> Sender<T> {
/// sync_code.join().unwrap()
/// }
/// ```
#[cfg(feature = "sync")]
pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>> {
let mut enter_handle = crate::runtime::enter::enter(false);
enter_handle.block_on(self.send(value)).unwrap()
crate::future::block_on(self.send(value))
}
/// Checks if the channel has been closed. This happens when the
+1 -1
View File
@@ -4,7 +4,7 @@ use crate::loom::sync::atomic::AtomicUsize;
use crate::loom::sync::Arc;
use crate::sync::mpsc::error::TryRecvError;
use crate::sync::mpsc::list;
use crate::sync::Notify;
use crate::sync::notify::Notify;
use std::fmt;
use std::process;
+74 -79
View File
@@ -17,7 +17,7 @@ cfg_rt_threaded! {
/// using the [`join!`] macro. To avoid this issue, use [`spawn_blocking`]
/// instead.
///
/// Note that this function can only be used on the [threaded scheduler].
/// Note that this function can only be used when using the `multi_thread` runtime.
///
/// Code running behind `block_in_place` cannot be cancelled. When you shut
/// down the executor, it will wait indefinitely for all blocking operations
@@ -27,7 +27,6 @@ cfg_rt_threaded! {
/// returns.
///
/// [blocking]: ../index.html#cpu-bound-tasks-and-blocking-code
/// [threaded scheduler]: fn@crate::runtime::Builder::threaded_scheduler
/// [`spawn_blocking`]: fn@crate::task::spawn_blocking
/// [`join!`]: macro@join
/// [`thread::spawn`]: fn@std::thread::spawn
@@ -44,7 +43,6 @@ cfg_rt_threaded! {
/// });
/// # }
/// ```
#[cfg_attr(docsrs, doc(cfg(feature = "blocking")))]
pub fn block_in_place<F, R>(f: F) -> R
where
F: FnOnce() -> R,
@@ -53,80 +51,77 @@ cfg_rt_threaded! {
}
}
cfg_blocking! {
/// Runs 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. This function runs the provided closure
/// on a thread dedicated to blocking operations. See the [CPU-bound tasks
/// and blocking code][blocking] section for more information.
///
/// Tokio will spawn more blocking threads when they are requested through
/// this function until the upper limit configured on the [`Builder`] is
/// reached. This limit is very large by default, because `spawn_blocking` is
/// often used for various kinds of IO operations that cannot be performed
/// asynchronously. When you run CPU-bound code using `spawn_blocking`, you
/// should keep this large upper limit in mind; to run your CPU-bound
/// computations on only a few threads, you should use a separate thread
/// pool such as [rayon] rather than configuring the number of blocking
/// threads.
///
/// This function is intended for non-async operations that eventually
/// finish on their own. If you want to spawn an ordinary thread, you should
/// use [`thread::spawn`] instead.
///
/// Closures spawned using `spawn_blocking` cannot be cancelled. When you
/// shut down the executor, it will wait indefinitely for all blocking
/// operations to finish. You can use [`shutdown_timeout`] to stop waiting
/// for them after a certain timeout. Be aware that this will still not
/// cancel the tasks — they are simply allowed to keep running after the
/// method returns.
///
/// Note that if you are using the [basic scheduler], this function will
/// still spawn additional threads for blocking operations. The basic
/// scheduler's single thread is only used for asynchronous code.
///
/// [`Builder`]: struct@crate::runtime::Builder
/// [blocking]: ../index.html#cpu-bound-tasks-and-blocking-code
/// [rayon]: https://docs.rs/rayon
/// [basic scheduler]: fn@crate::runtime::Builder::basic_scheduler
/// [`thread::spawn`]: fn@std::thread::spawn
/// [`shutdown_timeout`]: fn@crate::runtime::Runtime::shutdown_timeout
///
/// # 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,
{
#[cfg(feature = "tracing")]
let f = {
let span = tracing::trace_span!(
target: "tokio::task",
"task",
kind = %"blocking",
function = %std::any::type_name::<F>(),
);
move || {
let _g = span.enter();
f()
}
};
crate::runtime::spawn_blocking(f)
}
/// Runs 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. This function runs the provided closure
/// on a thread dedicated to blocking operations. See the [CPU-bound tasks
/// and blocking code][blocking] section for more information.
///
/// Tokio will spawn more blocking threads when they are requested through
/// this function until the upper limit configured on the [`Builder`] is
/// reached. This limit is very large by default, because `spawn_blocking` is
/// often used for various kinds of IO operations that cannot be performed
/// asynchronously. When you run CPU-bound code using `spawn_blocking`, you
/// should keep this large upper limit in mind; to run your CPU-bound
/// computations on only a few threads, you should use a separate thread
/// pool such as [rayon] rather than configuring the number of blocking
/// threads.
///
/// This function is intended for non-async operations that eventually
/// finish on their own. If you want to spawn an ordinary thread, you should
/// use [`thread::spawn`] instead.
///
/// Closures spawned using `spawn_blocking` cannot be cancelled. When you
/// shut down the executor, it will wait indefinitely for all blocking
/// operations to finish. You can use [`shutdown_timeout`] to stop waiting
/// for them after a certain timeout. Be aware that this will still not
/// cancel the tasks — they are simply allowed to keep running after the
/// method returns.
///
/// Note that if you are using the single threaded runtime, this function will
/// still spawn additional threads for blocking operations. The basic
/// scheduler's single thread is only used for asynchronous code.
///
/// [`Builder`]: struct@crate::runtime::Builder
/// [blocking]: ../index.html#cpu-bound-tasks-and-blocking-code
/// [rayon]: https://docs.rs/rayon
/// [`thread::spawn`]: fn@std::thread::spawn
/// [`shutdown_timeout`]: fn@crate::runtime::Runtime::shutdown_timeout
///
/// # 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,
{
#[cfg(feature = "tracing")]
let f = {
let span = tracing::trace_span!(
target: "tokio::task",
"task",
kind = %"blocking",
function = %std::any::type_name::<F>(),
);
move || {
let _g = span.enter();
f()
}
};
crate::runtime::spawn_blocking(f)
}
+2
View File
@@ -346,6 +346,8 @@ impl LocalSet {
/// [`Runtime::block_on`]: method@crate::runtime::Runtime::block_on
/// [in-place blocking]: fn@crate::task::block_in_place
/// [`spawn_blocking`]: fn@crate::task::spawn_blocking
#[cfg(feature = "rt-core")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-core")))]
pub fn block_on<F>(&self, rt: &crate::runtime::Runtime, future: F) -> F::Output
where
F: Future,
+10 -9
View File
@@ -214,26 +214,27 @@
//! [rt-threaded]: ../runtime/index.html#threaded-scheduler
//! [`task::yield_now`]: crate::task::yield_now()
//! [`thread::yield_now`]: std::thread::yield_now
cfg_blocking! {
cfg_task! {
pub use crate::runtime::task::{JoinError, JoinHandle};
}
cfg_rt_core! {
mod blocking;
pub use blocking::spawn_blocking;
mod spawn;
pub use spawn::spawn;
cfg_rt_threaded! {
pub use blocking::block_in_place;
}
}
cfg_rt_core! {
pub use crate::runtime::task::{JoinError, JoinHandle};
mod spawn;
pub use spawn::spawn;
cfg_rt_util! {
mod yield_now;
pub use yield_now::yield_now;
}
cfg_rt_util! {
mod local;
pub use local::{spawn_local, LocalSet};
+1 -1
View File
@@ -2,7 +2,7 @@ use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
doc_rt_core! {
cfg_rt_util! {
/// Yields execution back to the Tokio runtime.
///
/// A task yields by awaiting on `yield_now()`, and may resume when that
+32 -6
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "rt-core"), allow(dead_code))]
//! Source of time abstraction.
//!
//! By default, `std::time::Instant::now()` is used. However, when the
@@ -36,7 +38,18 @@ cfg_not_test_util! {
cfg_test_util! {
use crate::time::{Duration, Instant};
use std::sync::{Arc, Mutex};
use crate::runtime::context;
cfg_rt_core! {
fn clock() -> Option<Clock> {
crate::runtime::context::clock()
}
}
cfg_not_rt_core! {
fn clock() -> Option<Clock> {
None
}
}
/// A handle to a source of time.
#[derive(Debug, Clone)]
@@ -65,7 +78,7 @@ cfg_test_util! {
/// Panics if time is already frozen or if called from outside of the Tokio
/// runtime.
pub fn pause() {
let clock = context::clock().expect("time cannot be frozen from outside the Tokio runtime");
let clock = clock().expect("time cannot be frozen from outside the Tokio runtime");
clock.pause();
}
@@ -79,7 +92,7 @@ cfg_test_util! {
/// Panics if time is not frozen or if called from outside of the Tokio
/// runtime.
pub fn resume() {
let clock = context::clock().expect("time cannot be frozen from outside the Tokio runtime");
let clock = clock().expect("time cannot be frozen from outside the Tokio runtime");
let mut inner = clock.inner.lock().unwrap();
if inner.unfrozen.is_some() {
@@ -99,14 +112,27 @@ cfg_test_util! {
/// Panics if time is not frozen or if called from outside of the Tokio
/// runtime.
pub async fn advance(duration: Duration) {
let clock = context::clock().expect("time cannot be frozen from outside the Tokio runtime");
use crate::future::poll_fn;
use std::task::Poll;
let clock = clock().expect("time cannot be frozen from outside the Tokio runtime");
clock.advance(duration);
crate::task::yield_now().await;
let mut yielded = false;
poll_fn(|cx| {
if yielded {
Poll::Ready(())
} else {
yielded = true;
cx.waker().wake_by_ref();
Poll::Pending
}
}).await;
}
/// Return the current instant, factoring in frozen time.
pub(crate) fn now() -> Instant {
if let Some(clock) = context::clock() {
if let Some(clock) = clock() {
clock.now()
} else {
Instant::from_std(std::time::Instant::now())
+50 -22
View File
@@ -1,4 +1,3 @@
use crate::runtime::context;
use crate::time::driver::Inner;
use std::fmt;
use std::sync::{Arc, Weak};
@@ -15,33 +14,62 @@ impl Handle {
Handle { inner }
}
/// Tries to get a handle to the current timer.
///
/// # Panics
///
/// This function panics if there is no current timer set.
///
/// It can be triggered when `Builder::enable_time()` or
/// `Builder::enable_all()` are not included in the builder.
///
/// It can also panic whenever a timer is created outside of a Tokio
/// runtime. That is why `rt.block_on(sleep(...))` will panic,
/// since the function is executed outside of the runtime.
/// Whereas `rt.block_on(async {sleep(...).await})` doesn't
/// panic. And this is because wrapping the function on an async makes it
/// lazy, and so gets executed inside the runtime successfuly without
/// panicking.
pub(crate) fn current() -> Self {
context::time_handle()
.expect("there is no timer running, must be called from the context of Tokio runtime")
}
/// Tries to return a strong ref to the inner
pub(crate) fn inner(&self) -> Option<Arc<Inner>> {
self.inner.upgrade()
}
}
cfg_rt_core! {
impl Handle {
/// Tries to get a handle to the current timer.
///
/// # Panics
///
/// This function panics if there is no current timer set.
///
/// It can be triggered when `Builder::enable_time()` or
/// `Builder::enable_all()` are not included in the builder.
///
/// It can also panic whenever a timer is created outside of a Tokio
/// runtime. That is why `rt.block_on(delay_for(...))` will panic,
/// since the function is executed outside of the runtime.
/// Whereas `rt.block_on(async {delay_for(...).await})` doesn't
/// panic. And this is because wrapping the function on an async makes it
/// lazy, and so gets executed inside the runtime successfuly without
/// panicking.
pub(crate) fn current() -> Self {
crate::runtime::context::time_handle()
.expect("there is no timer running, must be called from the context of Tokio runtime")
}
}
}
cfg_not_rt_core! {
impl Handle {
/// Tries to get a handle to the current timer.
///
/// # Panics
///
/// This function panics if there is no current timer set.
///
/// It can be triggered when `Builder::enable_time()` or
/// `Builder::enable_all()` are not included in the builder.
///
/// It can also panic whenever a timer is created outside of a Tokio
/// runtime. That is why `rt.block_on(delay_for(...))` will panic,
/// since the function is executed outside of the runtime.
/// Whereas `rt.block_on(async {delay_for(...).await})` doesn't
/// panic. And this is because wrapping the function on an async makes it
/// lazy, and so gets executed inside the runtime successfuly without
/// panicking.
pub(crate) fn current() -> Self {
panic!("there is no timer running, must be called from the context of Tokio runtime or \
`rt-core` is not enabled")
}
}
}
impl fmt::Debug for Handle {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Handle")
+2
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "rt-core"), allow(dead_code))]
//! Time driver
mod atomic_stack;
+2
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "full"), allow(dead_code))]
//! An intrusive double linked list of data
//!
//! The data structure supports tracking pinned nodes. Most of the data
+13 -3
View File
@@ -3,16 +3,26 @@ cfg_io_driver! {
pub(crate) mod slab;
}
#[cfg(any(
feature = "fs",
feature = "process",
feature = "rt-core",
feature = "rt-util",
feature = "sync",
feature = "signal",
feature = "tcp",
feature = "udp",
feature = "uds",
))]
pub(crate) mod linked_list;
#[cfg(any(feature = "rt-threaded", feature = "macros", feature = "stream"))]
mod rand;
mod wake;
pub(crate) use wake::{waker_ref, Wake};
cfg_rt_core! {
mod wake;
pub(crate) use wake::WakerRef;
pub(crate) use wake::{waker_ref, Wake};
}
cfg_rt_threaded! {
+2
View File
@@ -1,3 +1,5 @@
#![cfg_attr(not(feature = "rt-core"), allow(dead_code))]
use crate::loom::cell::UnsafeCell;
use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
use crate::loom::sync::{Arc, Mutex};
+2 -2
View File
@@ -1,5 +1,5 @@
cfg_trace! {
cfg_rt_core! {
cfg_task! {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -47,7 +47,7 @@ cfg_trace! {
}
cfg_not_trace! {
cfg_rt_core! {
cfg_task! {
#[inline]
pub(crate) fn task<F>(task: F, _: &'static str) -> F {
// nop
+1 -2
View File
@@ -45,8 +45,7 @@ fn test_drop_on_notify() {
// shutting down. Then, when the task handle is dropped, the task itself is
// dropped.
let rt = runtime::Builder::new()
.basic_scheduler()
let rt = runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
+1 -2
View File
@@ -45,8 +45,7 @@ fn drop_wakes() {
}
fn rt() -> runtime::Runtime {
runtime::Builder::new()
.basic_scheduler()
runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
+1 -2
View File
@@ -129,8 +129,7 @@ fn acquire_mutex_in_drop() {
}
fn rt() -> Runtime {
tokio::runtime::Builder::new()
.basic_scheduler()
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
+6 -9
View File
@@ -6,12 +6,11 @@
macro_rules! rt_test {
($($t:tt)*) => {
mod basic_scheduler {
mod current_thread_scheduler {
$($t)*
fn rt() -> Arc<Runtime> {
tokio::runtime::Builder::new()
.basic_scheduler()
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
@@ -23,9 +22,8 @@ macro_rules! rt_test {
$($t)*
fn rt() -> Arc<Runtime> {
tokio::runtime::Builder::new()
.threaded_scheduler()
.core_threads(4)
tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.enable_all()
.build()
.unwrap()
@@ -37,9 +35,8 @@ macro_rules! rt_test {
$($t)*
fn rt() -> Arc<Runtime> {
tokio::runtime::Builder::new()
.threaded_scheduler()
.core_threads(1)
tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.unwrap()
+6 -11
View File
@@ -18,10 +18,9 @@ use std::task::{Context, Poll};
#[test]
fn single_thread() {
// No panic when starting a runtime w/ a single thread
let _ = runtime::Builder::new()
.threaded_scheduler()
let _ = runtime::Builder::new_multi_thread()
.enable_all()
.core_threads(1)
.worker_threads(1)
.build();
}
@@ -188,8 +187,7 @@ fn drop_threadpool_drops_futures() {
let a = num_inc.clone();
let b = num_dec.clone();
let rt = runtime::Builder::new()
.threaded_scheduler()
let rt = runtime::Builder::new_multi_thread()
.enable_all()
.on_thread_start(move || {
a.fetch_add(1, Relaxed);
@@ -228,8 +226,7 @@ fn start_stop_callbacks_called() {
let after_inner = after_start.clone();
let before_inner = before_stop.clone();
let rt = tokio::runtime::Builder::new()
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.on_thread_start(move || {
after_inner.clone().fetch_add(1, Ordering::Relaxed);
@@ -329,8 +326,7 @@ fn multi_threadpool() {
// channel yields occasionally even if there are values ready to receive.
#[test]
fn coop_and_block_in_place() {
let rt = tokio::runtime::Builder::new()
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
// Setting max threads to 1 prevents another thread from claiming the
// runtime worker yielded as part of `block_in_place` and guarantees the
// same thread will reclaim the worker at the end of the
@@ -380,8 +376,7 @@ fn coop_and_block_in_place() {
// Testing this does not panic
#[test]
fn max_threads() {
let _rt = tokio::runtime::Builder::new()
.threaded_scheduler()
let _rt = tokio::runtime::Builder::new_multi_thread()
.max_threads(1)
.build()
.unwrap();
+1 -2
View File
@@ -37,8 +37,7 @@ fn dropping_loops_does_not_cause_starvation() {
}
fn rt() -> Runtime {
tokio::runtime::Builder::new()
.basic_scheduler()
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
+1 -2
View File
@@ -47,8 +47,7 @@ fn multi_loop() {
}
fn rt() -> Runtime {
tokio::runtime::Builder::new()
.basic_scheduler()
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
+1 -1
View File
@@ -166,7 +166,7 @@ async fn write_order() {
}
// A single RwLock is contested by tasks in multiple threads
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn multithreaded() {
let barrier = Arc::new(Barrier::new(5));
let rwlock = Arc::new(RwLock::<u32>::new(0));
+15 -38
View File
@@ -28,7 +28,7 @@ async fn basic_blocking() {
}
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn block_in_blocking() {
// Run a few times
for _ in 0..100 {
@@ -51,7 +51,7 @@ async fn block_in_blocking() {
}
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn block_in_block() {
// Run a few times
for _ in 0..100 {
@@ -71,7 +71,7 @@ async fn block_in_block() {
}
}
#[tokio::test(basic_scheduler)]
#[tokio::test(flavor = "current_thread")]
#[should_panic]
async fn no_block_in_basic_scheduler() {
task::block_in_place(|| {});
@@ -79,10 +79,7 @@ async fn no_block_in_basic_scheduler() {
#[test]
fn yes_block_in_threaded_block_on() {
let rt = runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let rt = runtime::Runtime::new().unwrap();
rt.block_on(async {
task::block_in_place(|| {});
});
@@ -91,7 +88,7 @@ fn yes_block_in_threaded_block_on() {
#[test]
#[should_panic]
fn no_block_in_basic_block_on() {
let rt = runtime::Builder::new().basic_scheduler().build().unwrap();
let rt = runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
task::block_in_place(|| {});
});
@@ -99,15 +96,11 @@ fn no_block_in_basic_block_on() {
#[test]
fn can_enter_basic_rt_from_within_block_in_place() {
let outer = tokio::runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let outer = tokio::runtime::Runtime::new().unwrap();
outer.block_on(async {
tokio::task::block_in_place(|| {
let inner = tokio::runtime::Builder::new()
.basic_scheduler()
let inner = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
@@ -120,15 +113,11 @@ fn can_enter_basic_rt_from_within_block_in_place() {
fn useful_panic_message_when_dropping_rt_in_rt() {
use std::panic::{catch_unwind, AssertUnwindSafe};
let outer = tokio::runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let outer = tokio::runtime::Runtime::new().unwrap();
let result = catch_unwind(AssertUnwindSafe(|| {
outer.block_on(async {
let _ = tokio::runtime::Builder::new()
.basic_scheduler()
let _ = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
});
@@ -147,14 +136,10 @@ fn useful_panic_message_when_dropping_rt_in_rt() {
#[test]
fn can_shutdown_with_zero_timeout_in_runtime() {
let outer = tokio::runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let outer = tokio::runtime::Runtime::new().unwrap();
outer.block_on(async {
let rt = tokio::runtime::Builder::new()
.basic_scheduler()
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
rt.shutdown_timeout(Duration::from_nanos(0));
@@ -163,14 +148,10 @@ fn can_shutdown_with_zero_timeout_in_runtime() {
#[test]
fn can_shutdown_now_in_runtime() {
let outer = tokio::runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let outer = tokio::runtime::Runtime::new().unwrap();
outer.block_on(async {
let rt = tokio::runtime::Builder::new()
.basic_scheduler()
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
rt.shutdown_background();
@@ -179,8 +160,7 @@ fn can_shutdown_now_in_runtime() {
#[test]
fn coop_disabled_in_block_in_place() {
let outer = tokio::runtime::Builder::new()
.threaded_scheduler()
let outer = tokio::runtime::Builder::new_multi_thread()
.enable_time()
.build()
.unwrap();
@@ -213,10 +193,7 @@ fn coop_disabled_in_block_in_place_in_block_on() {
let (done_tx, done_rx) = std::sync::mpsc::channel();
let done = done_tx.clone();
thread::spawn(move || {
let outer = tokio::runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let outer = tokio::runtime::Runtime::new().unwrap();
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
for i in 0..200 {
+1 -1
View File
@@ -3,7 +3,7 @@ tokio::task_local! {
pub static FOO: bool;
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn local() {
let j1 = tokio::spawn(REQ_ID.scope(1, async move {
assert_eq!(REQ_ID.get(), 1);
+11 -16
View File
@@ -11,7 +11,7 @@ use std::sync::atomic::Ordering::{self, SeqCst};
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::time::Duration;
#[tokio::test(basic_scheduler)]
#[tokio::test(flavor = "current_thread")]
async fn local_basic_scheduler() {
LocalSet::new()
.run_until(async {
@@ -20,7 +20,7 @@ async fn local_basic_scheduler() {
.await;
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn local_threadpool() {
thread_local! {
static ON_RT_THREAD: Cell<bool> = Cell::new(false);
@@ -40,7 +40,7 @@ async fn local_threadpool() {
.await;
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn localset_future_threadpool() {
thread_local! {
static ON_LOCAL_THREAD: Cell<bool> = Cell::new(false);
@@ -55,7 +55,7 @@ async fn localset_future_threadpool() {
local.await;
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn localset_future_timers() {
static RAN1: AtomicBool = AtomicBool::new(false);
static RAN2: AtomicBool = AtomicBool::new(false);
@@ -99,7 +99,7 @@ async fn localset_future_drives_all_local_futs() {
assert!(RAN3.load(Ordering::SeqCst));
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn local_threadpool_timer() {
// This test ensures that runtime services like the timer are properly
// set for the local task set.
@@ -133,8 +133,7 @@ fn local_threadpool_blocking_in_place() {
ON_RT_THREAD.with(|cell| cell.set(true));
let rt = runtime::Builder::new()
.threaded_scheduler()
let rt = runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
@@ -149,7 +148,7 @@ fn local_threadpool_blocking_in_place() {
});
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn local_threadpool_blocking_run() {
thread_local! {
static ON_RT_THREAD: Cell<bool> = Cell::new(false);
@@ -177,7 +176,7 @@ async fn local_threadpool_blocking_run() {
.await;
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn all_spawns_are_local() {
use futures::future;
thread_local! {
@@ -203,7 +202,7 @@ async fn all_spawns_are_local() {
.await;
}
#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn nested_spawn_is_local() {
thread_local! {
static ON_RT_THREAD: Cell<bool> = Cell::new(false);
@@ -246,10 +245,7 @@ fn join_local_future_elsewhere() {
ON_RT_THREAD.with(|cell| cell.set(true));
let rt = runtime::Builder::new()
.threaded_scheduler()
.build()
.unwrap();
let rt = runtime::Runtime::new().unwrap();
let local = LocalSet::new();
local.block_on(&rt, async move {
let (tx, rx) = oneshot::channel();
@@ -491,8 +487,7 @@ async fn acquire_mutex_in_drop() {
}
fn rt() -> Runtime {
tokio::runtime::Builder::new()
.basic_scheduler()
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
+1 -5
View File
@@ -28,11 +28,7 @@ fn timer_with_threaded_runtime() {
fn timer_with_basic_scheduler() {
use tokio::runtime::Builder;
let rt = Builder::new()
.basic_scheduler()
.enable_all()
.build()
.unwrap();
let rt = Builder::new_current_thread().enable_all().build().unwrap();
let (tx, rx) = mpsc::channel();
rt.block_on(async move {