tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-28 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into rt-threaded-rework

Browse Source
This commit is contained in:
Carl Lerche
2023-06-21 01:24:43 +00:00
parent 6546b21581 56c4365584
commit cbacaa3e4f
29 changed files with 360 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+51 -31
View File
@@ -42,7 +42,6 @@ jobs:
- test-unstable
- miri
- asan
- semver
- cross-check
- cross-test
- no-atomic-u64
@@ -56,6 +55,7 @@ jobs:
- check-readme
- test-hyper
- x86_64-fortanix-unknown-sgx
- check-redox
- wasm32-unknown-unknown
- wasm32-wasi
- check-external-types
@@ -208,7 +208,7 @@ jobs:
# in order to run doctests for unstable features, we must also pass
# the unstable cfg to RustDoc
RUSTDOCFLAGS: --cfg tokio_unstable
test-unstable-taskdump:
name: test tokio full --unstable --taskdump
needs: basics
@@ -300,17 +300,19 @@ jobs:
# Ignore `trybuild` errors as they are irrelevant and flaky on nightly
TRYBUILD: overwrite
semver:
name: semver
needs: basics
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Check semver
uses: obi1kenobi/cargo-semver-checks-action@v2
with:
rust-toolchain: ${{ env.rust_stable }}
release-type: minor
# Re-enable this after the next release.
#
#semver:
# name: semver
# needs: basics
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v3
# - name: Check semver
# uses: obi1kenobi/cargo-semver-checks-action@v2
# with:
# rust-toolchain: ${{ env.rust_stable }}
# release-type: minor
cross-check:
name: cross-check
@@ -398,10 +400,10 @@ jobs:
RUSTFLAGS: --cfg tokio_unstable --cfg tokio_taskdump -Dwarnings --cfg tokio_no_atomic_u64
# https://github.com/tokio-rs/tokio/pull/5356
# https://github.com/tokio-rs/tokio/issues/5373
- run: cargo hack build -p tokio --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
- run: cargo hack build -p tokio --feature-powerset --depth 2 --keep-going
env:
RUSTFLAGS: --cfg tokio_unstable --cfg tokio_taskdump -Dwarnings --cfg tokio_no_atomic_u64 --cfg tokio_no_const_mutex_new
- run: cargo hack build -p tokio --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
- run: cargo hack build -p tokio --feature-powerset --depth 2 --keep-going
env:
RUSTFLAGS: --cfg tokio_unstable --cfg tokio_taskdump -Dwarnings --cfg tokio_no_atomic_u64
@@ -420,15 +422,15 @@ jobs:
- name: Install cargo-hack
uses: taiki-e/install-action@cargo-hack
- name: check --feature-powerset
run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
run: cargo hack check --all --feature-powerset --depth 2 --keep-going
# Try with unstable feature flags
- name: check --feature-powerset --unstable
run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
run: cargo hack check --all --feature-powerset --depth 2 --keep-going
env:
RUSTFLAGS: --cfg tokio_unstable -Dwarnings
# Try with unstable and taskdump feature flags
- name: check --feature-powerset --unstable --taskdump
run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
run: cargo hack check --all --feature-powerset --depth 2 --keep-going
env:
RUSTFLAGS: --cfg tokio_unstable --cfg tokio_taskdump -Dwarnings
@@ -494,8 +496,8 @@ jobs:
- name: "rustfmt --check"
# Workaround for rust-lang/cargo#7732
run: |
if ! rustfmt --check --edition 2018 $(git ls-files '*.rs'); then
printf "Please run \`rustfmt --edition 2018 \$(git ls-files '*.rs')\` to fix rustfmt errors.\nSee CONTRIBUTING.md for more details.\n" >&2
if ! rustfmt --check --edition 2021 $(git ls-files '*.rs'); then
printf "Please run \`rustfmt --edition 2021 \$(git ls-files '*.rs')\` to fix rustfmt errors.\nSee CONTRIBUTING.md for more details.\n" >&2
exit 1
fi
@@ -610,6 +612,21 @@ jobs:
run: cargo build --target x86_64-fortanix-unknown-sgx --features rt,sync
working-directory: tokio
check-redox:
name: build tokio for redox-os
needs: basics
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust ${{ env.rust_nightly }}
uses: dtolnay/rust-toolchain@master
with:
toolchain: ${{ env.rust_nightly }}
target: x86_64-unknown-redox
- name: check tokio on redox
run: cargo check --target x86_64-unknown-redox --all-features
working-directory: tokio
wasm32-unknown-unknown:
name: test tokio for wasm32-unknown-unknown
needs: basics
@@ -676,7 +693,7 @@ jobs:
working-directory: tests-integration
check-external-types:
name: check-external-types
name: check-external-types (${{ matrix.os }})
needs: basics
runs-on: ${{ matrix.os }}
strategy:
@@ -684,20 +701,23 @@ jobs:
os:
- windows-latest
- ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Rust nightly-2022-11-16
uses: dtolnay/rust-toolchain@master
with:
rust:
# `check-external-types` requires a specific Rust nightly version. See
# the README for details: https://github.com/awslabs/cargo-check-external-types
toolchain: nightly-2022-11-16
- nightly-2023-05-31
steps:
- uses: actions/checkout@v3
- name: Install Rust ${{ matrix.rust }}
uses: dtolnay/rust-toolchain@master
with:
toolchain: ${{ matrix.rust }}
- uses: Swatinem/rust-cache@v2
- name: Install cargo-check-external-types
uses: taiki-e/cache-cargo-install-action@v1
with:
tool: [email protected]
- name: check-external-types
run: |
set -x
cargo install cargo-check-external-types --locked --version 0.1.6
cargo check-external-types --all-features --config external-types.toml
run: cargo check-external-types --all-features --config external-types.toml
working-directory: tokio
check-fuzzing:
CONTRIBUTING.md
+4 -4
View File
@@ -173,10 +173,10 @@ command below instead:
```
# Mac or Linux
rustfmt --check --edition 2018 $(git ls-files '*.rs')
rustfmt --check --edition 2021 $(git ls-files '*.rs')
# Powershell
Get-ChildItem . -Filter "*.rs" -Recurse | foreach { rustfmt --check --edition 2018 $_.FullName }
Get-ChildItem . -Filter "*.rs" -Recurse | foreach { rustfmt --check --edition 2021 $_.FullName }
```
The `--check` argument prints the things that need to be fixed. If you remove
it, `rustfmt` will update your files locally instead.
@@ -230,7 +230,7 @@ integration tests in the crate and follow the style.
Some of our crates include a set of fuzz tests, this will be marked by a
directory `fuzz`. It is a good idea to run fuzz tests after each change.
To get started with fuzz testing you'll need to install
To get started with fuzz testing you'll need to install
[cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz).
`cargo install cargo-fuzz`
@@ -678,4 +678,4 @@ When releasing a new version of a crate, follow these steps:
[unit-tests]: https://doc.rust-lang.org/rust-by-example/testing/unit_testing.html
[integration-tests]: https://doc.rust-lang.org/rust-by-example/testing/integration_testing.html
[documentation-tests]: https://doc.rust-lang.org/rust-by-example/testing/doc_testing.html
[conditional-compilation]: https://doc.rust-lang.org/reference/conditional-compilation.html
[conditional-compilation]: https://doc.rust-lang.org/reference/conditional-compilation.html
Cargo.toml
+1 -1
View File
@@ -1,5 +1,5 @@
[workspace]
resolver = "2"
members = [
"tokio",
"tokio-macros",
benches/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "benches"
version = "0.0.0"
publish = false
edition = "2018"
edition = "2021"
[features]
test-util = ["tokio/test-util"]
examples/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "examples"
version = "0.0.0"
publish = false
edition = "2018"
edition = "2021"
# If you copy one of the examples into a new project, you should be using
# [dependencies] instead, and delete the **path**.
examples/dump.rs
+54 -22
View File
@@ -1,4 +1,6 @@
//! This example demonstrates tokio's experimental taskdumping functionality.
//! This example demonstrates tokio's experimental task dumping functionality.
//! This application deadlocks. Input CTRL+C to display traces of each task, or
//! input CTRL+C twice within 1 second to quit.
#[cfg(all(
tokio_unstable,
@@ -7,44 +9,74 @@
any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
))]
#[tokio::main]
async fn main() {
use std::hint::black_box;
async fn main() -> Result<(), Box<dyn std::error::Error>> {
use std::sync::Arc;
use tokio::sync::Barrier;
#[inline(never)]
async fn a() {
black_box(b()).await
async fn a(barrier: Arc<Barrier>) {
b(barrier).await
}
#[inline(never)]
async fn b() {
black_box(c()).await
async fn b(barrier: Arc<Barrier>) {
c(barrier).await
}
#[inline(never)]
async fn c() {
loop {
tokio::task::yield_now().await;
}
async fn c(barrier: Arc<Barrier>) {
barrier.wait().await;
}
async fn dump() {
// Prints a task dump upon receipt of CTRL+C, or returns if CTRL+C is
// inputted twice within a second.
async fn dump_or_quit() {
use tokio::time::{timeout, Duration, Instant};
let handle = tokio::runtime::Handle::current();
let dump = handle.dump().await;
let mut last_signal: Option<Instant> = None;
// wait for CTRL+C
while let Ok(_) = tokio::signal::ctrl_c().await {
// exit if a CTRL+C is inputted twice within 1 second
if let Some(time_since_last_signal) = last_signal.map(|i| i.elapsed()) {
if time_since_last_signal < Duration::from_secs(1) {
return;
}
}
last_signal = Some(Instant::now());
for (i, task) in dump.tasks().iter().enumerate() {
let trace = task.trace();
println!("task {i} trace:");
println!("{trace}\n");
// capture a dump, and print each trace
println!("{:-<80}", "");
if let Ok(dump) = timeout(Duration::from_secs(2), handle.dump()).await {
for (i, task) in dump.tasks().iter().enumerate() {
let trace = task.trace();
println!("TASK {i}:");
println!("{trace}\n");
}
} else {
println!("Task dumping timed out. Use a native debugger (like gdb) to debug the deadlock.");
}
println!("{:-<80}", "");
println!("Input CTRL+C twice within 1 second to exit.");
}
}
println!("This program has a deadlock.");
println!("Input CTRL+C to print a task dump.");
println!("Input CTRL+C twice within 1 second to exit.");
// oops! this barrier waits for one more task than will ever come.
let barrier = Arc::new(Barrier::new(3));
let task_1 = tokio::spawn(a(barrier.clone()));
let task_2 = tokio::spawn(a(barrier));
tokio::select!(
biased;
_ = tokio::spawn(a()) => {},
_ = tokio::spawn(b()) => {},
_ = tokio::spawn(c()) => {},
_ = dump() => {},
_ = dump_or_quit() => {},
_ = task_1 => {},
_ = task_2 => {},
);
Ok(())
}
#[cfg(not(all(
examples/tinyhttp.rs
+1 -1
View File
@@ -18,7 +18,7 @@ use futures::SinkExt;
use http::{header::HeaderValue, Request, Response, StatusCode};
#[macro_use]
extern crate serde_derive;
use std::{convert::TryFrom, env, error::Error, fmt, io};
use std::{env, error::Error, fmt, io};
use tokio::net::{TcpListener, TcpStream};
use tokio_stream::StreamExt;
use tokio_util::codec::{Decoder, Encoder, Framed};
stress-test/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "stress-test"
version = "0.1.0"
authors = ["Tokio Contributors <[email protected]>"]
edition = "2018"
edition = "2021"
publish = false
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
tests-build/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "tests-build"
version = "0.1.0"
authors = ["Tokio Contributors <[email protected]>"]
edition = "2018"
edition = "2021"
publish = false
[features]
tests-integration/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "tests-integration"
version = "0.1.0"
authors = ["Tokio Contributors <[email protected]>"]
edition = "2018"
edition = "2021"
publish = false
[[bin]]
tests-integration/tests/process_stdio.rs
-1
View File
@@ -7,7 +7,6 @@ use tokio::process::{Child, Command};
use tokio_test::assert_ok;
use futures::future::{self, FutureExt};
use std::convert::TryInto;
use std::env;
use std::io;
use std::process::{ExitStatus, Stdio};
tokio-macros/Cargo.toml
+1 -1
View File
@@ -5,7 +5,7 @@ name = "tokio-macros"
# - Update CHANGELOG.md.
# - Create "tokio-macros-1.x.y" git tag.
version = "2.1.0"
edition = "2018"
edition = "2021"
rust-version = "1.56"
authors = ["Tokio Contributors <[email protected]>"]
license = "MIT"
tokio-stream/Cargo.toml
+1 -1
View File
@@ -37,7 +37,7 @@ signal = ["tokio/signal"]
[dependencies]
futures-core = { version = "0.3.0" }
pin-project-lite = "0.2.0"
pin-project-lite = "0.2.7"
tokio = { version = "1.15.0", path = "../tokio", features = ["sync"] }
tokio-util = { version = "0.7.0", path = "../tokio-util", optional = true }
tokio-stream/fuzz/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "tokio-stream-fuzz"
version = "0.0.0"
publish = false
edition = "2018"
edition = "2021"
[package.metadata]
cargo-fuzz = true
tokio-util/Cargo.toml
+1 -1
View File
@@ -40,7 +40,7 @@ futures-core = "0.3.0"
futures-sink = "0.3.0"
futures-io = { version = "0.3.0", optional = true }
futures-util = { version = "0.3.0", optional = true }
pin-project-lite = "0.2.0"
pin-project-lite = "0.2.7"
slab = { version = "0.4.4", optional = true } # Backs `DelayQueue`
tracing = { version = "0.1.25", default-features = false, features = ["std"], optional = true }
tokio/Cargo.toml
+4 -4
View File
@@ -99,7 +99,7 @@ autocfg = "1.1"
[dependencies]
tokio-macros = { version = "~2.1.0", path = "../tokio-macros", optional = true }
pin-project-lite = "0.2.0"
pin-project-lite = "0.2.7"
# Everything else is optional...
bytes = { version = "1.0.0", optional = true }
@@ -121,18 +121,18 @@ tracing = { version = "0.1.25", default-features = false, features = ["std"], op
backtrace = { version = "0.3.58" }
[target.'cfg(unix)'.dependencies]
libc = { version = "0.2.42", optional = true }
libc = { version = "0.2.145", optional = true }
signal-hook-registry = { version = "1.1.1", optional = true }
[target.'cfg(unix)'.dev-dependencies]
libc = { version = "0.2.42" }
libc = { version = "0.2.145" }
nix = { version = "0.26", default-features = false, features = ["fs", "socket"] }
[target.'cfg(windows)'.dependencies.windows-sys]
version = "0.48"
optional = true
[target.'cfg(docsrs)'.dependencies.windows-sys]
[target.'cfg(windows)'.dev-dependencies.windows-sys]
version = "0.48"
features = [
"Win32_Foundation",
tokio/fuzz/Cargo.toml
+1 -1
View File
@@ -2,7 +2,7 @@
name = "tokio-fuzz"
version = "0.0.0"
publish = false
edition = "2018"
edition = "2021"
[package.metadata]
cargo-fuzz = true
tokio/src/net/unix/ucred.rs
+13 -3
View File
@@ -31,7 +31,12 @@ impl UCred {
}
}
#[cfg(any(target_os = "linux", target_os = "android", target_os = "openbsd"))]
#[cfg(any(
target_os = "linux",
target_os = "redox",
target_os = "android",
target_os = "openbsd"
))]
pub(crate) use self::impl_linux::get_peer_cred;
#[cfg(any(target_os = "netbsd"))]
@@ -49,7 +54,12 @@ pub(crate) use self::impl_solaris::get_peer_cred;
#[cfg(target_os = "aix")]
pub(crate) use self::impl_aix::get_peer_cred;
#[cfg(any(target_os = "linux", target_os = "android", target_os = "openbsd"))]
#[cfg(any(
target_os = "linux",
target_os = "redox",
target_os = "android",
target_os = "openbsd"
))]
pub(crate) mod impl_linux {
use crate::net::unix::{self, UnixStream};
@@ -58,7 +68,7 @@ pub(crate) mod impl_linux {
#[cfg(target_os = "openbsd")]
use libc::sockpeercred as ucred;
#[cfg(any(target_os = "linux", target_os = "android"))]
#[cfg(any(target_os = "linux", target_os = "redox", target_os = "android"))]
use libc::ucred;
pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
tokio/src/runtime/dump.rs
+10
View File
@@ -1,26 +1,36 @@
//! Snapshots of runtime state.
//!
//! See [Handle::dump][crate::runtime::Handle::dump].
use std::fmt;
/// A snapshot of a runtime's state.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Dump {
tasks: Tasks,
}
/// Snapshots of tasks.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Tasks {
tasks: Vec<Task>,
}
/// A snapshot of a task.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Task {
trace: Trace,
}
/// An execution trace of a task's last poll.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Trace {
inner: super::task::trace::Trace,
tokio/src/runtime/handle.rs
+117 -1
View File
@@ -373,7 +373,123 @@ cfg_metrics! {
cfg_taskdump! {
impl Handle {
/// Capture a snapshot of this runtime's state.
/// Captures a snapshot of the runtime's state.
///
/// This functionality is experimental, and comes with a number of
/// requirements and limitations.
///
/// # Examples
///
/// This can be used to get call traces of each task in the runtime.
/// Calls to `Handle::dump` should usually be enclosed in a
/// [timeout][crate::time::timeout], so that dumping does not escalate a
/// single blocked runtime thread into an entirely blocked runtime.
///
/// ```
/// # use tokio::runtime::Runtime;
/// # fn dox() {
/// # let rt = Runtime::new().unwrap();
/// # rt.spawn(async {
/// use tokio::runtime::Handle;
/// use tokio::time::{timeout, Duration};
///
/// // Inside an async block or function.
/// let handle = Handle::current();
/// if let Ok(dump) = timeout(Duration::from_secs(2), handle.dump()).await {
/// for (i, task) in dump.tasks().iter().enumerate() {
/// let trace = task.trace();
/// println!("TASK {i}:");
/// println!("{trace}\n");
/// }
/// }
/// # });
/// # }
/// ```
///
/// This produces highly detailed traces of tasks; e.g.:
///
/// ```plain
/// TASK 0:
/// ╼ dump::main::{{closure}}::a::{{closure}} at /tokio/examples/dump.rs:18:20
/// └╼ dump::main::{{closure}}::b::{{closure}} at /tokio/examples/dump.rs:23:20
/// └╼ dump::main::{{closure}}::c::{{closure}} at /tokio/examples/dump.rs:28:24
/// └╼ tokio::sync::barrier::Barrier::wait::{{closure}} at /tokio/tokio/src/sync/barrier.rs:129:10
/// └╼ <tokio::util::trace::InstrumentedAsyncOp<F> as core::future::future::Future>::poll at /tokio/tokio/src/util/trace.rs:77:46
/// └╼ tokio::sync::barrier::Barrier::wait_internal::{{closure}} at /tokio/tokio/src/sync/barrier.rs:183:36
/// └╼ tokio::sync::watch::Receiver<T>::changed::{{closure}} at /tokio/tokio/src/sync/watch.rs:604:55
/// └╼ tokio::sync::watch::changed_impl::{{closure}} at /tokio/tokio/src/sync/watch.rs:755:18
/// └╼ <tokio::sync::notify::Notified as core::future::future::Future>::poll at /tokio/tokio/src/sync/notify.rs:1103:9
/// └╼ tokio::sync::notify::Notified::poll_notified at /tokio/tokio/src/sync/notify.rs:996:32
/// ```
///
/// # Requirements
///
/// ## Debug Info Must Be Available
///
/// To produce task traces, the application must **not** be compiled
/// with split debuginfo. On Linux, including debuginfo within the
/// application binary is the (correct) default. You can further ensure
/// this behavior with the following directive in your `Cargo.toml`:
///
/// ```toml
/// [profile.*]
/// split-debuginfo = "off"
/// ```
///
/// ## Unstable Features
///
/// This functionality is **unstable**, and requires both the
/// `tokio_unstable` and `tokio_taskdump` cfg flags to be set.
///
/// You can do this by setting the `RUSTFLAGS` environment variable
/// before invoking `cargo`; e.g.:
/// ```bash
/// RUSTFLAGS="--cfg tokio_unstable --cfg tokio_taskdump" cargo run --example dump
/// ```
///
/// Or by [configuring][cargo-config] `rustflags` in
/// `.cargo/config.toml`:
/// ```text
/// [build]
/// rustflags = ["--cfg tokio_unstable", "--cfg tokio_taskdump"]
/// ```
///
/// [cargo-config]:
/// https://doc.rust-lang.org/cargo/reference/config.html
///
/// ## Platform Requirements
///
/// Task dumps are supported on Linux atop aarch64, x86 and x86_64.
///
/// ## Current Thread Runtime Requirements
///
/// On the `current_thread` runtime, task dumps may only be requested
/// from *within* the context of the runtime being dumped. Do not, for
/// example, await `Handle::dump()` on a different runtime.
///
/// # Limitations
///
/// ## Performance
///
/// Although enabling the `tokio_taskdump` feature imposes virtually no
/// additional runtime overhead, actually calling `Handle::dump` is
/// expensive. The runtime must synchronize and pause its workers, then
/// re-poll every task in a special tracing mode. Avoid requesting dumps
/// often.
///
/// ## Local Executors
///
/// Tasks managed by local executors (e.g., `FuturesUnordered` and
/// [`LocalSet`][crate::task::LocalSet]) may not appear in task dumps.
///
/// ## Non-Termination When Workers Are Blocked
///
/// The future produced by `Handle::dump` may never produce `Ready` if
/// another runtime worker is blocked for more than 250ms. This may
/// occur if a dump is requested during shutdown, or if another runtime
/// worker is infinite looping or synchronously deadlocked. For these
/// reasons, task dumping should usually be paired with an explicit
/// [timeout][crate::time::timeout].
pub async fn dump(&self) -> crate::runtime::Dump {
match &self.inner {
scheduler::Handle::CurrentThread(handle) => handle.dump(),
tokio/src/runtime/scheduler/multi_thread/queue.rs
+3 -3
View File
@@ -105,9 +105,9 @@ pub(crate) fn local<T: 'static>() -> (Steal<T>, Local<T>) {
}
impl<T> Local<T> {
/// Returns true if the queue has entries that can be stolen.
pub(crate) fn is_stealable(&self) -> bool {
!self.inner.is_empty()
/// Returns the number of entries in the queue
pub(crate) fn len(&self) -> usize {
self.inner.len() as usize
}
/// How many tasks can be pushed into the queue
tokio/src/runtime/scheduler/multi_thread/worker.rs
+5 -5
View File
@@ -1248,11 +1248,11 @@ impl Worker {
}
fn can_transition_to_parked(&self, cx: &Context, core: &mut Core) -> bool {
core.lifo_slot.is_none()
// cx.shared().remotes[core.index].lifo_slot.is_none()
&& core.run_queue.is_empty()
&& !self.is_shutdown
&& !self.is_traced
!self.has_tasks(core) && !self.is_shutdown && !self.is_traced
}
fn has_tasks(&self, core: &Core) -> bool {
core.lifo_slot.is_some() || !core.run_queue.is_empty()
}
/// Signals all tasks to shut down, and waits for them to complete. Must run
tokio/src/runtime/time/entry.rs
+5 -1
View File
@@ -72,6 +72,10 @@ type TimerResult = Result<(), crate::time::error::Error>;
const STATE_DEREGISTERED: u64 = u64::MAX;
const STATE_PENDING_FIRE: u64 = STATE_DEREGISTERED - 1;
const STATE_MIN_VALUE: u64 = STATE_PENDING_FIRE;
/// The largest safe integer to use for ticks.
///
/// This value should be updated if any other signal values are added above.
pub(super) const MAX_SAFE_MILLIS_DURATION: u64 = u64::MAX - 2;
/// This structure holds the current shared state of the timer - its scheduled
/// time (if registered), or otherwise the result of the timer completing, as
@@ -126,7 +130,7 @@ impl StateCell {
fn when(&self) -> Option<u64> {
let cur_state = self.state.load(Ordering::Relaxed);
if cur_state == u64::MAX {
if cur_state == STATE_DEREGISTERED {
None
} else {
Some(cur_state)
tokio/src/runtime/time/mod.rs
+1 -1
View File
@@ -8,7 +8,7 @@
mod entry;
pub(crate) use entry::TimerEntry;
use entry::{EntryList, TimerHandle, TimerShared};
use entry::{EntryList, TimerHandle, TimerShared, MAX_SAFE_MILLIS_DURATION};
mod handle;
pub(crate) use self::handle::Handle;
tokio/src/runtime/time/source.rs
+2 -1
View File
@@ -1,3 +1,4 @@
use super::MAX_SAFE_MILLIS_DURATION;
use crate::time::{Clock, Duration, Instant};
/// A structure which handles conversion from Instants to u64 timestamps.
@@ -25,7 +26,7 @@ impl TimeSource {
.unwrap_or_else(|| Duration::from_secs(0));
let ms = dur.as_millis();
ms.try_into().unwrap_or(u64::MAX)
ms.try_into().unwrap_or(MAX_SAFE_MILLIS_DURATION)
}
pub(crate) fn tick_to_duration(&self, t: u64) -> Duration {
tokio/src/sync/batch_semaphore.rs
+1 -1
View File
@@ -264,7 +264,7 @@ impl Semaphore {
match self.permits.compare_exchange(curr, next, AcqRel, Acquire) {
Ok(_) => {
// TODO: Instrument once issue has been solved}
// TODO: Instrument once issue has been solved
return Ok(());
}
Err(actual) => curr = actual,
tokio/src/task/join_set.rs
+3 -2
View File
@@ -362,7 +362,7 @@ impl<T: 'static> JoinSet<T> {
/// This can happen if the [coop budget] is reached.
///
/// [coop budget]: crate::task#cooperative-scheduling
fn poll_join_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<T, JoinError>>> {
pub fn poll_join_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<T, JoinError>>> {
// The call to `pop_notified` moves the entry to the `idle` list. It is moved back to
// the `notified` list if the waker is notified in the `poll` call below.
let mut entry = match self.inner.pop_notified(cx.waker()) {
@@ -419,7 +419,8 @@ impl<T: 'static> JoinSet<T> {
/// [coop budget]: crate::task#cooperative-scheduling
/// [task ID]: crate::task::Id
#[cfg(tokio_unstable)]
fn poll_join_next_with_id(
#[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
pub fn poll_join_next_with_id(
&mut self,
cx: &mut Context<'_>,
) -> Poll<Option<Result<(Id, T), JoinError>>> {
tokio/src/task/task_local.rs
+61 -58
View File
@@ -1,3 +1,4 @@
use pin_project_lite::pin_project;
use std::cell::RefCell;
use std::error::Error;
use std::future::Future;
@@ -299,36 +300,53 @@ impl<T: 'static> fmt::Debug for LocalKey<T> {
}
}
/// A future that sets a value `T` of a task local for the future `F` during
/// its execution.
///
/// The value of the task-local must be `'static` and will be dropped on the
/// completion of the future.
///
/// Created by the function [`LocalKey::scope`](self::LocalKey::scope).
///
/// ### Examples
///
/// ```
/// # async fn dox() {
/// tokio::task_local! {
/// static NUMBER: u32;
/// }
///
/// NUMBER.scope(1, async move {
/// println!("task local value: {}", NUMBER.get());
/// }).await;
/// # }
/// ```
// Doesn't use pin_project due to custom Drop.
pub struct TaskLocalFuture<T, F>
where
T: 'static,
{
local: &'static LocalKey<T>,
slot: Option<T>,
future: Option<F>,
_pinned: PhantomPinned,
pin_project! {
/// A future that sets a value `T` of a task local for the future `F` during
/// its execution.
///
/// The value of the task-local must be `'static` and will be dropped on the
/// completion of the future.
///
/// Created by the function [`LocalKey::scope`](self::LocalKey::scope).
///
/// ### Examples
///
/// ```
/// # async fn dox() {
/// tokio::task_local! {
/// static NUMBER: u32;
/// }
///
/// NUMBER.scope(1, async move {
/// println!("task local value: {}", NUMBER.get());
/// }).await;
/// # }
/// ```
pub struct TaskLocalFuture<T, F>
where
T: 'static,
{
local: &'static LocalKey<T>,
slot: Option<T>,
#[pin]
future: Option<F>,
#[pin]
_pinned: PhantomPinned,
}
impl<T: 'static, F> PinnedDrop for TaskLocalFuture<T, F> {
fn drop(this: Pin<&mut Self>) {
let this = this.project();
if mem::needs_drop::<F>() && this.future.is_some() {
// Drop the future while the task-local is set, if possible. Otherwise
// the future is dropped normally when the `Option<F>` field drops.
let mut future = this.future;
let _ = this.local.scope_inner(this.slot, || {
future.set(None);
});
}
}
}
}
impl<T: 'static, F: Future> Future for TaskLocalFuture<T, F> {
@@ -336,23 +354,21 @@ impl<T: 'static, F: Future> Future for TaskLocalFuture<T, F> {
#[track_caller]
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// safety: The TaskLocalFuture struct is `!Unpin` so there is no way to
// move `self.future` from now on.
let this = unsafe { Pin::into_inner_unchecked(self) };
let mut future_opt = unsafe { Pin::new_unchecked(&mut this.future) };
let this = self.project();
let mut future_opt = this.future;
let res =
this.local
.scope_inner(&mut this.slot, || match future_opt.as_mut().as_pin_mut() {
Some(fut) => {
let res = fut.poll(cx);
if res.is_ready() {
future_opt.set(None);
}
Some(res)
let res = this
.local
.scope_inner(this.slot, || match future_opt.as_mut().as_pin_mut() {
Some(fut) => {
let res = fut.poll(cx);
if res.is_ready() {
future_opt.set(None);
}
None => None,
});
Some(res)
}
None => None,
});
match res {
Ok(Some(res)) => res,
@@ -362,19 +378,6 @@ impl<T: 'static, F: Future> Future for TaskLocalFuture<T, F> {
}
}
impl<T: 'static, F> Drop for TaskLocalFuture<T, F> {
fn drop(&mut self) {
if mem::needs_drop::<F>() && self.future.is_some() {
// Drop the future while the task-local is set, if possible. Otherwise
// the future is dropped normally when the `Option<F>` field drops.
let future = &mut self.future;
let _ = self.local.scope_inner(&mut self.slot, || {
*future = None;
});
}
}
}
impl<T: 'static, F> fmt::Debug for TaskLocalFuture<T, F>
where
T: fmt::Debug,
tokio/tests/time_sleep.rs
+14
View File
@@ -267,6 +267,20 @@ async fn exactly_max() {
time::sleep(ms(MAX_DURATION)).await;
}
#[tokio::test]
async fn issue_5183() {
time::pause();
let big = std::time::Duration::from_secs(u64::MAX / 10);
// This is a workaround since awaiting sleep(big) will never finish.
#[rustfmt::skip]
tokio::select! {
biased;
_ = tokio::time::sleep(big) => {}
_ = tokio::time::sleep(std::time::Duration::from_nanos(1)) => {}
}
}
#[tokio::test]
async fn no_out_of_bounds_close_to_max() {
time::pause();