Compare commits

...
Author SHA1 Message Date
Carl Lerche ec4b1d7215 chore: forward port 1.43.x 2025-04-04 16:13:58 -07:00
Carl Lerche e3c3a56718 Merge branch 'tokio-1.43.x' into forward-port-1.43.x 2025-04-04 16:11:53 -07:00
Alice Ryhl d413c9c02a chore: prepare Tokio v1.44.1 (#7217) 2025-03-13 09:13:25 +01:00
Carl Lerche addbfb9204 rt: skip defer queue in block_in_place context (#7216) 2025-03-13 08:18:13 +01:00
Alice Ryhl 8182ecf262 chore: prepare Tokio v1.44.0 (#7202) 2025-03-07 21:11:03 +01:00
Motoyuki Kimura a258bff701 ci: enable printing in multi thread loom tests (#7200) 2025-03-07 13:33:16 +01:00
Stepan Koltsov e076d21f67 process: clarify Child::kill behavior (#7162) 2025-03-06 14:59:51 +03:30
Noah Kennedy 042433cdcc net: debug_assert on creating a tokio socket from a blocking one (#7166)
See #5595 and #7172.

This adds a debug assertion that checks that a supplied underlying std socket is set to nonblocking mode when constructing a tokio socket object from such an object.

This only works on unix.
2025-03-05 18:10:30 +00:00
M.Amin Rayej 0284d1b5c8 macros: make select! budget-aware (#7164) 2025-03-05 01:07:18 +03:30
Carl Lerche 710bc8071e rt: coop should yield using waker defer strategy (#7185) 2025-03-04 15:02:43 +01:00
Alice Ryhl a2b12bd579 readme: adjust release schedule to once per month (#7191) 2025-03-03 14:04:42 +03:30
Jonathan Hiles e7b593cbee process: fix grammar of the ChildStdin struct doc comment (#7192) 2025-03-03 13:01:35 +03:30
kilavvy 3aaf4a5377 coop: adjust grammar in tests/coop_budget.rs (#7173) 2025-03-03 09:59:03 +01:00
Alice Ryhl 8e741c1c0e tokio: mark 1.43 as LTS (#7189) 2025-03-03 09:54:26 +01:00
dlzht 47d46455bd util: optimize buffer reserve for AnyDelimiterCodec::encode (#7188) 2025-03-02 19:53:19 +03:30
Adriano Mourão 20c1fdc678 runtime: consistently use worker_threads instead of core_threads (#7186) 2025-02-28 10:19:24 +01:00
Josh Triplett 638ce93591 io: add read_exact_arc to safely read a new uninitialized Arc (#7165) 2025-02-27 14:07:49 +01:00
Kyle Cotton c853991b1e io: swap reader/writer in simplex doc test (#7176) 2025-02-25 14:49:19 +00:00
Josh Triplett 6d410f6c90 util: fix example of Buf implementor in StreamReader docs (#7167) 2025-02-21 09:22:16 +03:30
Finomnis a27575f284 signal: fix CTRL_CLOSE, CTRL_LOGOFF, CTRL_SHUTDOWN on windows (#7122) 2025-02-19 12:27:13 +01:00
Stepan Koltsov 13fbdace66 process: add test for Child::kill after Child::wait (#7163) 2025-02-19 00:52:50 +03:30
Timo 4380c3d821 sync: Added WeakSender to sync::broadcast::channel (#7100) 2025-02-17 21:24:29 +01:00
katelyn martin 383da87313 sync: implement oneshot::Receiver::is_empty() (#7153) 2025-02-17 02:01:11 +03:30
katelyn martin 17117b591e sync: implement oneshot::Receiver::is_terminated() (#7152) 2025-02-16 22:55:50 +03:30
Nathaniel Bajo aa70f6c5f0 io: add documentation for SyncIoBridge with examples and alternatives (#6815) 2025-02-16 19:01:52 +01:00
Dylan Laufenberg 67c343d9e9 docs: fix nesting of next sections under examples (#7159)
Promote the feature flags and supported platforms sections out from
under examples, as they are not examples. Adjust their subsections
accordingly. Expose these subsections via navigation sidebar.
2025-02-15 23:08:13 +01:00
Oleksandr Babak 34cdcc7d87 macros: docs about select! alternatives (#7110) 2025-02-15 22:49:23 +01:00
Stepan Koltsov 8e134172dd process: calling start_kill on exited child should not fail (#7160) 2025-02-15 13:48:22 +03:30
M.Amin Rayej 605ef578df coop: expose coop as a public module (#7116) 2025-02-14 18:56:12 +03:30
M.Amin Rayej 9b578f0c9d ci: bump freeBSD image version (#7158) 2025-02-14 17:37:16 +03:30
M.Amin Rayej 0a15768380 io: clean up buffer casts (#7142) 2025-02-10 19:57:25 +03:30
Motoyuki Kimura eb1a2ee990 net: rename the argument for send_to (#7146) 2025-02-08 15:05:05 +03:30
M.Amin Rayej 8713d39228 process: add example for reading Child stdout (#7141) 2025-02-08 13:31:02 +03:30
Florian Gäbler 7e27911911 fs: align symlink and hardlink parameter names with std (#7143) 2025-02-07 23:40:15 +03:30
Alice Ryhl 4b3da20c98 fs: empty reads on File should not start a background read (#7139) 2025-02-06 01:37:29 +03:30
Jason Gin b8ac94ed70 rt: add before and after task poll callbacks (#7120)
Add callbacks for poll start and stop, enabling users to instrument these points
in the runtime's life cycle.
2025-01-30 21:14:00 +00:00
Oliver Wanglerandow 5086e56dcb io: implemented get_ref and get_mut for SyncIoBridge (#7128)
Co-authored-by: ow <[email protected]>
2025-01-28 15:28:07 +01:00
Ariel Ben-YehudaandAriel Ben-Yehuda 2671ffb55b tracing: make the task tracing API unstable piblkc (#6972)
* make self-tracing public

* address review comments

* try to fix doctest

* adjust imports to fit standard

* more documentation

---------

Co-authored-by: Ariel Ben-Yehuda <[email protected]>
2025-01-27 13:09:23 -08:00
Taiki Endo 7f09959b0a chore: use [lints] to address unexpected_cfgs lint (#7124) 2025-01-25 17:46:21 +01:00
Taiki Endo fb7dec0e95 ci: test AArch64/Armv7hf Linux on ubuntu-22.04-arm runner (#7123) 2025-01-25 11:17:37 +01:00
M.Amin Rayej ee19b0ed73 net: fix warnings when building the docs (#7113) 2025-01-22 11:48:43 +01:00
Josh McKinney c081dfe3ce macros: characterization tests for ? operator fail (#7069)
When a `?` operator is used in a tokio entry point function (wrapped in
`#[tokio::main]`), which has a Option or Result return type, but where
the function does not actually return that type correctly, currently the
compiler returns two errors instead of just one. The first of which is
incorrect and only exists due to the macro expanding to an async block.

```
cannot use the `?` operator in an async block that returns `()`
```

This commit is a characterization test for this behavior to help show
when it's fixed (or even changed for better / worse)
2025-01-22 10:55:00 +01:00
M.Amin Rayej 21a13f9eea runtime: clean up magic number in registration set (#7112) 2025-01-21 14:40:32 +01:00
Motoyuki Kimura a82bdeebe9 sync: handle panic during mpsc drop (#7094) 2025-01-13 18:36:51 +01:00
Evan Rittenhouse 435e39001b sync: fix sync::broadcast::Sender<T>::closed() doctest (#7090)
The test's previous iteration could sometimes flake since we didn't
await the completion of the first task. Since the tasks only existed to
`move` the relevant `rx`'s in, to force a drop, we can omit them
entirely and drop the `rx`s via `drop()`. This prevents any
scheduling-related flakes.
2025-01-12 12:33:07 +01:00
Marshall Lee dabae570b1 ci: add spellcheck.dic validation (#7062) 2025-01-10 13:39:23 +01:00
29 6bd3be2e45 process: add Command::get_kill_on_drop() (#7086) 2025-01-10 12:35:13 +01:00
Alice Ryhl 6fc1a8c8da ci: fix ci error about wasm32-wasip1 (#7085) 2025-01-10 18:59:35 +09:00
Evan Rittenhouse 5c8cd33820 sync: add broadcast::Sender::closed (#6685) 2025-01-09 16:37:49 +01:00
126 changed files with 2763 additions and 329 deletions
-25
View File
@@ -1,25 +0,0 @@
version: 2.1
jobs:
test-arm:
machine:
image: default
resource_class: arm.medium
environment:
# Change to pin rust version
RUST_STABLE: stable
steps:
- checkout
- run:
name: Install Rust
command: |
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs -o rustup.sh
chmod +x rustup.sh
./rustup.sh -y --default-toolchain $RUST_STABLE
source "$HOME"/.cargo/env
# Only run Tokio tests
- run: cargo test --all-features -p tokio
workflows:
ci:
jobs:
- test-arm
+1 -1
View File
@@ -4,7 +4,7 @@ freebsd_instance:
image_family: freebsd-14-2
env:
RUST_STABLE: stable
RUST_NIGHTLY: nightly-2024-05-05
RUST_NIGHTLY: nightly-2025-01-25
RUSTFLAGS: -D warnings
# Test FreeBSD in a full VM on cirrus-ci.com. Test the i686 target too, in the
+33 -25
View File
@@ -16,9 +16,9 @@ env:
RUSTUP_WINDOWS_PATH_ADD_BIN: 1
# Change to specific Rust release to pin
rust_stable: stable
rust_nightly: nightly-2024-05-05
rust_nightly: nightly-2025-01-25
# Pin a specific miri version
rust_miri_nightly: nightly-2024-10-21
rust_miri_nightly: nightly-2025-01-25
rust_clippy: '1.77'
# When updating this, also update:
# - README.md
@@ -540,15 +540,19 @@ jobs:
cross-test-with-parking_lot:
needs: basics
runs-on: ubuntu-latest
runs-on: ${{ matrix.os }}
strategy:
matrix:
include:
- target: i686-unknown-linux-gnu
os: ubuntu-latest
rustflags: --cfg tokio_taskdump
- target: armv5te-unknown-linux-gnueabi
os: ubuntu-latest
- target: armv7-unknown-linux-gnueabihf
os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
- target: aarch64-unknown-linux-gnu
os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
rustflags: --cfg tokio_taskdump
steps:
- uses: actions/checkout@v4
@@ -580,15 +584,19 @@ jobs:
cross-test-without-parking_lot:
needs: basics
runs-on: ubuntu-latest
runs-on: ${{ matrix.os }}
strategy:
matrix:
include:
- target: i686-unknown-linux-gnu
os: ubuntu-latest
rustflags: --cfg tokio_taskdump
- target: armv5te-unknown-linux-gnueabi
os: ubuntu-latest
- target: armv7-unknown-linux-gnueabihf
os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
- target: aarch64-unknown-linux-gnu
os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
rustflags: --cfg tokio_taskdump
steps:
- uses: actions/checkout@v4
@@ -1093,23 +1101,6 @@ jobs:
run: cargo check-external-types --all-features
working-directory: tokio
check-unexpected-lints-cfgs:
name: check unexpected lints and cfgs
needs: basics
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Install Rust ${{ env.rust_nightly }}
uses: dtolnay/rust-toolchain@master
with:
toolchain: ${{ env.rust_nightly }}
- name: don't allow warnings
run: sed -i '/#!\[allow(unknown_lints, unexpected_cfgs)\]/d' */src/lib.rs */tests/*.rs
- name: check for unknown lints and cfgs
run: cargo check --all-features --tests
env:
RUSTFLAGS: -Dwarnings --check-cfg=cfg(loom,tokio_unstable,tokio_taskdump,fuzzing,mio_unsupported_force_poll_poll,tokio_internal_mt_counters,fs,tokio_no_parking_lot,tokio_no_tuning_tests) -Funexpected_cfgs -Funknown_lints
check-fuzzing:
name: check-fuzzing
needs: basics
@@ -1147,8 +1138,25 @@ jobs:
- uses: actions/checkout@v4
- name: Make sure dictionary words are sorted and unique
run: |
# `sed` removes the first line (number of words) and
# the last line (new line).
FILE="spellcheck.dic"
# Verify the first line is an integer.
first_line=$(head -n 1 "$FILE")
if ! [[ "$first_line" =~ ^[0-9]+$ ]]; then
echo "Error: The first line of $FILE must be an integer, but got: '$first_line'"
exit 1
fi
expected_count="$first_line"
# Check that the number of lines matches the integer.
# xargs (with no arguments) will strip leading/trailing whitespacefrom wc's output.
actual_count=$(sed '1d' "$FILE" | wc -l | xargs)
if [ "$expected_count" -ne "$actual_count" ]; then
echo "Error: The number of lines ($actual_count) does not match $expected_count."
exit 1
fi
# `sed` removes the first line (number of words).
#
# `sort` makes sure everything in between is sorted
# and contains no duplicates.
@@ -1158,10 +1166,10 @@ jobs:
# environments.
(
sed '1d; $d' spellcheck.dic | LC_ALL=en_US.UTF8 sort -uc
sed '1d' $FILE | LC_ALL=en_US.UTF8 sort -uc
) || {
echo "Dictionary is not in sorted order. Correct order is:"
LC_ALL=en_US.UTF8 sort -u <(sed '1d; $d' spellcheck.dic)
LC_ALL=en_US.UTF8 sort -u <(sed '1d' $FILE)
false
}
- name: Run cargo-spellcheck
+2 -2
View File
@@ -91,7 +91,7 @@ jobs:
toolchain: ${{ env.rust_stable }}
- uses: Swatinem/rust-cache@v2
- name: loom ${{ matrix.scope }}
run: cargo test --lib --release --features full -- $SCOPE
run: cargo test --lib --release --features full -- --nocapture $SCOPE
working-directory: tokio
env:
SCOPE: ${{ matrix.scope }}
@@ -116,7 +116,7 @@ jobs:
toolchain: ${{ env.rust_stable }}
- uses: Swatinem/rust-cache@v2
- name: loom ${{ matrix.scope }}
run: cargo test --lib --release --features full -- $SCOPE
run: cargo test --lib --release --features full -- --nocapture $SCOPE
working-directory: tokio
env:
SCOPE: ${{ matrix.scope }}
+6 -5
View File
@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
```toml
[dependencies]
tokio = { version = "1.43.1", features = ["full"] }
tokio = { version = "1.44.2", features = ["full"] }
```
Then, on your main.rs:
@@ -205,8 +205,8 @@ works with the MSRV of that minor release.
## Release schedule
Tokio doesn't follow a fixed release schedule, but we typically make one to two
new minor releases each month. We make patch releases for bugfixes as necessary.
Tokio doesn't follow a fixed release schedule, but we typically make one minor
release each month. We make patch releases for bugfixes as necessary.
## Bug patching policy
@@ -216,8 +216,8 @@ warrants a patch release with a fix for the bug, it will be backported and
released as a new patch release for each LTS minor version. Our current LTS
releases are:
* `1.36.x` - LTS release until March 2025. (MSRV 1.63)
* `1.38.x` - LTS release until July 2025. (MSRV 1.63)
* `1.43.x` - LTS release until March 2026. (MSRV 1.70)
Each LTS release will continue to receive backported fixes for at least a year.
If you wish to use a fixed minor release in your project, we recommend that you
@@ -227,7 +227,7 @@ To use a fixed minor version, you can specify the version with a tilde. For
example, to specify that you wish to use the newest `1.32.x` patch release, you
can use the following dependency specification:
```text
tokio = { version = "~1.32", features = [...] }
tokio = { version = "~1.38", features = [...] }
```
### Previous LTS releases
@@ -238,6 +238,7 @@ tokio = { version = "~1.32", features = [...] }
* `1.20.x` - LTS release until September 2023.
* `1.25.x` - LTS release until March 2024.
* `1.32.x` - LTS release until September 2024.
* `1.36.x` - LTS release until March 2025.
## License
+3
View File
@@ -95,3 +95,6 @@ harness = false
name = "time_timeout"
path = "time_timeout.rs"
harness = false
[lints]
workspace = true
-2
View File
@@ -1,5 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
//! 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.
+2 -1
View File
@@ -1,4 +1,4 @@
298
300
&
+
<
@@ -78,6 +78,7 @@ deallocate
deallocated
Deallocates
debuginfo
decrement
decrementing
demangled
dequeued
+3
View File
@@ -13,3 +13,6 @@ tokio = { version = "1.0.0", path = "../tokio/", features = ["full"] }
[dev-dependencies]
rand = "0.8"
[lints]
workspace = true
+3
View File
@@ -15,3 +15,6 @@ tokio = { version = "1.0.0", path = "../tokio", optional = true }
[dev-dependencies]
trybuild = "1.0"
[lints]
workspace = true
@@ -23,6 +23,40 @@ async fn extra_semicolon() -> Result<(), ()> {
Ok(());
}
/// This test is a characterization test for the `?` operator.
///
/// See <https://github.com/tokio-rs/tokio/issues/6930#issuecomment-2572502517> for more details.
///
/// It should fail with a single error message about the return type of the function, but instead
/// if fails with an extra error message due to the `?` operator being used within the async block
/// rather than the original function.
///
/// ```text
/// 28 | None?;
/// | ^ cannot use the `?` operator in an async block that returns `()`
/// ```
#[tokio::main]
async fn question_mark_operator_with_invalid_option() -> Option<()> {
None?;
}
/// This test is a characterization test for the `?` operator.
///
/// See <https://github.com/tokio-rs/tokio/issues/6930#issuecomment-2572502517> for more details.
///
/// It should fail with a single error message about the return type of the function, but instead
/// if fails with an extra error message due to the `?` operator being used within the async block
/// rather than the original function.
///
/// ```text
/// 33 | Ok(())?;
/// | ^ cannot use the `?` operator in an async block that returns `()`
/// ```
#[tokio::main]
async fn question_mark_operator_with_invalid_result() -> Result<(), ()> {
Ok(())?;
}
// https://github.com/tokio-rs/tokio/issues/4635
#[allow(redundant_semicolons)]
#[rustfmt::skip]
@@ -49,11 +49,68 @@ help: try adding an expression at the end of the block
24 + Ok(())
|
error[E0308]: mismatched types
--> tests/fail/macros_type_mismatch.rs:32:5
error[E0277]: the `?` operator can only be used in an async block that returns `Result` or `Option` (or another type that implements `FromResidual`)
--> tests/fail/macros_type_mismatch.rs:40:9
|
30 | async fn issue_4635() {
38 | #[tokio::main]
| -------------- this function should return `Result` or `Option` to accept `?`
39 | async fn question_mark_operator_with_invalid_option() -> Option<()> {
40 | None?;
| ^ cannot use the `?` operator in an async block that returns `()`
|
= help: the trait `FromResidual<Option<Infallible>>` is not implemented for `()`
error[E0308]: mismatched types
--> tests/fail/macros_type_mismatch.rs:40:5
|
39 | async fn question_mark_operator_with_invalid_option() -> Option<()> {
| ---------- expected `Option<()>` because of return type
40 | None?;
| ^^^^^^ expected `Option<()>`, found `()`
|
= note: expected enum `Option<()>`
found unit type `()`
help: try adding an expression at the end of the block
|
40 ~ None?;;
41 + None
|
40 ~ None?;;
41 + Some(())
|
error[E0277]: the `?` operator can only be used in an async block that returns `Result` or `Option` (or another type that implements `FromResidual`)
--> tests/fail/macros_type_mismatch.rs:57:11
|
55 | #[tokio::main]
| -------------- this function should return `Result` or `Option` to accept `?`
56 | async fn question_mark_operator_with_invalid_result() -> Result<(), ()> {
57 | Ok(())?;
| ^ cannot use the `?` operator in an async block that returns `()`
|
= help: the trait `FromResidual<Result<Infallible, _>>` is not implemented for `()`
error[E0308]: mismatched types
--> tests/fail/macros_type_mismatch.rs:57:5
|
56 | async fn question_mark_operator_with_invalid_result() -> Result<(), ()> {
| -------------- expected `Result<(), ()>` because of return type
57 | Ok(())?;
| ^^^^^^^^ expected `Result<(), ()>`, found `()`
|
= note: expected enum `Result<(), ()>`
found unit type `()`
help: try adding an expression at the end of the block
|
57 ~ Ok(())?;;
58 + Ok(())
|
error[E0308]: mismatched types
--> tests/fail/macros_type_mismatch.rs:66:5
|
64 | async fn issue_4635() {
| - help: try adding a return type: `-> i32`
31 | return 1;
32 | ;
65 | return 1;
66 | ;
| ^ expected `()`, found integer
+3
View File
@@ -61,3 +61,6 @@ tokio-test = { version = "0.4", path = "../tokio-test", optional = true }
doc-comment = "0.3.1"
futures = { version = "0.3.0", features = ["async-await"] }
bytes = "1.0.0"
[lints]
workspace = true
+3
View File
@@ -31,3 +31,6 @@ tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
[package.metadata.docs.rs]
all-features = true
[lints]
workspace = true
-5
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![allow(clippy::needless_doctest_main)]
#![warn(
missing_debug_implementations,
@@ -211,7 +210,6 @@ use proc_macro::TokenStream;
/// This option is only compatible with the `current_thread` runtime.
///
/// ```no_run
/// # #![allow(unknown_lints, unexpected_cfgs)]
/// #[cfg(tokio_unstable)]
/// #[tokio::main(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
/// async fn main() {
@@ -226,7 +224,6 @@ use proc_macro::TokenStream;
/// Equivalent code not using `#[tokio::main]`
///
/// ```no_run
/// # #![allow(unknown_lints, unexpected_cfgs)]
/// #[cfg(tokio_unstable)]
/// fn main() {
/// tokio::runtime::Builder::new_current_thread()
@@ -480,7 +477,6 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
/// This option is only compatible with the `current_thread` runtime.
///
/// ```no_run
/// # #![allow(unknown_lints, unexpected_cfgs)]
/// #[cfg(tokio_unstable)]
/// #[tokio::test(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
/// async fn my_test() {
@@ -495,7 +491,6 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
/// Equivalent code not using `#[tokio::test]`
///
/// ```no_run
/// # #![allow(unknown_lints, unexpected_cfgs)]
/// #[cfg(tokio_unstable)]
/// #[test]
/// fn my_test() {
+3
View File
@@ -56,3 +56,6 @@ rustdoc-args = ["--cfg", "docsrs"]
# This should allow `docsrs` to be read across projects, so that `tokio-stream`
# can pick up stubbed types exported by `tokio`.
rustc-args = ["--cfg", "docsrs"]
[lints]
workspace = true
-1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![allow(
clippy::cognitive_complexity,
clippy::large_enum_variant,
+3
View File
@@ -30,3 +30,6 @@ futures-util = "0.3.0"
[package.metadata.docs.rs]
all-features = true
[lints]
workspace = true
-1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![warn(
missing_debug_implementations,
missing_docs,
+4 -1
View File
@@ -35,7 +35,7 @@ __docs_rs = ["futures-util"]
[dependencies]
tokio = { version = "1.28.0", path = "../tokio", features = ["sync"] }
bytes = "1.0.0"
bytes = "1.5.0"
futures-core = "0.3.0"
futures-sink = "0.3.0"
futures-io = { version = "0.3.0", optional = true }
@@ -68,3 +68,6 @@ rustc-args = ["--cfg", "docsrs", "--cfg", "tokio_unstable"]
[package.metadata.playground]
features = ["full"]
[lints]
workspace = true
+1 -1
View File
@@ -217,7 +217,7 @@ where
fn encode(&mut self, chunk: T, buf: &mut BytesMut) -> Result<(), AnyDelimiterCodecError> {
let chunk = chunk.as_ref();
buf.reserve(chunk.len() + 1);
buf.reserve(chunk.len() + self.sequence_writer.len());
buf.put(chunk.as_bytes());
buf.put(self.sequence_writer.as_ref());
+3
View File
@@ -18,6 +18,9 @@ mod sink_writer;
mod stream_reader;
cfg_io_util! {
mod read_arc;
pub use self::read_arc::read_exact_arc;
mod sync_bridge;
pub use self::sync_bridge::SyncIoBridge;
}
+44
View File
@@ -0,0 +1,44 @@
use std::io;
use std::mem::MaybeUninit;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};
/// Read data from an `AsyncRead` into an `Arc`.
///
/// This uses `Arc::new_uninit_slice` and reads into the resulting uninitialized `Arc`.
///
/// # Example
///
/// ```
/// # #[tokio::main]
/// # async fn main() -> std::io::Result<()> {
/// use tokio_util::io::read_exact_arc;
///
/// let read = tokio::io::repeat(42);
///
/// let arc = read_exact_arc(read, 4).await?;
///
/// assert_eq!(&arc[..], &[42; 4]);
/// # Ok(())
/// # }
/// ```
pub async fn read_exact_arc<R: AsyncRead>(read: R, len: usize) -> io::Result<Arc<[u8]>> {
tokio::pin!(read);
// TODO(MSRV 1.82): When bumping MSRV, switch to `Arc::new_uninit_slice(len)`. The following is
// equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
let arc: Arc<[MaybeUninit<u8>]> = (0..len).map(|_| MaybeUninit::uninit()).collect();
// TODO(MSRV future): Use `Arc::get_mut_unchecked` once it's stabilized.
// SAFETY: We're the only owner of the `Arc`, and we keep the `Arc` valid throughout this loop
// as we write through this reference.
let mut buf = unsafe { &mut *(Arc::as_ptr(&arc) as *mut [MaybeUninit<u8>]) };
while !buf.is_empty() {
if read.read_buf(&mut buf).await? == 0 {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
}
}
// TODO(MSRV 1.82): When bumping MSRV, switch to `arc.assume_init()`. The following is
// equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
// SAFETY: This changes `[MaybeUninit<u8>]` to `[u8]`, and we've initialized all the bytes in
// the loop above.
Ok(unsafe { Arc::from_raw(Arc::into_raw(arc) as *const [u8]) })
}
+1 -1
View File
@@ -169,7 +169,7 @@ where
/// Convert a stream of byte chunks into an [`AsyncRead`].
///
/// The item should be a [`Result`] with the ok variant being something that
/// implements the [`Buf`] trait (e.g. `Vec<u8>` or `Bytes`). The error
/// implements the [`Buf`] trait (e.g. `Cursor<Vec<u8>>` or `Bytes`). The error
/// should be convertible into an [io error].
///
/// [`Result`]: std::result::Result
+264 -1
View File
@@ -5,7 +5,258 @@ use tokio::io::{
};
/// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
/// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
/// a [`tokio::io::AsyncWrite`] synchronously as a [`std::io::Write`].
///
/// # Alternatives
///
/// In many cases, there are better alternatives to using `SyncIoBridge`, especially
/// if you want to avoid blocking the async runtime. Consider the following scenarios:
///
/// When hashing data, using `SyncIoBridge` can lead to suboptimal performance and
/// might not fully leverage the async capabilities of the system.
///
/// ### Why It Matters:
///
/// `SyncIoBridge` allows you to use asynchronous I/O operations in an synchronous
/// context by blocking the current thread. However, this can be inefficient because:
/// - **Inefficient Resource Usage**: `SyncIoBridge` takes up an entire OS thread,
/// which is inefficient compared to asynchronous code that can multiplex many
/// tasks on a single thread.
/// - **Thread Pool Saturation**: Excessive use of `SyncIoBridge` can exhaust the
/// async runtime's thread pool, reducing the number of threads available for
/// other tasks and impacting overall performance.
/// - **Missed Concurrency Benefits**: By using synchronous operations with
/// `SyncIoBridge`, you lose the ability to interleave tasks efficiently,
/// which is a key advantage of asynchronous programming.
///
/// ## Example 1: Hashing Data
///
/// The use of `SyncIoBridge` is unnecessary when hashing data. Instead, you can
/// process the data asynchronously by reading it into memory, which avoids blocking
/// the async runtime.
///
/// There are two strategies for avoiding `SyncIoBridge` when hashing data. When
/// the data fits into memory, the easiest is to read the data into a `Vec<u8>`
/// and hash it:
///
/// Explanation: This example demonstrates how to asynchronously read data from a
/// reader into memory and hash it using a synchronous hashing function. The
/// `SyncIoBridge` is avoided, ensuring that the async runtime is not blocked.
/// ```rust
/// use tokio::io::AsyncReadExt;
/// use tokio::io::AsyncRead;
/// use std::io::Cursor;
/// # mod blake3 { pub fn hash(_: &[u8]) {} }
///
/// async fn hash_contents(mut reader: impl AsyncRead + Unpin) -> Result<(), std::io::Error> {
/// // Read all data from the reader into a Vec<u8>.
/// let mut data = Vec::new();
/// reader.read_to_end(&mut data).await?;
///
/// // Hash the data using the blake3 hashing function.
/// let hash = blake3::hash(&data);
///
/// Ok(hash)
///}
///
/// #[tokio::main]
/// async fn main() -> Result<(), std::io::Error> {
/// // Example: In-memory data.
/// let data = b"Hello, world!"; // A byte slice.
/// let reader = Cursor::new(data); // Create an in-memory AsyncRead.
/// hash_contents(reader).await
/// }
/// ```
///
/// When the data doesn't fit into memory, the hashing library will usually
/// provide a `hasher` that you can repeatedly call `update` on to hash the data
/// one chunk at the time.
///
/// Explanation: This example demonstrates how to asynchronously stream data in
/// chunks for hashing. Each chunk is read asynchronously, and the hash is updated
/// incrementally. This avoids blocking and improves performance over using
/// `SyncIoBridge`.
///
/// ```rust
/// use tokio::io::AsyncReadExt;
/// use tokio::io::AsyncRead;
/// use std::io::Cursor;
/// # struct Hasher;
/// # impl Hasher { pub fn update(&mut self, _: &[u8]) {} pub fn finalize(&self) {} }
///
/// /// Asynchronously streams data from an async reader, processes it in chunks,
/// /// and hashes the data incrementally.
/// async fn hash_stream(mut reader: impl AsyncRead + Unpin, mut hasher: Hasher) -> Result<(), std::io::Error> {
/// // Create a buffer to read data into, sized for performance.
/// let mut data = vec![0; 64 * 1024];
/// loop {
/// // Read data from the reader into the buffer.
/// let len = reader.read(&mut data).await?;
/// if len == 0 { break; } // Exit loop if no more data.
///
/// // Update the hash with the data read.
/// hasher.update(&data[..len]);
/// }
///
/// // Finalize the hash after all data has been processed.
/// let hash = hasher.finalize();
///
/// Ok(hash)
///}
///
/// #[tokio::main]
/// async fn main() -> Result<(), std::io::Error> {
/// // Example: In-memory data.
/// let data = b"Hello, world!"; // A byte slice.
/// let reader = Cursor::new(data); // Create an in-memory AsyncRead.
/// let hasher = Hasher;
/// hash_stream(reader, hasher).await
/// }
/// ```
///
///
/// ## Example 2: Compressing Data
///
/// When compressing data, the use of `SyncIoBridge` is unnecessary as it introduces
/// blocking and inefficient code. Instead, you can utilize an async compression library
/// such as the [`async-compression`](https://docs.rs/async-compression/latest/async_compression/)
/// crate, which is built to handle asynchronous data streams efficiently.
///
/// Explanation: This example shows how to asynchronously compress data using an
/// async compression library. By reading and writing asynchronously, it avoids
/// blocking and is more efficient than using `SyncIoBridge` with a non-async
/// compression library.
///
/// ```ignore
/// use async_compression::tokio::write::GzipEncoder;
/// use std::io::Cursor;
/// use tokio::io::AsyncRead;
///
/// /// Asynchronously compresses data from an async reader using Gzip and an async encoder.
/// async fn compress_data(mut reader: impl AsyncRead + Unpin) -> Result<(), std::io::Error> {
/// let writer = tokio::io::sink();
///
/// // Create a Gzip encoder that wraps the writer.
/// let mut encoder = GzipEncoder::new(writer);
///
/// // Copy data from the reader to the encoder, compressing it.
/// tokio::io::copy(&mut reader, &mut encoder).await?;
///
/// Ok(())
///}
///
/// #[tokio::main]
/// async fn main() -> Result<(), std::io::Error> {
/// // Example: In-memory data.
/// let data = b"Hello, world!"; // A byte slice.
/// let reader = Cursor::new(data); // Create an in-memory AsyncRead.
/// compress_data(reader).await?;
///
/// Ok(())
/// }
/// ```
///
///
/// ## Example 3: Parsing Data Formats
///
///
/// `SyncIoBridge` is not ideal when parsing data formats such as `JSON`, as it
/// blocks async operations. A more efficient approach is to read data asynchronously
/// into memory and then `deserialize` it, avoiding unnecessary synchronization overhead.
///
/// Explanation: This example shows how to asynchronously read data into memory
/// and then parse it as `JSON`. By avoiding `SyncIoBridge`, the asynchronous runtime
/// remains unblocked, leading to better performance when working with asynchronous
/// I/O streams.
///
/// ```rust,no_run
/// use tokio::io::AsyncRead;
/// use tokio::io::AsyncReadExt;
/// use std::io::Cursor;
/// # mod serde {
/// # pub trait DeserializeOwned: 'static {}
/// # impl<T: 'static> DeserializeOwned for T {}
/// # }
/// # mod serde_json {
/// # use super::serde::DeserializeOwned;
/// # pub fn from_slice<T: DeserializeOwned>(_: &[u8]) -> Result<T, std::io::Error> {
/// # unimplemented!()
/// # }
/// # }
/// # #[derive(Debug)] struct MyStruct;
///
///
/// async fn parse_json(mut reader: impl AsyncRead + Unpin) -> Result<MyStruct, std::io::Error> {
/// // Read all data from the reader into a Vec<u8>.
/// let mut data = Vec::new();
/// reader.read_to_end(&mut data).await?;
///
/// // Deserialize the data from the Vec<u8> into a MyStruct instance.
/// let value: MyStruct = serde_json::from_slice(&data)?;
///
/// Ok(value)
///}
///
/// #[tokio::main]
/// async fn main() -> Result<(), std::io::Error> {
/// // Example: In-memory data.
/// let data = b"Hello, world!"; // A byte slice.
/// let reader = Cursor::new(data); // Create an in-memory AsyncRead.
/// parse_json(reader).await?;
/// Ok(())
/// }
/// ```
///
/// ## Correct Usage of `SyncIoBridge` inside `spawn_blocking`
///
/// `SyncIoBridge` is mainly useful when you need to interface with synchronous
/// libraries from an asynchronous context.
///
/// Explanation: This example shows how to use `SyncIoBridge` inside a `spawn_blocking`
/// task to safely perform synchronous I/O without blocking the async runtime. The
/// `spawn_blocking` ensures that the synchronous code is offloaded to a dedicated
/// thread pool, preventing it from interfering with the async tasks.
///
/// ```rust
/// use tokio::task::spawn_blocking;
/// use tokio_util::io::SyncIoBridge;
/// use tokio::io::AsyncRead;
/// use std::marker::Unpin;
/// use std::io::Cursor;
///
/// /// Wraps an async reader with `SyncIoBridge` and performs synchronous I/O operations in a blocking task.
/// async fn process_sync_io(reader: impl AsyncRead + Unpin + Send + 'static) -> Result<Vec<u8>, std::io::Error> {
/// // Wrap the async reader with `SyncIoBridge` to allow synchronous reading.
/// let mut sync_reader = SyncIoBridge::new(reader);
///
/// // Spawn a blocking task to perform synchronous I/O operations.
/// let result = spawn_blocking(move || {
/// // Create an in-memory buffer to hold the copied data.
/// let mut buffer = Vec::new();
/// // Copy data from the sync_reader to the buffer.
/// std::io::copy(&mut sync_reader, &mut buffer)?;
/// // Return the buffer containing the copied data.
/// Ok::<_, std::io::Error>(buffer)
/// })
/// .await??;
///
/// // Return the result from the blocking task.
/// Ok(result)
///}
///
/// #[tokio::main]
/// async fn main() -> Result<(), std::io::Error> {
/// // Example: In-memory data.
/// let data = b"Hello, world!"; // A byte slice.
/// let reader = Cursor::new(data); // Create an in-memory AsyncRead.
/// let result = process_sync_io(reader).await?;
///
/// // You can use `result` here as needed.
///
/// Ok(())
/// }
/// ```
///
#[derive(Debug)]
pub struct SyncIoBridge<T> {
src: T,
@@ -154,3 +405,15 @@ impl<T: Unpin> SyncIoBridge<T> {
self.src
}
}
impl<T> AsMut<T> for SyncIoBridge<T> {
fn as_mut(&mut self) -> &mut T {
&mut self.src
}
}
impl<T> AsRef<T> for SyncIoBridge<T> {
fn as_ref(&self) -> &T {
&self.src
}
}
-1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![allow(clippy::needless_doctest_main)]
#![warn(
missing_debug_implementations,
+4 -3
View File
@@ -5,13 +5,13 @@ use tokio::{io::ReadBuf, net::UdpSocket};
use bytes::{BufMut, BytesMut};
use futures_sink::Sink;
use std::io;
use std::pin::Pin;
use std::task::{ready, Context, Poll};
use std::{
borrow::Borrow,
net::{Ipv4Addr, SocketAddr, SocketAddrV4},
};
use std::{io, mem::MaybeUninit};
/// A unified [`Stream`] and [`Sink`] interface to an underlying `UdpSocket`, using
/// the `Encoder` and `Decoder` traits to encode and decode frames.
@@ -83,7 +83,7 @@ where
let addr = {
// Safety: `chunk_mut()` returns a `&mut UninitSlice`, and `UninitSlice` is a
// transparent wrapper around `[MaybeUninit<u8>]`.
let buf = unsafe { &mut *(pin.rd.chunk_mut() as *mut _ as *mut [MaybeUninit<u8>]) };
let buf = unsafe { pin.rd.chunk_mut().as_uninit_slice_mut() };
let mut read = ReadBuf::uninit(buf);
let ptr = read.filled().as_ptr();
let res = ready!(pin.socket.borrow().poll_recv_from(cx, &mut read));
@@ -91,9 +91,10 @@ where
assert_eq!(ptr, read.filled().as_ptr());
let addr = res?;
let filled = read.filled().len();
// Safety: This is guaranteed to be the number of initialized (and read) bytes due
// to the invariants provided by `ReadBuf::filled`.
unsafe { pin.rd.advance_mut(read.filled().len()) };
unsafe { pin.rd.advance_mut(filled) };
addr
};
+1 -2
View File
@@ -2,7 +2,6 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use bytes::{Buf, BufMut};
use std::io::{self, IoSlice};
use std::mem::MaybeUninit;
use std::pin::Pin;
use std::task::{ready, Context, Poll};
@@ -59,7 +58,7 @@ pub fn poll_read_buf<T: AsyncRead + ?Sized, B: BufMut>(
// Safety: `chunk_mut()` returns a `&mut UninitSlice`, and `UninitSlice` is a
// transparent wrapper around `[MaybeUninit<u8>]`.
let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
let dst = unsafe { dst.as_uninit_slice_mut() };
let mut buf = ReadBuf::uninit(dst);
let ptr = buf.filled().as_ptr();
ready!(io.poll_read(cx, &mut buf)?);
-1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "rt", tokio_unstable))]
+101
View File
@@ -1,3 +1,104 @@
# 1.44.2 (April 4, 2025)
This release fixes a soundness issue in the broadcast channel. The channel
accepts values that are `Send` but `!Sync`. Previously, the channel called
`clone()` on these values without synchronizing. This release fixes the channel
by synchronizing calls to `.clone()` (Thanks Austin Bonander for finding and
reporting the issue).
### Fixed
- sync: synchronize `clone()` call in broadcast channel ([#7232])
[#7232]: https://github.com/tokio-rs/tokio/pull/7232
# 1.44.1 (March 13th, 2025)
### Fixed
- rt: skip defer queue in `block_in_place` context ([#7216])
[#7216]: https://github.com/tokio-rs/tokio/pull/7216
# 1.44.0 (March 7th, 2025)
This release changes the `from_std` method on sockets to panic if a blocking
socket is provided. We determined this change is not a breaking change as Tokio is not
intended to operate using blocking sockets. Doing so results in runtime hangs and
should be considered a bug. Accidentally passing a blocking socket to Tokio is one
of the most common user mistakes. If this change causes an issue for you, please
comment on [#7172].
### Added
- coop: add `task::coop` module ([#7116])
- process: add `Command::get_kill_on_drop()` ([#7086])
- sync: add `broadcast::Sender::closed` ([#6685], [#7090])
- sync: add `broadcast::WeakSender` ([#7100])
- sync: add `oneshot::Receiver::is_empty()` ([#7153])
- sync: add `oneshot::Receiver::is_terminated()` ([#7152])
### Fixed
- fs: empty reads on `File` should not start a background read ([#7139])
- process: calling `start_kill` on exited child should not fail ([#7160])
- signal: fix `CTRL_CLOSE`, `CTRL_LOGOFF`, `CTRL_SHUTDOWN` on windows ([#7122])
- sync: properly handle panic during mpsc drop ([#7094])
### Changes
- runtime: clean up magic number in registration set ([#7112])
- coop: make coop yield using waker defer strategy ([#7185])
- macros: make `select!` budget-aware ([#7164])
- net: panic when passing a blocking socket to `from_std` ([#7166])
- io: clean up buffer casts ([#7142])
### Changes to unstable APIs
- rt: add before and after task poll callbacks ([#7120])
- tracing: make the task tracing API unstable public ([#6972])
### Documented
- docs: fix nesting of sections in top-level docs ([#7159])
- fs: rename symlink and hardlink parameter names ([#7143])
- io: swap reader/writer in simplex doc test ([#7176])
- macros: docs about `select!` alternatives ([#7110])
- net: rename the argument for `send_to` ([#7146])
- process: add example for reading `Child` stdout ([#7141])
- process: clarify `Child::kill` behavior ([#7162])
- process: fix grammar of the `ChildStdin` struct doc comment ([#7192])
- runtime: consistently use `worker_threads` instead of `core_threads` ([#7186])
[#6685]: https://github.com/tokio-rs/tokio/pull/6685
[#6972]: https://github.com/tokio-rs/tokio/pull/6972
[#7086]: https://github.com/tokio-rs/tokio/pull/7086
[#7090]: https://github.com/tokio-rs/tokio/pull/7090
[#7094]: https://github.com/tokio-rs/tokio/pull/7094
[#7100]: https://github.com/tokio-rs/tokio/pull/7100
[#7110]: https://github.com/tokio-rs/tokio/pull/7110
[#7112]: https://github.com/tokio-rs/tokio/pull/7112
[#7116]: https://github.com/tokio-rs/tokio/pull/7116
[#7120]: https://github.com/tokio-rs/tokio/pull/7120
[#7122]: https://github.com/tokio-rs/tokio/pull/7122
[#7139]: https://github.com/tokio-rs/tokio/pull/7139
[#7141]: https://github.com/tokio-rs/tokio/pull/7141
[#7142]: https://github.com/tokio-rs/tokio/pull/7142
[#7143]: https://github.com/tokio-rs/tokio/pull/7143
[#7146]: https://github.com/tokio-rs/tokio/pull/7146
[#7152]: https://github.com/tokio-rs/tokio/pull/7152
[#7153]: https://github.com/tokio-rs/tokio/pull/7153
[#7159]: https://github.com/tokio-rs/tokio/pull/7159
[#7160]: https://github.com/tokio-rs/tokio/pull/7160
[#7162]: https://github.com/tokio-rs/tokio/pull/7162
[#7164]: https://github.com/tokio-rs/tokio/pull/7164
[#7166]: https://github.com/tokio-rs/tokio/pull/7166
[#7172]: https://github.com/tokio-rs/tokio/pull/7172
[#7176]: https://github.com/tokio-rs/tokio/pull/7176
[#7185]: https://github.com/tokio-rs/tokio/pull/7185
[#7186]: https://github.com/tokio-rs/tokio/pull/7186
[#7192]: https://github.com/tokio-rs/tokio/pull/7192
# 1.43.1 (April 2nd, 2025)
This release fixes a soundness issue in the broadcast channel. The channel
+3 -2
View File
@@ -6,7 +6,7 @@ name = "tokio"
# - README.md
# - Update CHANGELOG.md.
# - Create "v1.x.y" git tag.
version = "1.43.1"
version = "1.44.2"
edition = "2021"
rust-version = "1.70"
authors = ["Tokio Contributors <[email protected]>"]
@@ -91,7 +91,7 @@ tokio-macros = { version = "~2.5.0", path = "../tokio-macros", optional = true }
pin-project-lite = "0.2.11"
# Everything else is optional...
bytes = { version = "1.1.0", optional = true }
bytes = { version = "1.2.1", optional = true }
mio = { version = "1.0.1", optional = true, default-features = false }
parking_lot = { version = "0.12.0", optional = true }
@@ -133,6 +133,7 @@ tokio-stream = { version = "0.1", path = "../tokio-stream" }
futures = { version = "0.3.0", features = ["async-await"] }
mockall = "0.11.1"
async-stream = "0.3"
futures-concurrency = "7.6.3"
[target.'cfg(not(target_family = "wasm"))'.dev-dependencies]
socket2 = "0.5.5"
+6 -5
View File
@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
```toml
[dependencies]
tokio = { version = "1.43.1", features = ["full"] }
tokio = { version = "1.44.2", features = ["full"] }
```
Then, on your main.rs:
@@ -205,8 +205,8 @@ works with the MSRV of that minor release.
## Release schedule
Tokio doesn't follow a fixed release schedule, but we typically make one to two
new minor releases each month. We make patch releases for bugfixes as necessary.
Tokio doesn't follow a fixed release schedule, but we typically make one minor
release each month. We make patch releases for bugfixes as necessary.
## Bug patching policy
@@ -216,8 +216,8 @@ warrants a patch release with a fix for the bug, it will be backported and
released as a new patch release for each LTS minor version. Our current LTS
releases are:
* `1.36.x` - LTS release until March 2025. (MSRV 1.63)
* `1.38.x` - LTS release until July 2025. (MSRV 1.63)
* `1.43.x` - LTS release until March 2026. (MSRV 1.70)
Each LTS release will continue to receive backported fixes for at least a year.
If you wish to use a fixed minor release in your project, we recommend that you
@@ -227,7 +227,7 @@ To use a fixed minor version, you can specify the version with a tilde. For
example, to specify that you wish to use the newest `1.32.x` patch release, you
can use the following dependency specification:
```text
tokio = { version = "~1.32", features = [...] }
tokio = { version = "~1.38", features = [...] }
```
### Previous LTS releases
@@ -238,6 +238,7 @@ tokio = { version = "~1.32", features = [...] }
* `1.20.x` - LTS release until September 2023.
* `1.25.x` - LTS release until March 2024.
* `1.32.x` - LTS release until September 2024.
* `1.36.x` - LTS release until March 2025.
## License
+7 -7
View File
@@ -24,21 +24,21 @@ pub enum NotDefinedHere {}
impl mio::event::Source for NotDefinedHere {
fn register(
&mut self,
registry: &mio::Registry,
token: mio::Token,
interests: mio::Interest,
_registry: &mio::Registry,
_token: mio::Token,
_interests: mio::Interest,
) -> std::io::Result<()> {
Ok(())
}
fn reregister(
&mut self,
registry: &mio::Registry,
token: mio::Token,
interests: mio::Interest,
_registry: &mio::Registry,
_token: mio::Token,
_interests: mio::Interest,
) -> std::io::Result<()> {
Ok(())
}
fn deregister(&mut self, registry: &mio::Registry) -> std::io::Result<()> {
fn deregister(&mut self, _registry: &mio::Registry) -> std::io::Result<()> {
Ok(())
}
}
+2 -1
View File
@@ -587,6 +587,7 @@ impl AsyncRead for File {
dst: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
ready!(crate::trace::trace_leaf(cx));
let me = self.get_mut();
let inner = me.inner.get_mut();
@@ -595,7 +596,7 @@ impl AsyncRead for File {
State::Idle(ref mut buf_cell) => {
let mut buf = buf_cell.take().unwrap();
if !buf.is_empty() {
if !buf.is_empty() || dst.remaining() == 0 {
buf.copy_to(dst);
*buf_cell = Some(buf);
return Poll::Ready(Ok(()));
+6 -6
View File
@@ -7,7 +7,7 @@ use std::path::Path;
///
/// This is an async version of [`std::fs::hard_link`].
///
/// The `dst` path will be a link pointing to the `src` path. Note that systems
/// The `link` path will be a link pointing to the `original` path. Note that systems
/// often require these two paths to both be located on the same filesystem.
///
/// # Platform-specific behavior
@@ -23,7 +23,7 @@ use std::path::Path;
/// This function will return an error in the following situations, but is not
/// limited to just these cases:
///
/// * The `src` path is not a file or doesn't exist.
/// * The `original` path is not a file or doesn't exist.
///
/// # Examples
///
@@ -36,9 +36,9 @@ use std::path::Path;
/// Ok(())
/// }
/// ```
pub async fn hard_link(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
let src = src.as_ref().to_owned();
let dst = dst.as_ref().to_owned();
pub async fn hard_link(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
let original = original.as_ref().to_owned();
let link = link.as_ref().to_owned();
asyncify(move || std::fs::hard_link(src, dst)).await
asyncify(move || std::fs::hard_link(original, link)).await
}
+5 -5
View File
@@ -5,12 +5,12 @@ use std::path::Path;
/// Creates a new symbolic link on the filesystem.
///
/// The `dst` path will be a symbolic link pointing to the `src` path.
/// The `link` path will be a symbolic link pointing to the `original` path.
///
/// This is an async version of [`std::os::unix::fs::symlink`].
pub async fn symlink(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
let src = src.as_ref().to_owned();
let dst = dst.as_ref().to_owned();
pub async fn symlink(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
let original = original.as_ref().to_owned();
let link = link.as_ref().to_owned();
asyncify(move || std::os::unix::fs::symlink(src, dst)).await
asyncify(move || std::os::unix::fs::symlink(original, link)).await
}
+5 -5
View File
@@ -5,15 +5,15 @@ use std::path::Path;
/// Creates a new directory symlink on the filesystem.
///
/// The `dst` path will be a directory symbolic link pointing to the `src`
/// The `link` path will be a directory symbolic link pointing to the `original`
/// path.
///
/// This is an async version of [`std::os::windows::fs::symlink_dir`][std]
///
/// [std]: https://doc.rust-lang.org/std/os/windows/fs/fn.symlink_dir.html
pub async fn symlink_dir(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
let src = src.as_ref().to_owned();
let dst = dst.as_ref().to_owned();
pub async fn symlink_dir(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
let original = original.as_ref().to_owned();
let link = link.as_ref().to_owned();
asyncify(move || std::os::windows::fs::symlink_dir(src, dst)).await
asyncify(move || std::os::windows::fs::symlink_dir(original, link)).await
}
+5 -5
View File
@@ -5,15 +5,15 @@ use std::path::Path;
/// Creates a new file symbolic link on the filesystem.
///
/// The `dst` path will be a file symbolic link pointing to the `src`
/// The `link` path will be a file symbolic link pointing to the `original`
/// path.
///
/// This is an async version of [`std::os::windows::fs::symlink_file`][std]
///
/// [std]: https://doc.rust-lang.org/std/os/windows/fs/fn.symlink_file.html
pub async fn symlink_file(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
let src = src.as_ref().to_owned();
let dst = dst.as_ref().to_owned();
pub async fn symlink_file(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
let original = original.as_ref().to_owned();
let link = link.as_ref().to_owned();
asyncify(move || std::os::windows::fs::symlink_file(src, dst)).await
asyncify(move || std::os::windows::fs::symlink_file(original, link)).await
}
+1 -1
View File
@@ -94,7 +94,7 @@ impl CopyBuffer {
feature = "time",
))]
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
loop {
// If there is some space left in our buffer, then we try to read some
// data to continue, thus maximizing the chances of a large write.
+5 -5
View File
@@ -197,8 +197,8 @@ impl Drop for DuplexStream {
/// ```
/// # async fn ex() -> std::io::Result<()> {
/// # use tokio::io::{AsyncReadExt, AsyncWriteExt};
/// let (writer, reader) = tokio::io::simplex(64);
/// let mut simplex_stream = writer.unsplit(reader);
/// let (reader, writer) = tokio::io::simplex(64);
/// let mut simplex_stream = reader.unsplit(writer);
/// simplex_stream.write_all(b"hello").await?;
///
/// let mut buf = [0u8; 5];
@@ -332,7 +332,7 @@ impl AsyncRead for SimplexStream {
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
ready!(crate::trace::trace_leaf(cx));
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let ret = self.poll_read_internal(cx, buf);
if ret.is_ready() {
@@ -362,7 +362,7 @@ impl AsyncWrite for SimplexStream {
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
ready!(crate::trace::trace_leaf(cx));
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let ret = self.poll_write_internal(cx, buf);
if ret.is_ready() {
@@ -390,7 +390,7 @@ impl AsyncWrite for SimplexStream {
bufs: &[std::io::IoSlice<'_>],
) -> Poll<Result<usize, std::io::Error>> {
ready!(crate::trace::trace_leaf(cx));
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let ret = self.poll_write_vectored_internal(cx, bufs);
if ret.is_ready() {
+1 -1
View File
@@ -88,7 +88,7 @@ cfg_io_util! {
cfg_coop! {
fn poll_proceed_and_make_progress(cx: &mut std::task::Context<'_>) -> std::task::Poll<()> {
let coop = std::task::ready!(crate::runtime::coop::poll_proceed(cx));
let coop = std::task::ready!(crate::task::coop::poll_proceed(cx));
coop.made_progress();
std::task::Poll::Ready(())
}
+1 -2
View File
@@ -41,7 +41,6 @@ where
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
use crate::io::ReadBuf;
use std::mem::MaybeUninit;
let me = self.project();
@@ -51,7 +50,7 @@ where
let n = {
let dst = me.buf.chunk_mut();
let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
let dst = unsafe { dst.as_uninit_slice_mut() };
let mut buf = ReadBuf::uninit(dst);
let ptr = buf.filled().as_ptr();
ready!(Pin::new(me.reader).poll_read(cx, &mut buf)?);
+5 -6
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![allow(
clippy::cognitive_complexity,
clippy::large_enum_variant,
@@ -303,7 +302,7 @@
//! }
//! ```
//!
//! ## Feature flags
//! # Feature flags
//!
//! Tokio uses a set of [feature flags] to reduce the amount of compiled code. It
//! is possible to just enable certain features over others. By default, Tokio
@@ -341,7 +340,7 @@
//! _Note: `AsyncRead` and `AsyncWrite` traits do not require any features and are
//! always available._
//!
//! ### Unstable features
//! ## Unstable features
//!
//! Some feature flags are only available when specifying the `tokio_unstable` flag:
//!
@@ -393,7 +392,7 @@
//! [unstable features]: https://internals.rust-lang.org/t/feature-request-unstable-opt-in-non-transitive-crate-features/16193#why-not-a-crate-feature-2
//! [feature flags]: https://doc.rust-lang.org/cargo/reference/manifest.html#the-features-section
//!
//! ## Supported platforms
//! # Supported platforms
//!
//! Tokio currently guarantees support for the following platforms:
//!
@@ -419,7 +418,7 @@
//!
//! [mio-supported]: https://crates.io/crates/mio#platforms
//!
//! ### `WASM` support
//! ## `WASM` support
//!
//! Tokio has some limited support for the `WASM` platform. Without the
//! `tokio_unstable` flag, the following features are supported:
@@ -441,7 +440,7 @@
//! immediately instead of blocking forever. On platforms that don't support
//! time, this means that the runtime can never be idle in any way.
//!
//! ### Unstable `WASM` support
//! ## Unstable `WASM` support
//!
//! Tokio also has unstable support for some additional `WASM` features. This
//! requires the use of the `tokio_unstable` flag.
+1 -1
View File
@@ -1,7 +1,7 @@
use std::sync::{self, MutexGuard, TryLockError};
/// Adapter for `std::Mutex` that removes the poisoning aspects
/// from its api.
/// from its API.
#[derive(Debug)]
pub(crate) struct Mutex<T: ?Sized>(sync::Mutex<T>);
+151
View File
@@ -398,6 +398,153 @@ macro_rules! doc {
/// }
/// }
/// ```
/// # Alternatives from the Ecosystem
///
/// The `select!` macro is a powerful tool for managing multiple asynchronous
/// branches, enabling tasks to run concurrently within the same thread. However,
/// its use can introduce challenges, particularly around cancellation safety, which
/// can lead to subtle and hard-to-debug errors. For many use cases, ecosystem
/// alternatives may be preferable as they mitigate these concerns by offering
/// clearer syntax, more predictable control flow, and reducing the need to manually
/// handle issues like fuse semantics or cancellation safety.
///
/// ## Merging Streams
///
/// For cases where `loop { select! { ... } }` is used to poll multiple tasks,
/// stream merging offers a concise alternative, inherently handle cancellation-safe
/// processing, removing the risk of data loss. Libraries such as [`tokio_stream`],
/// [`futures::stream`] and [`futures_concurrency`] provide tools for merging
/// streams and handling their outputs sequentially.
///
/// [`tokio_stream`]: https://docs.rs/tokio-stream/latest/tokio_stream/
/// [`futures::stream`]: https://docs.rs/futures/latest/futures/stream/
/// [`futures_concurrency`]: https://docs.rs/futures-concurrency/latest/futures_concurrency/
///
/// ### Example with `select!`
///
/// ```
/// struct File;
/// struct Channel;
/// struct Socket;
///
/// impl Socket {
/// async fn read_packet(&mut self) -> Vec<u8> {
/// vec![]
/// }
/// }
///
/// async fn read_send(_file: &mut File, _channel: &mut Channel) {
/// // do work that is not cancel safe
/// }
///
/// #[tokio::main]
/// async fn main() {
/// // open our IO types
/// let mut file = File;
/// let mut channel = Channel;
/// let mut socket = Socket;
///
/// loop {
/// tokio::select! {
/// _ = read_send(&mut file, &mut channel) => { /* ... */ },
/// _data = socket.read_packet() => { /* ... */ }
/// _ = futures::future::ready(()) => break
/// }
/// }
/// }
///
/// ```
///
/// ### Moving to `merge`
///
/// By using merge, you can unify multiple asynchronous tasks into a single stream,
/// eliminating the need to manage tasks manually and reducing the risk of
/// unintended behavior like data loss.
///
/// ```
/// use std::pin::pin;
///
/// use futures::stream::unfold;
/// use tokio_stream::StreamExt;
///
/// struct File;
/// struct Channel;
/// struct Socket;
///
/// impl Socket {
/// async fn read_packet(&mut self) -> Vec<u8> {
/// vec![]
/// }
/// }
///
/// async fn read_send(_file: &mut File, _channel: &mut Channel) {
/// // do work that is not cancel safe
/// }
///
/// enum Message {
/// Stop,
/// Sent,
/// Data(Vec<u8>),
/// }
///
/// #[tokio::main]
/// async fn main() {
/// // open our IO types
/// let file = File;
/// let channel = Channel;
/// let socket = Socket;
///
/// let a = unfold((file, channel), |(mut file, mut channel)| async {
/// read_send(&mut file, &mut channel).await;
/// Some((Message::Sent, (file, channel)))
/// });
/// let b = unfold(socket, |mut socket| async {
/// let data = socket.read_packet().await;
/// Some((Message::Data(data), socket))
/// });
/// let c = tokio_stream::iter([Message::Stop]);
///
/// let mut s = pin!(a.merge(b).merge(c));
/// while let Some(msg) = s.next().await {
/// match msg {
/// Message::Data(_data) => { /* ... */ }
/// Message::Sent => continue,
/// Message::Stop => break,
/// }
/// }
/// }
/// ```
///
/// ## Racing Futures
///
/// If you need to wait for the first completion among several asynchronous tasks,
/// ecosystem utilities such as
/// [`futures`](https://docs.rs/futures/latest/futures/),
/// [`futures-lite`](https://docs.rs/futures-lite/latest/futures_lite/) or
/// [`futures-concurrency`](https://docs.rs/futures-concurrency/latest/futures_concurrency/)
/// provide streamlined syntax for racing futures:
///
/// - [`futures_concurrency::future::Race`](https://docs.rs/futures-concurrency/latest/futures_concurrency/future/trait.Race.html)
/// - [`futures::select`](https://docs.rs/futures/latest/futures/macro.select.html)
/// - [`futures::stream::select_all`](https://docs.rs/futures/latest/futures/stream/select_all/index.html) (for streams)
/// - [`futures_lite::future::or`](https://docs.rs/futures-lite/latest/futures_lite/future/fn.or.html)
/// - [`futures_lite::future::race`](https://docs.rs/futures-lite/latest/futures_lite/future/fn.race.html)
///
/// ```
/// use futures_concurrency::future::Race;
///
/// #[tokio::main]
/// async fn main() {
/// let task_a = async { Ok("ok") };
/// let task_b = async { Err("error") };
/// let result = (task_a, task_b).race().await;
///
/// match result {
/// Ok(output) => println!("First task completed with: {output}"),
/// Err(err) => eprintln!("Error occurred: {err}"),
/// }
/// }
/// ```
#[macro_export]
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
$select
@@ -513,6 +660,10 @@ doc! {macro_rules! select {
let mut futures = &mut futures;
$crate::macros::support::poll_fn(|cx| {
// Return `Pending` when the task budget is depleted since budget-aware futures
// are going to yield anyway and other futures will not cooperate.
::std::task::ready!($crate::macros::support::poll_budget_available(cx));
// Track if any branch returns pending. If no branch completes
// **or** returns pending, this implies that all branches are
// disabled.
+17 -1
View File
@@ -7,8 +7,24 @@ cfg_macros! {
pub fn thread_rng_n(n: u32) -> u32 {
crate::runtime::context::thread_rng_n(n)
}
cfg_coop! {
#[doc(hidden)]
#[inline]
pub fn poll_budget_available(cx: &mut Context<'_>) -> Poll<()> {
crate::task::coop::poll_budget_available(cx)
}
}
cfg_not_coop! {
#[doc(hidden)]
#[inline]
pub fn poll_budget_available(_: &mut Context<'_>) -> Poll<()> {
Poll::Ready(())
}
}
}
pub use std::future::{Future, IntoFuture};
pub use std::pin::Pin;
pub use std::task::Poll;
pub use std::task::{Context, Poll};
+7
View File
@@ -1,5 +1,6 @@
use crate::io::{Interest, PollEvented};
use crate::net::tcp::TcpStream;
use crate::util::check_socket_for_blocking;
cfg_not_wasi! {
use crate::net::{to_socket_addrs, ToSocketAddrs};
@@ -209,6 +210,10 @@ impl TcpListener {
/// will block the thread, which will cause unexpected behavior.
/// Non-blocking mode can be set using [`set_nonblocking`].
///
/// Passing a listener in blocking mode is always erroneous,
/// and the behavior in that case may change in the future.
/// For example, it could panic.
///
/// [`set_nonblocking`]: std::net::TcpListener::set_nonblocking
///
/// # Examples
@@ -236,6 +241,8 @@ impl TcpListener {
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
#[track_caller]
pub fn from_std(listener: net::TcpListener) -> io::Result<TcpListener> {
check_socket_for_blocking(&listener)?;
let io = mio::net::TcpListener::from_std(listener);
let io = PollEvented::new(io)?;
Ok(TcpListener { io })
+7
View File
@@ -7,6 +7,7 @@ cfg_not_wasi! {
use crate::io::{AsyncRead, AsyncWrite, Interest, PollEvented, ReadBuf, Ready};
use crate::net::tcp::split::{split, ReadHalf, WriteHalf};
use crate::net::tcp::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
use crate::util::check_socket_for_blocking;
use std::fmt;
use std::io;
@@ -173,6 +174,10 @@ impl TcpStream {
/// will block the thread, which will cause unexpected behavior.
/// Non-blocking mode can be set using [`set_nonblocking`].
///
/// Passing a listener in blocking mode is always erroneous,
/// and the behavior in that case may change in the future.
/// For example, it could panic.
///
/// [`set_nonblocking`]: std::net::TcpStream::set_nonblocking
///
/// # Examples
@@ -200,6 +205,8 @@ impl TcpStream {
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
#[track_caller]
pub fn from_std(stream: std::net::TcpStream) -> io::Result<TcpStream> {
check_socket_for_blocking(&stream)?;
let io = mio::net::TcpStream::from_std(stream);
let io = PollEvented::new(io)?;
Ok(TcpStream { io })
+9 -2
View File
@@ -1,5 +1,6 @@
use crate::io::{Interest, PollEvented, ReadBuf, Ready};
use crate::net::{to_socket_addrs, ToSocketAddrs};
use crate::util::check_socket_for_blocking;
use std::fmt;
use std::io;
@@ -192,6 +193,10 @@ impl UdpSocket {
/// will block the thread, which will cause unexpected behavior.
/// Non-blocking mode can be set using [`set_nonblocking`].
///
/// Passing a listener in blocking mode is always erroneous,
/// and the behavior in that case may change in the future.
/// For example, it could panic.
///
/// [`set_nonblocking`]: std::net::UdpSocket::set_nonblocking
///
/// # Panics
@@ -220,6 +225,8 @@ impl UdpSocket {
/// ```
#[track_caller]
pub fn from_std(socket: net::UdpSocket) -> io::Result<UdpSocket> {
check_socket_for_blocking(&socket)?;
let io = mio::net::UdpSocket::from_std(socket);
UdpSocket::new(io)
}
@@ -1164,8 +1171,8 @@ impl UdpSocket {
/// Ok(())
/// }
/// ```
pub async fn send_to<A: ToSocketAddrs>(&self, buf: &[u8], target: A) -> io::Result<usize> {
let mut addrs = to_socket_addrs(target).await?;
pub async fn send_to<A: ToSocketAddrs>(&self, buf: &[u8], addr: A) -> io::Result<usize> {
let mut addrs = to_socket_addrs(addr).await?;
match addrs.next() {
Some(target) => self.send_to_addr(buf, target).await,
+7
View File
@@ -1,5 +1,6 @@
use crate::io::{Interest, PollEvented, ReadBuf, Ready};
use crate::net::unix::SocketAddr;
use crate::util::check_socket_for_blocking;
use std::fmt;
use std::io;
@@ -449,6 +450,10 @@ impl UnixDatagram {
/// will block the thread, which will cause unexpected behavior.
/// Non-blocking mode can be set using [`set_nonblocking`].
///
/// Passing a listener in blocking mode is always erroneous,
/// and the behavior in that case may change in the future.
/// For example, it could panic.
///
/// [`set_nonblocking`]: std::os::unix::net::UnixDatagram::set_nonblocking
///
/// # Panics
@@ -484,6 +489,8 @@ impl UnixDatagram {
/// ```
#[track_caller]
pub fn from_std(datagram: net::UnixDatagram) -> io::Result<UnixDatagram> {
check_socket_for_blocking(&datagram)?;
let socket = mio::net::UnixDatagram::from_std(datagram);
let io = PollEvented::new(socket)?;
Ok(UnixDatagram { io })
+7
View File
@@ -1,5 +1,6 @@
use crate::io::{Interest, PollEvented};
use crate::net::unix::{SocketAddr, UnixStream};
use crate::util::check_socket_for_blocking;
use std::fmt;
use std::io;
@@ -106,6 +107,10 @@ impl UnixListener {
/// will block the thread, which will cause unexpected behavior.
/// Non-blocking mode can be set using [`set_nonblocking`].
///
/// Passing a listener in blocking mode is always erroneous,
/// and the behavior in that case may change in the future.
/// For example, it could panic.
///
/// [`set_nonblocking`]: std::os::unix::net::UnixListener::set_nonblocking
///
/// # Examples
@@ -133,6 +138,8 @@ impl UnixListener {
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
#[track_caller]
pub fn from_std(listener: net::UnixListener) -> io::Result<UnixListener> {
check_socket_for_blocking(&listener)?;
let listener = mio::net::UnixListener::from_std(listener);
let io = PollEvented::new(listener)?;
Ok(UnixListener { io })
+7
View File
@@ -3,6 +3,7 @@ use crate::net::unix::split::{split, ReadHalf, WriteHalf};
use crate::net::unix::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
use crate::net::unix::ucred::{self, UCred};
use crate::net::unix::SocketAddr;
use crate::util::check_socket_for_blocking;
use std::fmt;
use std::future::poll_fn;
@@ -791,6 +792,10 @@ impl UnixStream {
/// will block the thread, which will cause unexpected behavior.
/// Non-blocking mode can be set using [`set_nonblocking`].
///
/// Passing a listener in blocking mode is always erroneous,
/// and the behavior in that case may change in the future.
/// For example, it could panic.
///
/// [`set_nonblocking`]: std::os::unix::net::UnixStream::set_nonblocking
///
/// # Examples
@@ -818,6 +823,8 @@ impl UnixStream {
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
#[track_caller]
pub fn from_std(stream: net::UnixStream) -> io::Result<UnixStream> {
check_socket_for_blocking(&stream)?;
let stream = mio::net::UnixStream::from_std(stream);
let io = PollEvented::new(stream)?;
+71 -7
View File
@@ -981,6 +981,26 @@ impl Command {
async { child?.wait_with_output().await }
}
/// Returns the boolean value that was previously set by [`Command::kill_on_drop`].
///
/// Note that if you have not previously called [`Command::kill_on_drop`], the
/// default value of `false` will be returned here.
///
/// # Examples
///
/// ```
/// use tokio::process::Command;
///
/// let mut cmd = Command::new("echo");
/// assert!(!cmd.get_kill_on_drop());
///
/// cmd.kill_on_drop(true);
/// assert!(cmd.get_kill_on_drop());
/// ```
pub fn get_kill_on_drop(&self) -> bool {
self.kill_on_drop
}
}
impl From<StdCommand> for Command {
@@ -1028,7 +1048,7 @@ where
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
ready!(crate::trace::trace_leaf(cx));
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let ret = Pin::new(&mut self.inner).poll(cx);
@@ -1138,16 +1158,20 @@ impl Child {
pub fn start_kill(&mut self) -> io::Result<()> {
match &mut self.child {
FusedChild::Child(child) => child.kill(),
FusedChild::Done(_) => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"invalid argument: can't kill an exited process",
)),
FusedChild::Done(_) => Ok(()),
}
}
/// Forces the child to exit.
///
/// This is equivalent to sending a `SIGKILL` on unix platforms.
/// This is equivalent to sending a `SIGKILL` on unix platforms
/// followed by [`wait`](Child::wait).
///
/// Note: std version of [`Child::kill`](std::process::Child::kill) does not `wait`.
/// For an equivalent of `Child::kill` in the standard library,
/// use [`start_kill`](Child::start_kill).
///
/// # Examples
///
/// If the child has to be killed remotely, it is possible to do it using
/// a combination of the select! macro and a `oneshot` channel. In the following
@@ -1170,6 +1194,46 @@ impl Child {
/// }
/// }
/// ```
///
/// You can also interact with the child's standard I/O. For example, you can
/// read its stdout while waiting for it to exit.
///
/// ```no_run
/// # use std::process::Stdio;
/// #
/// # use tokio::io::AsyncReadExt;
/// # use tokio::process::Command;
/// # use tokio::sync::oneshot::channel;
///
/// #[tokio::main]
/// async fn main() {
/// let (_tx, rx) = channel::<()>();
///
/// let mut child = Command::new("echo")
/// .arg("Hello World!")
/// .stdout(Stdio::piped())
/// .spawn()
/// .unwrap();
///
/// let mut stdout = child.stdout.take().expect("stdout is not captured");
///
/// let read_stdout = tokio::spawn(async move {
/// let mut buff = Vec::new();
/// let _ = stdout.read_to_end(&mut buff).await;
///
/// buff
/// });
///
/// tokio::select! {
/// _ = child.wait() => {}
/// _ = rx => { child.kill().await.expect("kill failed") },
/// }
///
/// let buff = read_stdout.await.unwrap();
///
/// assert_eq!(buff, b"Hello World!\n");
/// }
/// ```
pub async fn kill(&mut self) -> io::Result<()> {
self.start_kill()?;
self.wait().await?;
@@ -1323,7 +1387,7 @@ impl Child {
/// The standard input stream for spawned children.
///
/// This type implements the `AsyncWrite` trait to pass data to the stdin handle of
/// This type implements the `AsyncWrite` trait to pass data to the stdin
/// handle of a child process asynchronously.
#[derive(Debug)]
pub struct ChildStdin {
+1 -1
View File
@@ -128,7 +128,7 @@ pub(crate) struct Task {
#[derive(PartialEq, Eq)]
pub(crate) enum Mandatory {
#[cfg_attr(not(fs), allow(dead_code))]
#[cfg_attr(not(feature = "fs"), allow(dead_code))]
Mandatory,
NonMandatory,
}
+1 -1
View File
@@ -37,7 +37,7 @@ where
// currently goes through Task::poll(), and so is subject to budgeting. That isn't really
// what we want; a blocking task may itself want to run tasks (it might be a Worker!), so
// we want it to start without any budgeting.
crate::runtime::coop::stop();
crate::task::coop::stop();
Poll::Ready(func())
}
+119 -8
View File
@@ -88,6 +88,14 @@ pub struct Builder {
/// To run before each task is spawned.
pub(super) before_spawn: Option<TaskCallback>,
/// To run before each poll
#[cfg(tokio_unstable)]
pub(super) before_poll: Option<TaskCallback>,
/// To run after each poll
#[cfg(tokio_unstable)]
pub(super) after_poll: Option<TaskCallback>,
/// To run after each task is terminated.
pub(super) after_termination: Option<TaskCallback>,
@@ -306,6 +314,11 @@ impl Builder {
before_spawn: None,
after_termination: None,
#[cfg(tokio_unstable)]
before_poll: None,
#[cfg(tokio_unstable)]
after_poll: None,
keep_alive: None,
// Defaults for these values depend on the scheduler kind, so we get them
@@ -743,6 +756,92 @@ impl Builder {
self
}
/// Executes function `f` just before a task is polled
///
/// `f` is called within the Tokio context, so functions like
/// [`tokio::spawn`](crate::spawn) can be called, and may result in this callback being
/// invoked immediately.
///
/// **Note**: This is an [unstable API][unstable]. The public API of this type
/// may break in 1.x releases. See [the documentation on unstable
/// features][unstable] for details.
///
/// [unstable]: crate#unstable-features
///
/// # Examples
///
/// ```
/// # use std::sync::{atomic::AtomicUsize, Arc};
/// # use tokio::task::yield_now;
/// # pub fn main() {
/// let poll_start_counter = Arc::new(AtomicUsize::new(0));
/// let poll_start = poll_start_counter.clone();
/// let rt = tokio::runtime::Builder::new_multi_thread()
/// .enable_all()
/// .on_before_task_poll(move |meta| {
/// println!("task {} is about to be polled", meta.id())
/// })
/// .build()
/// .unwrap();
/// let task = rt.spawn(async {
/// yield_now().await;
/// });
/// let _ = rt.block_on(task);
///
/// # }
/// ```
#[cfg(tokio_unstable)]
pub fn on_before_task_poll<F>(&mut self, f: F) -> &mut Self
where
F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
{
self.before_poll = Some(std::sync::Arc::new(f));
self
}
/// Executes function `f` just after a task is polled
///
/// `f` is called within the Tokio context, so functions like
/// [`tokio::spawn`](crate::spawn) can be called, and may result in this callback being
/// invoked immediately.
///
/// **Note**: This is an [unstable API][unstable]. The public API of this type
/// may break in 1.x releases. See [the documentation on unstable
/// features][unstable] for details.
///
/// [unstable]: crate#unstable-features
///
/// # Examples
///
/// ```
/// # use std::sync::{atomic::AtomicUsize, Arc};
/// # use tokio::task::yield_now;
/// # pub fn main() {
/// let poll_stop_counter = Arc::new(AtomicUsize::new(0));
/// let poll_stop = poll_stop_counter.clone();
/// let rt = tokio::runtime::Builder::new_multi_thread()
/// .enable_all()
/// .on_after_task_poll(move |meta| {
/// println!("task {} completed polling", meta.id());
/// })
/// .build()
/// .unwrap();
/// let task = rt.spawn(async {
/// yield_now().await;
/// });
/// let _ = rt.block_on(task);
///
/// # }
/// ```
#[cfg(tokio_unstable)]
pub fn on_after_task_poll<F>(&mut self, f: F) -> &mut Self
where
F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
{
self.after_poll = Some(std::sync::Arc::new(f));
self
}
/// Executes function `f` just after a task is terminated.
///
/// `f` is called within the Tokio context, so functions like
@@ -1410,6 +1509,10 @@ impl Builder {
before_park: self.before_park.clone(),
after_unpark: self.after_unpark.clone(),
before_spawn: self.before_spawn.clone(),
#[cfg(tokio_unstable)]
before_poll: self.before_poll.clone(),
#[cfg(tokio_unstable)]
after_poll: self.after_poll.clone(),
after_termination: self.after_termination.clone(),
global_queue_interval: self.global_queue_interval,
event_interval: self.event_interval,
@@ -1537,13 +1640,13 @@ cfg_rt_multi_thread! {
use crate::runtime::{Config, runtime::Scheduler};
use crate::runtime::scheduler::{self, MultiThread};
let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
let worker_threads = self.worker_threads.unwrap_or_else(num_cpus);
let (driver, driver_handle) = driver::Driver::new(self.get_cfg(core_threads))?;
let (driver, driver_handle) = driver::Driver::new(self.get_cfg(worker_threads))?;
// Create the blocking pool
let blocking_pool =
blocking::create_blocking_pool(self, self.max_blocking_threads + core_threads);
blocking::create_blocking_pool(self, self.max_blocking_threads + worker_threads);
let blocking_spawner = blocking_pool.spawner().clone();
// Generate a rng seed for this runtime.
@@ -1551,7 +1654,7 @@ cfg_rt_multi_thread! {
let seed_generator_2 = self.seed_generator.next_generator();
let (scheduler, handle, launch) = MultiThread::new(
core_threads,
worker_threads,
driver,
driver_handle,
blocking_spawner,
@@ -1560,6 +1663,10 @@ cfg_rt_multi_thread! {
before_park: self.before_park.clone(),
after_unpark: self.after_unpark.clone(),
before_spawn: self.before_spawn.clone(),
#[cfg(tokio_unstable)]
before_poll: self.before_poll.clone(),
#[cfg(tokio_unstable)]
after_poll: self.after_poll.clone(),
after_termination: self.after_termination.clone(),
global_queue_interval: self.global_queue_interval,
event_interval: self.event_interval,
@@ -1587,12 +1694,12 @@ cfg_rt_multi_thread! {
use crate::runtime::{Config, runtime::Scheduler};
use crate::runtime::scheduler::MultiThreadAlt;
let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
let (driver, driver_handle) = driver::Driver::new(self.get_cfg(core_threads))?;
let worker_threads = self.worker_threads.unwrap_or_else(num_cpus);
let (driver, driver_handle) = driver::Driver::new(self.get_cfg(worker_threads))?;
// Create the blocking pool
let blocking_pool =
blocking::create_blocking_pool(self, self.max_blocking_threads + core_threads);
blocking::create_blocking_pool(self, self.max_blocking_threads + worker_threads);
let blocking_spawner = blocking_pool.spawner().clone();
// Generate a rng seed for this runtime.
@@ -1600,7 +1707,7 @@ cfg_rt_multi_thread! {
let seed_generator_2 = self.seed_generator.next_generator();
let (scheduler, handle) = MultiThreadAlt::new(
core_threads,
worker_threads,
driver,
driver_handle,
blocking_spawner,
@@ -1610,6 +1717,10 @@ cfg_rt_multi_thread! {
after_unpark: self.after_unpark.clone(),
before_spawn: self.before_spawn.clone(),
after_termination: self.after_termination.clone(),
#[cfg(tokio_unstable)]
before_poll: self.before_poll.clone(),
#[cfg(tokio_unstable)]
after_poll: self.after_poll.clone(),
global_queue_interval: self.global_queue_interval,
event_interval: self.event_interval,
local_queue_capacity: self.local_queue_capacity,
+8
View File
@@ -27,6 +27,14 @@ pub(crate) struct Config {
/// To run after each task is terminated.
pub(crate) after_termination: Option<TaskCallback>,
/// To run before each poll
#[cfg(tokio_unstable)]
pub(crate) before_poll: Option<TaskCallback>,
/// To run after each poll
#[cfg(tokio_unstable)]
pub(crate) after_poll: Option<TaskCallback>,
/// The multi-threaded scheduler includes a per-worker LIFO slot used to
/// store the last scheduled task. This can improve certain usage patterns,
/// especially message passing between tasks. However, this LIFO slot is not
+2 -2
View File
@@ -1,5 +1,5 @@
use crate::loom::thread::AccessError;
use crate::runtime::coop;
use crate::task::coop;
use std::cell::Cell;
@@ -135,7 +135,7 @@ pub(crate) fn thread_rng_n(n: u32) -> u32 {
})
}
pub(super) fn budget<R>(f: impl FnOnce(&Cell<coop::Budget>) -> R) -> Result<R, AccessError> {
pub(crate) fn budget<R>(f: impl FnOnce(&Cell<coop::Budget>) -> R) -> Result<R, AccessError> {
CONTEXT.try_with(|ctx| f(&ctx.budget))
}
+1 -1
View File
@@ -87,7 +87,7 @@ impl BlockingRegionGuard {
let when = Instant::now() + timeout;
loop {
if let Ready(v) = crate::runtime::coop::budget(|| f.as_mut().poll(&mut cx)) {
if let Ready(v) = crate::task::coop::budget(|| f.as_mut().poll(&mut cx)) {
return Ok(v);
}
+69 -1
View File
@@ -3,7 +3,9 @@
//! See [`Handle::dump`][crate::runtime::Handle::dump].
use crate::task::Id;
use std::{fmt, path::Path};
use std::{fmt, future::Future, path::Path};
pub use crate::runtime::task::trace::Root;
/// A snapshot of a runtime's state.
///
@@ -214,6 +216,72 @@ impl Trace {
})
.collect()
}
/// Runs the function `f` in tracing mode, and returns its result along with the resulting [`Trace`].
///
/// This is normally called with `f` being the poll function of a future, and will give you a backtrace
/// that tells you what that one future is doing.
///
/// Use [`Handle::dump`] instead if you want to know what *all the tasks* in your program are doing.
/// Also see [`Handle::dump`] for more documentation about dumps, but unlike [`Handle::dump`], this function
/// should not be much slower than calling `f` directly.
///
/// Due to the way tracing is implemented, Tokio leaf futures will usually, instead of doing their
/// actual work, do the equivalent of a `yield_now` (returning a `Poll::Pending` and scheduling the
/// current context for execution), which means forward progress will probably not happen unless
/// you eventually call your future outside of `capture`.
///
/// [`Handle::dump`]: crate::runtime::Handle::dump
///
/// Example usage:
/// ```
/// use std::future::Future;
/// use std::task::Poll;
/// use tokio::runtime::dump::Trace;
///
/// # async fn test_fn() {
/// // some future
/// let mut test_future = std::pin::pin!(async move { tokio::task::yield_now().await; 0 });
///
/// // trace it once, see what it's doing
/// let (trace, res) = Trace::root(std::future::poll_fn(|cx| {
/// let (res, trace) = Trace::capture(|| test_future.as_mut().poll(cx));
/// Poll::Ready((trace, res))
/// })).await;
///
/// // await it to let it finish, outside of a `capture`
/// let output = match res {
/// Poll::Ready(output) => output,
/// Poll::Pending => test_future.await,
/// };
///
/// println!("{trace}");
/// # }
/// ```
///
/// ### Nested calls
///
/// Nested calls to `capture` might return partial traces, but will not do any other undesirable behavior (for
/// example, they will not panic).
pub fn capture<F, R>(f: F) -> (R, Trace)
where
F: FnOnce() -> R,
{
let (res, trace) = super::task::trace::Trace::capture(f);
(res, Trace { inner: trace })
}
/// Create a root for stack traces captured using [`Trace::capture`]. Stack frames above
/// the root will not be captured.
///
/// Nesting multiple [`Root`] futures is fine. Captures will stop at the first root. Not having
/// a [`Root`] is fine as well, but there is no guarantee on where the capture will stop.
pub fn root<F>(f: F) -> Root<F>
where
F: Future,
{
crate::runtime::task::trace::Trace::root(f)
}
}
impl Dump {
+3
View File
@@ -447,6 +447,9 @@ cfg_taskdump! {
impl Handle {
/// Captures a snapshot of the runtime's state.
///
/// If you only want to capture a snapshot of a single future's state, you can use
/// [`Trace::capture`][crate::runtime::dump::Trace].
///
/// This functionality is experimental, and comes with a number of
/// requirements and limitations.
///
+2 -2
View File
@@ -148,7 +148,7 @@ impl Registration {
) -> Poll<io::Result<ReadyEvent>> {
ready!(crate::trace::trace_leaf(cx));
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let ev = ready!(self.shared.poll_readiness(cx, direction));
if ev.is_shutdown {
@@ -219,7 +219,7 @@ impl Registration {
loop {
let event = self.readiness(interest).await?;
let coop = std::future::poll_fn(crate::runtime::coop::poll_proceed).await;
let coop = std::future::poll_fn(crate::task::coop::poll_proceed).await;
match f() {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+4 -4
View File
@@ -7,6 +7,9 @@ use std::ptr::NonNull;
use std::sync::atomic::Ordering::{Acquire, Release};
use std::sync::Arc;
// Kind of arbitrary, but buffering 16 `ScheduledIo`s doesn't seem like much
const NOTIFY_AFTER: usize = 16;
pub(super) struct RegistrationSet {
num_pending_release: AtomicUsize,
}
@@ -35,7 +38,7 @@ impl RegistrationSet {
let synced = Synced {
is_shutdown: false,
registrations: LinkedList::new(),
pending_release: Vec::with_capacity(16),
pending_release: Vec::with_capacity(NOTIFY_AFTER),
};
(set, synced)
@@ -69,9 +72,6 @@ impl RegistrationSet {
// Returns `true` if the caller should unblock the I/O driver to purge
// registrations pending release.
pub(super) fn deregister(&self, synced: &mut Synced, registration: &Arc<ScheduledIo>) -> bool {
// Kind of arbitrary, but buffering 16 `ScheduledIo`s doesn't seem like much
const NOTIFY_AFTER: usize = 16;
synced.pending_release.push(registration.clone());
let len = synced.pending_release.len();
+1 -3
View File
@@ -310,7 +310,7 @@
//! [`event_interval`]: crate::runtime::Builder::event_interval
//! [`disable_lifo_slot`]: crate::runtime::Builder::disable_lifo_slot
//! [the lifo slot optimization]: crate::runtime::Builder::disable_lifo_slot
//! [coop budget]: crate::task#cooperative-scheduling
//! [coop budget]: crate::task::coop#cooperative-scheduling
//! [`worker_mean_poll_time`]: crate::runtime::RuntimeMetrics::worker_mean_poll_time
// At the top due to macros
@@ -321,8 +321,6 @@ mod tests;
pub(crate) mod context;
pub(crate) mod coop;
pub(crate) mod park;
mod driver;
+1 -1
View File
@@ -281,7 +281,7 @@ impl CachedParkThread {
pin!(f);
loop {
if let Ready(v) = crate::runtime::coop::budget(|| f.as_mut().poll(&mut cx)) {
if let Ready(v) = crate::task::coop::budget(|| f.as_mut().poll(&mut cx)) {
return Ok(v);
}
@@ -145,6 +145,10 @@ impl CurrentThread {
task_hooks: TaskHooks {
task_spawn_callback: config.before_spawn.clone(),
task_terminate_callback: config.after_termination.clone(),
#[cfg(tokio_unstable)]
before_poll_callback: config.before_poll.clone(),
#[cfg(tokio_unstable)]
after_poll_callback: config.after_poll.clone(),
},
shared: Shared {
inject: Inject::new(),
@@ -361,7 +365,7 @@ impl Context {
/// thread-local context.
fn run_task<R>(&self, mut core: Box<Core>, f: impl FnOnce() -> R) -> (Box<Core>, R) {
core.metrics.start_poll();
let mut ret = self.enter(core, || crate::runtime::coop::budget(f));
let mut ret = self.enter(core, || crate::task::coop::budget(f));
ret.0.metrics.end_poll();
ret
}
@@ -726,7 +730,7 @@ impl CoreGuard<'_> {
if handle.reset_woken() {
let (c, res) = context.enter(core, || {
crate::runtime::coop::budget(|| future.as_mut().poll(&mut cx))
crate::task::coop::budget(|| future.as_mut().poll(&mut cx))
});
core = c;
@@ -766,8 +770,17 @@ impl CoreGuard<'_> {
let task = context.handle.shared.owned.assert_owner(task);
#[cfg(tokio_unstable)]
let task_id = task.task_id();
let (c, ()) = context.run_task(core, || {
#[cfg(tokio_unstable)]
context.handle.task_hooks.poll_start_callback(task_id);
task.run();
#[cfg(tokio_unstable)]
context.handle.task_hooks.poll_stop_callback(task_id);
});
core = c;
@@ -63,10 +63,9 @@ use crate::runtime::scheduler::multi_thread::{
};
use crate::runtime::scheduler::{inject, Defer, Lock};
use crate::runtime::task::{OwnedTasks, TaskHarnessScheduleHooks};
use crate::runtime::{
blocking, coop, driver, scheduler, task, Config, SchedulerMetrics, WorkerMetrics,
};
use crate::runtime::{blocking, driver, scheduler, task, Config, SchedulerMetrics, WorkerMetrics};
use crate::runtime::{context, TaskHooks};
use crate::task::coop;
use crate::util::atomic_cell::AtomicCell;
use crate::util::rand::{FastRand, RngSeedGenerator};
@@ -282,10 +281,7 @@ pub(super) fn create(
let remotes_len = remotes.len();
let handle = Arc::new(Handle {
task_hooks: TaskHooks {
task_spawn_callback: config.before_spawn.clone(),
task_terminate_callback: config.after_termination.clone(),
},
task_hooks: TaskHooks::from_config(&config),
shared: Shared {
remotes: remotes.into_boxed_slice(),
inject,
@@ -574,6 +570,9 @@ impl Context {
}
fn run_task(&self, task: Notified, mut core: Box<Core>) -> RunResult {
#[cfg(tokio_unstable)]
let task_id = task.task_id();
let task = self.worker.handle.shared.owned.assert_owner(task);
// Make sure the worker is not in the **searching** state. This enables
@@ -593,7 +592,16 @@ impl Context {
// Run the task
coop::budget(|| {
// Unlike the poll time above, poll start callback is attached to the task id,
// so it is tightly associated with the actual poll invocation.
#[cfg(tokio_unstable)]
self.worker.handle.task_hooks.poll_start_callback(task_id);
task.run();
#[cfg(tokio_unstable)]
self.worker.handle.task_hooks.poll_stop_callback(task_id);
let mut lifo_polls = 0;
// As long as there is budget remaining and a task exists in the
@@ -656,7 +664,17 @@ impl Context {
// Run the LIFO task, then loop
*self.core.borrow_mut() = Some(core);
let task = self.worker.handle.shared.owned.assert_owner(task);
#[cfg(tokio_unstable)]
let task_id = task.task_id();
#[cfg(tokio_unstable)]
self.worker.handle.task_hooks.poll_start_callback(task_id);
task.run();
#[cfg(tokio_unstable)]
self.worker.handle.task_hooks.poll_stop_callback(task_id);
}
})
}
@@ -764,7 +782,13 @@ impl Context {
}
pub(crate) fn defer(&self, waker: &Waker) {
self.defer.defer(waker);
if self.core.borrow().is_none() {
// If there is no core, then the worker is currently in a block_in_place. In this case,
// we cannot use the defer queue as we aren't really in the current runtime.
waker.wake_by_ref();
} else {
self.defer.defer(waker);
}
}
#[allow(dead_code)]
@@ -64,8 +64,9 @@ use crate::runtime::scheduler::multi_thread_alt::{
};
use crate::runtime::scheduler::{self, inject, Lock};
use crate::runtime::task::{OwnedTasks, TaskHarnessScheduleHooks};
use crate::runtime::{blocking, coop, driver, task, Config, SchedulerMetrics, WorkerMetrics};
use crate::runtime::{blocking, driver, task, Config, SchedulerMetrics, WorkerMetrics};
use crate::runtime::{context, TaskHooks};
use crate::task::coop;
use crate::util::atomic_cell::AtomicCell;
use crate::util::rand::{FastRand, RngSeedGenerator};
@@ -303,10 +304,7 @@ pub(super) fn create(
let (inject, inject_synced) = inject::Shared::new();
let handle = Arc::new(Handle {
task_hooks: TaskHooks {
task_spawn_callback: config.before_spawn.clone(),
task_terminate_callback: config.after_termination.clone(),
},
task_hooks: TaskHooks::from_config(&config),
shared: Shared {
remotes: remotes.into_boxed_slice(),
inject,
+1 -1
View File
@@ -322,7 +322,7 @@ impl<T> Future for JoinHandle<T> {
let mut ret = Poll::Pending;
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
// Try to read the task output. If the task is not yet complete, the
// waker is stored and is notified once the task does complete.
+23 -9
View File
@@ -256,6 +256,13 @@ pub(crate) struct LocalNotified<S: 'static> {
_not_send: PhantomData<*const ()>,
}
impl<S> LocalNotified<S> {
#[cfg(tokio_unstable)]
pub(crate) fn task_id(&self) -> Id {
self.task.id()
}
}
/// A task that is not owned by any `OwnedTasks`. Used for blocking tasks.
/// This type holds two ref-counts.
pub(crate) struct UnownedTask<S: 'static> {
@@ -386,6 +393,16 @@ impl<S: 'static> Task<S> {
self.raw.header_ptr()
}
/// Returns a [task ID] that uniquely identifies this task relative to other
/// currently spawned tasks.
///
/// [task ID]: crate::task::Id
#[cfg(tokio_unstable)]
pub(crate) fn id(&self) -> crate::task::Id {
// Safety: The header pointer is valid.
unsafe { Header::get_id(self.raw.header_ptr()) }
}
cfg_taskdump! {
/// Notify the task for task dumping.
///
@@ -400,15 +417,6 @@ impl<S: 'static> Task<S> {
}
}
/// Returns a [task ID] that uniquely identifies this task relative to other
/// currently spawned tasks.
///
/// [task ID]: crate::task::Id
#[cfg(tokio_unstable)]
pub(crate) fn id(&self) -> crate::task::Id {
// Safety: The header pointer is valid.
unsafe { Header::get_id(self.raw.header_ptr()) }
}
}
}
@@ -416,6 +424,12 @@ impl<S: 'static> Notified<S> {
fn header(&self) -> &Header {
self.0.header()
}
#[cfg(tokio_unstable)]
#[allow(dead_code)]
pub(crate) fn task_id(&self) -> crate::task::Id {
self.0.id()
}
}
impl<S: 'static> Notified<S> {
+2 -1
View File
@@ -56,7 +56,8 @@ pub(crate) struct Trace {
pin_project_lite::pin_project! {
#[derive(Debug, Clone)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub(crate) struct Root<T> {
/// A future wrapper that roots traces (captured with [`Trace::capture`]).
pub struct Root<T> {
#[pin]
future: T,
}
+40
View File
@@ -1,17 +1,57 @@
use std::marker::PhantomData;
use super::Config;
impl TaskHooks {
pub(crate) fn spawn(&self, meta: &TaskMeta<'_>) {
if let Some(f) = self.task_spawn_callback.as_ref() {
f(meta)
}
}
#[allow(dead_code)]
pub(crate) fn from_config(config: &Config) -> Self {
Self {
task_spawn_callback: config.before_spawn.clone(),
task_terminate_callback: config.after_termination.clone(),
#[cfg(tokio_unstable)]
before_poll_callback: config.before_poll.clone(),
#[cfg(tokio_unstable)]
after_poll_callback: config.after_poll.clone(),
}
}
#[cfg(tokio_unstable)]
#[inline]
pub(crate) fn poll_start_callback(&self, id: super::task::Id) {
if let Some(poll_start) = &self.before_poll_callback {
(poll_start)(&TaskMeta {
id,
_phantom: std::marker::PhantomData,
})
}
}
#[cfg(tokio_unstable)]
#[inline]
pub(crate) fn poll_stop_callback(&self, id: super::task::Id) {
if let Some(poll_stop) = &self.after_poll_callback {
(poll_stop)(&TaskMeta {
id,
_phantom: std::marker::PhantomData,
})
}
}
}
#[derive(Clone)]
pub(crate) struct TaskHooks {
pub(crate) task_spawn_callback: Option<TaskCallback>,
pub(crate) task_terminate_callback: Option<TaskCallback>,
#[cfg(tokio_unstable)]
pub(crate) before_poll_callback: Option<TaskCallback>,
#[cfg(tokio_unstable)]
pub(crate) after_poll_callback: Option<TaskCallback>,
}
/// Task metadata supplied to user-provided hooks for task events.
+39 -6
View File
@@ -33,6 +33,21 @@ fn new(signum: u32) -> io::Result<RxFuture> {
Ok(RxFuture::new(rx))
}
fn event_requires_infinite_sleep_in_handler(signum: u32) -> bool {
// Returning from the handler function of those events immediately terminates the process.
// So for async systems, the easiest solution is to simply never return from
// the handler function.
//
// For more information, see:
// https://learn.microsoft.com/en-us/windows/console/handlerroutine#remarks
match signum {
console::CTRL_CLOSE_EVENT => true,
console::CTRL_LOGOFF_EVENT => true,
console::CTRL_SHUTDOWN_EVENT => true,
_ => false,
}
}
#[derive(Debug)]
pub(crate) struct OsStorage {
ctrl_break: EventInfo,
@@ -114,7 +129,15 @@ unsafe extern "system" fn handler(ty: u32) -> BOOL {
// the handler routine is always invoked in a new thread, thus we don't
// have the same restrictions as in Unix signal handlers, meaning we can
// go ahead and perform the broadcast here.
if globals.broadcast() {
let event_was_handled = globals.broadcast();
if event_was_handled && event_requires_infinite_sleep_in_handler(ty) {
loop {
std::thread::park();
}
}
if event_was_handled {
1
} else {
// No one is listening for this notification any more
@@ -130,6 +153,16 @@ mod tests {
use tokio_test::{assert_ok, assert_pending, assert_ready_ok, task};
unsafe fn raise_event(signum: u32) {
if event_requires_infinite_sleep_in_handler(signum) {
// Those events will enter an infinite loop in `handler`, so
// we need to run them on a separate thread
std::thread::spawn(move || super::handler(signum));
} else {
super::handler(signum);
}
}
#[test]
fn ctrl_c() {
let rt = rt();
@@ -143,7 +176,7 @@ mod tests {
// like sending signals on Unix, so we'll stub out the actual OS
// integration and test that our handling works.
unsafe {
super::handler(console::CTRL_C_EVENT);
raise_event(console::CTRL_C_EVENT);
}
assert_ready_ok!(ctrl_c.poll());
@@ -160,7 +193,7 @@ mod tests {
// like sending signals on Unix, so we'll stub out the actual OS
// integration and test that our handling works.
unsafe {
super::handler(console::CTRL_BREAK_EVENT);
raise_event(console::CTRL_BREAK_EVENT);
}
ctrl_break.recv().await.unwrap();
@@ -178,7 +211,7 @@ mod tests {
// like sending signals on Unix, so we'll stub out the actual OS
// integration and test that our handling works.
unsafe {
super::handler(console::CTRL_CLOSE_EVENT);
raise_event(console::CTRL_CLOSE_EVENT);
}
ctrl_close.recv().await.unwrap();
@@ -196,7 +229,7 @@ mod tests {
// like sending signals on Unix, so we'll stub out the actual OS
// integration and test that our handling works.
unsafe {
super::handler(console::CTRL_SHUTDOWN_EVENT);
raise_event(console::CTRL_SHUTDOWN_EVENT);
}
ctrl_shutdown.recv().await.unwrap();
@@ -214,7 +247,7 @@ mod tests {
// like sending signals on Unix, so we'll stub out the actual OS
// integration and test that our handling works.
unsafe {
super::handler(console::CTRL_LOGOFF_EVENT);
raise_event(console::CTRL_LOGOFF_EVENT);
}
ctrl_logoff.recv().await.unwrap();
+2 -2
View File
@@ -591,11 +591,11 @@ impl Future for Acquire<'_> {
#[cfg(all(tokio_unstable, feature = "tracing"))]
let coop = ready!(trace_poll_op!(
"poll_acquire",
crate::runtime::coop::poll_proceed(cx),
crate::task::coop::poll_proceed(cx),
));
#[cfg(not(all(tokio_unstable, feature = "tracing")))]
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let result = match semaphore.poll_acquire(cx, needed, node, *queued) {
Poll::Pending => {
+222 -5
View File
@@ -119,7 +119,7 @@
use crate::loom::cell::UnsafeCell;
use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
use crate::loom::sync::{Arc, Mutex, MutexGuard};
use crate::runtime::coop::cooperative;
use crate::task::coop::cooperative;
use crate::util::linked_list::{self, GuardedLinkedList, LinkedList};
use crate::util::WakeList;
@@ -128,7 +128,7 @@ use std::future::Future;
use std::marker::PhantomPinned;
use std::pin::Pin;
use std::ptr::NonNull;
use std::sync::atomic::Ordering::{Acquire, Relaxed, Release, SeqCst};
use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release, SeqCst};
use std::task::{ready, Context, Poll, Waker};
/// Sending-half of the [`broadcast`] channel.
@@ -166,6 +166,40 @@ pub struct Sender<T> {
shared: Arc<Shared<T>>,
}
/// A sender that does not prevent the channel from being closed.
///
/// If all [`Sender`] instances of a channel were dropped and only `WeakSender`
/// instances remain, the channel is closed.
///
/// In order to send messages, the `WeakSender` needs to be upgraded using
/// [`WeakSender::upgrade`], which returns `Option<Sender>`. It returns `None`
/// if all `Sender`s have been dropped, and otherwise it returns a `Sender`.
///
/// [`Sender`]: Sender
/// [`WeakSender::upgrade`]: WeakSender::upgrade
///
/// # Examples
///
/// ```
/// use tokio::sync::broadcast::channel;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, _rx) = channel::<i32>(15);
/// let tx_weak = tx.downgrade();
///
/// // Upgrading will succeed because `tx` still exists.
/// assert!(tx_weak.upgrade().is_some());
///
/// // If we drop `tx`, then it will fail.
/// drop(tx);
/// assert!(tx_weak.clone().upgrade().is_none());
/// }
/// ```
pub struct WeakSender<T> {
shared: Arc<Shared<T>>,
}
/// Receiving-half of the [`broadcast`] channel.
///
/// Must not be used concurrently. Messages may be retrieved using
@@ -301,6 +335,8 @@ pub mod error {
use self::error::{RecvError, SendError, TryRecvError};
use super::Notify;
/// Data shared between senders and receivers.
struct Shared<T> {
/// slots in the channel.
@@ -314,6 +350,12 @@ struct Shared<T> {
/// Number of outstanding Sender handles.
num_tx: AtomicUsize,
/// Number of outstanding weak Sender handles.
num_weak_tx: AtomicUsize,
/// Notify when the last subscribed [`Receiver`] drops.
notify_last_rx_drop: Notify,
}
/// Next position to write a value.
@@ -526,6 +568,8 @@ impl<T> Sender<T> {
waiters: LinkedList::new(),
}),
num_tx: AtomicUsize::new(1),
num_weak_tx: AtomicUsize::new(0),
notify_last_rx_drop: Notify::new(),
});
Sender { shared }
@@ -648,6 +692,18 @@ impl<T> Sender<T> {
new_receiver(shared)
}
/// Converts the `Sender` to a [`WeakSender`] that does not count
/// towards RAII semantics, i.e. if all `Sender` instances of the
/// channel were dropped and only `WeakSender` instances remain,
/// the channel is closed.
#[must_use = "Downgrade creates a WeakSender without destroying the original non-weak sender."]
pub fn downgrade(&self) -> WeakSender<T> {
self.shared.num_weak_tx.fetch_add(1, Relaxed);
WeakSender {
shared: self.shared.clone(),
}
}
/// Returns the number of queued values.
///
/// A value is queued until it has either been seen by all receivers that were alive at the time
@@ -803,12 +859,63 @@ impl<T> Sender<T> {
Arc::ptr_eq(&self.shared, &other.shared)
}
/// A future which completes when the number of [Receiver]s subscribed to this `Sender` reaches
/// zero.
///
/// # Examples
///
/// ```
/// use futures::FutureExt;
/// use tokio::sync::broadcast;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx1) = broadcast::channel::<u32>(16);
/// let mut rx2 = tx.subscribe();
///
/// let _ = tx.send(10);
///
/// assert_eq!(rx1.recv().await.unwrap(), 10);
/// drop(rx1);
/// assert!(tx.closed().now_or_never().is_none());
///
/// assert_eq!(rx2.recv().await.unwrap(), 10);
/// drop(rx2);
/// assert!(tx.closed().now_or_never().is_some());
/// }
/// ```
pub async fn closed(&self) {
loop {
let notified = self.shared.notify_last_rx_drop.notified();
{
// Ensure the lock drops if the channel isn't closed
let tail = self.shared.tail.lock();
if tail.closed {
return;
}
}
notified.await;
}
}
fn close_channel(&self) {
let mut tail = self.shared.tail.lock();
tail.closed = true;
self.shared.notify_rx(tail);
}
/// Returns the number of [`Sender`] handles.
pub fn strong_count(&self) -> usize {
self.shared.num_tx.load(Acquire)
}
/// Returns the number of [`WeakSender`] handles.
pub fn weak_count(&self) -> usize {
self.shared.num_weak_tx.load(Acquire)
}
}
/// Create a new `Receiver` which reads starting from the tail.
@@ -817,8 +924,14 @@ fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");
tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
if tail.rx_cnt == 0 {
// Potentially need to re-open the channel, if a new receiver has been added between calls
// to poll(). Note that we use rx_cnt == 0 instead of is_closed since is_closed also
// applies if the sender has been dropped
tail.closed = false;
}
tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
let next = tail.pos;
drop(tail);
@@ -943,7 +1056,7 @@ impl<T> Shared<T> {
impl<T> Clone for Sender<T> {
fn clone(&self) -> Sender<T> {
let shared = self.shared.clone();
shared.num_tx.fetch_add(1, SeqCst);
shared.num_tx.fetch_add(1, Relaxed);
Sender { shared }
}
@@ -951,12 +1064,68 @@ impl<T> Clone for Sender<T> {
impl<T> Drop for Sender<T> {
fn drop(&mut self) {
if 1 == self.shared.num_tx.fetch_sub(1, SeqCst) {
if 1 == self.shared.num_tx.fetch_sub(1, AcqRel) {
self.close_channel();
}
}
}
impl<T> WeakSender<T> {
/// Tries to convert a `WeakSender` into a [`Sender`].
///
/// This will return `Some` if there are other `Sender` instances alive and
/// the channel wasn't previously dropped, otherwise `None` is returned.
#[must_use]
pub fn upgrade(&self) -> Option<Sender<T>> {
let mut tx_count = self.shared.num_tx.load(Acquire);
loop {
if tx_count == 0 {
// channel is closed so this WeakSender can not be upgraded
return None;
}
match self
.shared
.num_tx
.compare_exchange_weak(tx_count, tx_count + 1, Relaxed, Acquire)
{
Ok(_) => {
return Some(Sender {
shared: self.shared.clone(),
})
}
Err(prev_count) => tx_count = prev_count,
}
}
}
/// Returns the number of [`Sender`] handles.
pub fn strong_count(&self) -> usize {
self.shared.num_tx.load(Acquire)
}
/// Returns the number of [`WeakSender`] handles.
pub fn weak_count(&self) -> usize {
self.shared.num_weak_tx.load(Acquire)
}
}
impl<T> Clone for WeakSender<T> {
fn clone(&self) -> WeakSender<T> {
let shared = self.shared.clone();
shared.num_weak_tx.fetch_add(1, Relaxed);
Self { shared }
}
}
impl<T> Drop for WeakSender<T> {
fn drop(&mut self) {
self.shared.num_weak_tx.fetch_sub(1, AcqRel);
}
}
impl<T> Receiver<T> {
/// Returns the number of messages that were sent into the channel and that
/// this [`Receiver`] has yet to receive.
@@ -1158,6 +1327,42 @@ impl<T> Receiver<T> {
Ok(RecvGuard { slot })
}
/// Returns the number of [`Sender`] handles.
pub fn sender_strong_count(&self) -> usize {
self.shared.num_tx.load(Acquire)
}
/// Returns the number of [`WeakSender`] handles.
pub fn sender_weak_count(&self) -> usize {
self.shared.num_weak_tx.load(Acquire)
}
/// Checks if a channel is closed.
///
/// This method returns `true` if the channel has been closed. The channel is closed
/// when all [`Sender`] have been dropped.
///
/// [`Sender`]: crate::sync::broadcast::Sender
///
/// # Examples
/// ```
/// use tokio::sync::broadcast;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, rx) = broadcast::channel::<()>(10);
/// assert!(!rx.is_closed());
///
/// drop(tx);
///
/// assert!(rx.is_closed());
/// }
/// ```
pub fn is_closed(&self) -> bool {
// Channel is closed when there are no strong senders left active
self.shared.num_tx.load(Acquire) == 0
}
}
impl<T: Clone> Receiver<T> {
@@ -1344,6 +1549,12 @@ impl<T> Drop for Receiver<T> {
tail.rx_cnt -= 1;
let until = tail.pos;
let remaining_rx = tail.rx_cnt;
if remaining_rx == 0 {
self.shared.notify_last_rx_drop.notify_waiters();
tail.closed = true;
}
drop(tail);
@@ -1475,6 +1686,12 @@ impl<T> fmt::Debug for Sender<T> {
}
}
impl<T> fmt::Debug for WeakSender<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "broadcast::WeakSender")
}
}
impl<T> fmt::Debug for Receiver<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "broadcast::Receiver")
+1 -1
View File
@@ -439,7 +439,7 @@
//! or even use them from non-Tokio runtimes.
//!
//! When used in a Tokio runtime, the synchronization primitives participate in
//! [cooperative scheduling](crate::task#cooperative-scheduling) to avoid
//! [cooperative scheduling](crate::task::coop#cooperative-scheduling) to avoid
//! starvation. This feature does not apply when used from non-Tokio runtimes.
//!
//! As an exception, methods ending in `_timeout` are not runtime agnostic
+29 -5
View File
@@ -292,7 +292,7 @@ impl<T, S: Semaphore> Rx<T, S> {
ready!(crate::trace::trace_leaf(cx));
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
self.inner.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
@@ -354,7 +354,7 @@ impl<T, S: Semaphore> Rx<T, S> {
ready!(crate::trace::trace_leaf(cx));
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
if limit == 0 {
coop.made_progress();
@@ -490,10 +490,34 @@ impl<T, S: Semaphore> Drop for Rx<T, S> {
self.inner.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
while let Some(Value(_)) = rx_fields.list.pop(&self.inner.tx) {
self.inner.semaphore.add_permit();
struct Guard<'a, T, S: Semaphore> {
list: &'a mut list::Rx<T>,
tx: &'a list::Tx<T>,
sem: &'a S,
}
impl<'a, T, S: Semaphore> Guard<'a, T, S> {
fn drain(&mut self) {
// call T's destructor.
while let Some(Value(_)) = self.list.pop(self.tx) {
self.sem.add_permit();
}
}
}
impl<'a, T, S: Semaphore> Drop for Guard<'a, T, S> {
fn drop(&mut self) {
self.drain();
}
}
let mut guard = Guard {
list: &mut rx_fields.list,
tx: &self.inner.tx,
sem: &self.inner.semaphore,
};
guard.drain();
});
}
}
+1 -1
View File
@@ -75,7 +75,7 @@
//! runtimes.
//!
//! When used in a Tokio runtime, it participates in
//! [cooperative scheduling](crate::task#cooperative-scheduling) to avoid
//! [cooperative scheduling](crate::task::coop#cooperative-scheduling) to avoid
//! starvation. This feature does not apply when used from non-Tokio runtimes.
//!
//! As an exception, methods ending in `_timeout` are not runtime agnostic
+161 -5
View File
@@ -794,7 +794,7 @@ impl<T> Sender<T> {
ready!(crate::trace::trace_leaf(cx));
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let inner = self.inner.as_ref().unwrap();
@@ -931,6 +931,149 @@ impl<T> Receiver<T> {
}
}
/// Checks if this receiver is terminated.
///
/// This function returns true if this receiver has already yielded a [`Poll::Ready`] result.
/// If so, this receiver should no longer be polled.
///
/// # Examples
///
/// Sending a value and polling it.
///
/// ```
/// use tokio::sync::oneshot;
///
/// use std::task::Poll;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = oneshot::channel();
///
/// // A receiver is not terminated when it is initialized.
/// assert!(!rx.is_terminated());
///
/// // A receiver is not terminated it is polled and is still pending.
/// let poll = futures::poll!(&mut rx);
/// assert_eq!(poll, Poll::Pending);
/// assert!(!rx.is_terminated());
///
/// // A receiver is not terminated if a value has been sent, but not yet read.
/// tx.send(0).unwrap();
/// assert!(!rx.is_terminated());
///
/// // A receiver *is* terminated after it has been polled and yielded a value.
/// assert_eq!((&mut rx).await, Ok(0));
/// assert!(rx.is_terminated());
/// }
/// ```
///
/// Dropping the sender.
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = oneshot::channel::<()>();
///
/// // A receiver is not immediately terminated when the sender is dropped.
/// drop(tx);
/// assert!(!rx.is_terminated());
///
/// // A receiver *is* terminated after it has been polled and yielded an error.
/// let _ = (&mut rx).await.unwrap_err();
/// assert!(rx.is_terminated());
/// }
/// ```
pub fn is_terminated(&self) -> bool {
self.inner.is_none()
}
/// Checks if a channel is empty.
///
/// This method returns `true` if the channel has no messages.
///
/// It is not necessarily safe to poll an empty receiver, which may have
/// already yielded a value. Use [`is_terminated()`][Self::is_terminated]
/// to check whether or not a receiver can be safely polled, instead.
///
/// # Examples
///
/// Sending a value.
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = oneshot::channel();
/// assert!(rx.is_empty());
///
/// tx.send(0).unwrap();
/// assert!(!rx.is_empty());
///
/// let _ = (&mut rx).await;
/// assert!(rx.is_empty());
/// }
/// ```
///
/// Dropping the sender.
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = oneshot::channel::<()>();
///
/// // A channel is empty if the sender is dropped.
/// drop(tx);
/// assert!(rx.is_empty());
///
/// // A closed channel still yields an error, however.
/// (&mut rx).await.expect_err("should yield an error");
/// assert!(rx.is_empty());
/// }
/// ```
///
/// Terminated channels are empty.
///
/// ```should_panic
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = oneshot::channel();
/// tx.send(0).unwrap();
/// let _ = (&mut rx).await;
///
/// // NB: an empty channel is not necessarily safe to poll!
/// assert!(rx.is_empty());
/// let _ = (&mut rx).await;
/// }
/// ```
pub fn is_empty(&self) -> bool {
let Some(inner) = self.inner.as_ref() else {
// The channel has already terminated.
return true;
};
let state = State::load(&inner.state, Acquire);
if state.is_complete() {
// SAFETY: If `state.is_complete()` returns true, then the
// `VALUE_SENT` bit has been set and the sender side of the
// channel will no longer attempt to access the inner
// `UnsafeCell`. Therefore, it is now safe for us to access the
// cell.
//
// The channel is empty if it does not have a value.
unsafe { !inner.has_value() }
} else {
// The receiver closed the channel or no value has been sent yet.
true
}
}
/// Attempts to receive a value.
///
/// If a pending value exists in the channel, it is returned. If no value
@@ -1106,10 +1249,10 @@ impl<T> Future for Receiver<T> {
let ret = if let Some(inner) = self.as_ref().get_ref().inner.as_ref() {
#[cfg(all(tokio_unstable, feature = "tracing"))]
let res = ready!(trace_poll_op!("poll_recv", inner.poll_recv(cx)))?;
let res = ready!(trace_poll_op!("poll_recv", inner.poll_recv(cx))).map_err(Into::into);
#[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
let res = ready!(inner.poll_recv(cx))?;
let res = ready!(inner.poll_recv(cx)).map_err(Into::into);
res
} else {
@@ -1117,7 +1260,7 @@ impl<T> Future for Receiver<T> {
};
self.inner = None;
Ready(Ok(ret))
Ready(ret)
}
}
@@ -1142,7 +1285,7 @@ impl<T> Inner<T> {
fn poll_recv(&self, cx: &mut Context<'_>) -> Poll<Result<T, RecvError>> {
ready!(crate::trace::trace_leaf(cx));
// Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
// Load the state
let mut state = State::load(&self.state, Acquire);
@@ -1233,6 +1376,19 @@ impl<T> Inner<T> {
unsafe fn consume_value(&self) -> Option<T> {
self.value.with_mut(|ptr| (*ptr).take())
}
/// Returns true if there is a value. This function does not check `state`.
///
/// # Safety
///
/// Calling this method concurrently on multiple threads will result in a
/// data race. The `VALUE_SENT` state bit is used to ensure that only the
/// sender *or* the receiver will call this method at a given point in time.
/// If `VALUE_SENT` is not set, then only the sender may call this method;
/// if it is set, then only the receiver may call this method.
unsafe fn has_value(&self) -> bool {
self.value.with(|ptr| (*ptr).is_some())
}
}
unsafe impl<T: Send> Send for Inner<T> {}
+1 -1
View File
@@ -111,8 +111,8 @@
//! [`Sender::closed`]: crate::sync::watch::Sender::closed
//! [`Sender::subscribe()`]: crate::sync::watch::Sender::subscribe
use crate::runtime::coop::cooperative;
use crate::sync::notify::Notify;
use crate::task::coop::cooperative;
use crate::loom::sync::atomic::AtomicUsize;
use crate::loom::sync::atomic::Ordering::{AcqRel, Relaxed};
@@ -1,5 +1,3 @@
use std::task::{ready, Poll};
/// Consumes a unit of budget and returns the execution back to the Tokio
/// runtime *if* the task's coop budget was exhausted.
///
@@ -25,14 +23,14 @@ use std::task::{ready, Poll};
/// ```
#[cfg_attr(docsrs, doc(cfg(feature = "rt")))]
pub async fn consume_budget() {
let mut status = Poll::Pending;
let mut status = std::task::Poll::Pending;
std::future::poll_fn(move |cx| {
ready!(crate::trace::trace_leaf(cx));
std::task::ready!(crate::trace::trace_leaf(cx));
if status.is_ready() {
return status;
}
status = crate::runtime::coop::poll_proceed(cx).map(|restore| {
status = crate::task::coop::poll_proceed(cx).map(|restore| {
restore.made_progress();
});
status
@@ -1,10 +1,70 @@
#![cfg_attr(not(feature = "full"), allow(dead_code))]
#![cfg_attr(not(feature = "rt"), allow(unreachable_pub))]
//! Yield points for improved cooperative scheduling.
//! Utilities for improved cooperative scheduling.
//!
//! Documentation for this can be found in the [`tokio::task`] module.
//! ### Cooperative scheduling
//!
//! [`tokio::task`]: crate::task.
//! A single call to [`poll`] on a top-level task may potentially do a lot of
//! work before it returns `Poll::Pending`. If a task runs for a long period of
//! time without yielding back to the executor, it can starve other tasks
//! waiting on that executor to execute them, or drive underlying resources.
//! Since Rust does not have a runtime, it is difficult to forcibly preempt a
//! long-running task. Instead, this module provides an opt-in mechanism for
//! futures to collaborate with the executor to avoid starvation.
//!
//! Consider a future like this one:
//!
//! ```
//! # use tokio_stream::{Stream, StreamExt};
//! async fn drop_all<I: Stream + Unpin>(mut input: I) {
//! while let Some(_) = input.next().await {}
//! }
//! ```
//!
//! It may look harmless, but consider what happens under heavy load if the
//! input stream is _always_ ready. If we spawn `drop_all`, the task will never
//! yield, and will starve other tasks and resources on the same executor.
//!
//! To account for this, Tokio has explicit yield points in a number of library
//! functions, which force tasks to return to the executor periodically.
//!
//!
//! #### unconstrained
//!
//! If necessary, [`task::unconstrained`] lets you opt a future out of Tokio's cooperative
//! scheduling. When a future is wrapped with `unconstrained`, it will never be forced to yield to
//! Tokio. For example:
//!
//! ```
//! # #[tokio::main]
//! # async fn main() {
//! use tokio::{task, sync::mpsc};
//!
//! let fut = async {
//! let (tx, mut rx) = mpsc::unbounded_channel();
//!
//! for i in 0..1000 {
//! let _ = tx.send(());
//! // This will always be ready. If coop was in effect, this code would be forced to yield
//! // periodically. However, if left unconstrained, then this code will never yield.
//! rx.recv().await;
//! }
//! };
//!
//! task::coop::unconstrained(fut).await;
//! # }
//! ```
//! [`poll`]: method@std::future::Future::poll
//! [`task::unconstrained`]: crate::task::unconstrained()
cfg_rt! {
mod consume_budget;
pub use consume_budget::consume_budget;
mod unconstrained;
pub use unconstrained::{unconstrained, Unconstrained};
}
// ```ignore
// # use tokio_stream::{Stream, StreamExt};
@@ -57,7 +117,7 @@ impl Budget {
}
/// Returns an unconstrained budget. Operations will not be limited.
pub(super) const fn unconstrained() -> Budget {
pub(crate) const fn unconstrained() -> Budget {
Budget(None)
}
@@ -107,8 +167,60 @@ fn with_budget<R>(budget: Budget, f: impl FnOnce() -> R) -> R {
f()
}
/// Returns `true` if there is still budget left on the task.
///
/// # Examples
///
/// This example defines a `Timeout` future that requires a given `future` to complete before the
/// specified duration elapses. If it does, its result is returned; otherwise, an error is returned
/// and the future is canceled.
///
/// Note that the future could exhaust the budget before we evaluate the timeout. Using `has_budget_remaining`,
/// we can detect this scenario and ensure the timeout is always checked.
///
/// ```
/// # use std::future::Future;
/// # use std::pin::{pin, Pin};
/// # use std::task::{ready, Context, Poll};
/// # use tokio::task::coop;
/// # use tokio::time::Sleep;
/// pub struct Timeout<T> {
/// future: T,
/// delay: Pin<Box<Sleep>>,
/// }
///
/// impl<T> Future for Timeout<T>
/// where
/// T: Future + Unpin,
/// {
/// type Output = Result<T::Output, ()>;
///
/// fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
/// let this = Pin::into_inner(self);
/// let future = Pin::new(&mut this.future);
/// let delay = Pin::new(&mut this.delay);
///
/// // check if the future is ready
/// let had_budget_before = coop::has_budget_remaining();
/// if let Poll::Ready(v) = future.poll(cx) {
/// return Poll::Ready(Ok(v));
/// }
/// let has_budget_now = coop::has_budget_remaining();
///
/// // evaluate the timeout
/// if let (true, false) = (had_budget_before, has_budget_now) {
/// // it is the underlying future that exhausted the budget
/// ready!(pin!(coop::unconstrained(delay)).poll(cx));
/// } else {
/// ready!(delay.poll(cx));
/// }
/// return Poll::Ready(Err(()));
/// }
/// }
///```
#[inline(always)]
pub(crate) fn has_budget_remaining() -> bool {
#[cfg_attr(docsrs, doc(cfg(feature = "rt")))]
pub fn has_budget_remaining() -> bool {
// If the current budget cannot be accessed due to the thread-local being
// shutdown, then we assume there is budget remaining.
context::budget(|cell| cell.get().has_remaining()).unwrap_or(true)
@@ -193,12 +305,27 @@ cfg_coop! {
Poll::Ready(restore)
} else {
cx.waker().wake_by_ref();
register_waker(cx);
Poll::Pending
}
}).unwrap_or(Poll::Ready(RestoreOnPending(Cell::new(Budget::unconstrained()))))
}
/// Returns `Poll::Ready` if the current task has budget to consume, and `Poll::Pending` otherwise.
///
/// Note that in contrast to `poll_proceed`, this method does not consume any budget and is used when
/// polling for budget availability.
#[inline]
pub(crate) fn poll_budget_available(cx: &mut Context<'_>) -> Poll<()> {
if has_budget_remaining() {
Poll::Ready(())
} else {
register_waker(cx);
Poll::Pending
}
}
cfg_rt! {
cfg_unstable_metrics! {
#[inline(always)]
@@ -213,11 +340,19 @@ cfg_coop! {
#[inline(always)]
fn inc_budget_forced_yield_count() {}
}
fn register_waker(cx: &mut Context<'_>) {
context::defer(cx.waker());
}
}
cfg_not_rt! {
#[inline(always)]
fn inc_budget_forced_yield_count() {}
fn register_waker(cx: &mut Context<'_>) {
cx.waker().wake_by_ref()
}
}
impl Budget {
@@ -22,7 +22,7 @@ where
cfg_coop! {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let inner = self.project().inner;
crate::runtime::coop::with_unconstrained(|| inner.poll(cx))
crate::task::coop::with_unconstrained(|| inner.poll(cx))
}
}
+2 -2
View File
@@ -482,7 +482,7 @@ impl<T: 'static> JoinSet<T> {
/// Note that this method may return `Poll::Pending` even if one of the tasks has completed.
/// This can happen if the [coop budget] is reached.
///
/// [coop budget]: crate::task#cooperative-scheduling
/// [coop budget]: crate::task::coop#cooperative-scheduling
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.
@@ -537,7 +537,7 @@ impl<T: 'static> JoinSet<T> {
/// Note that this method may return `Poll::Pending` even if one of the tasks has completed.
/// This can happen if the [coop budget] is reached.
///
/// [coop budget]: crate::task#cooperative-scheduling
/// [coop budget]: crate::task::coop#cooperative-scheduling
/// [task ID]: crate::task::Id
pub fn poll_join_next_with_id(
&mut self,
+1 -1
View File
@@ -736,7 +736,7 @@ impl LocalSet {
// task initially. Because `LocalSet` itself is `!Send`, and
// `spawn_local` spawns into the `LocalSet` on the current
// thread, the invariant is maintained.
Some(task) => crate::runtime::coop::budget(|| task.run()),
Some(task) => crate::task::coop::budget(|| task.run()),
// We have fully drained the queue of notified tasks, so the
// local future doesn't need to be notified again — it can wait
// until something else wakes a task in the local set.
+14 -60
View File
@@ -260,66 +260,11 @@
//! # }
//! ```
//!
//! ### Cooperative scheduling
//!
//! A single call to [`poll`] on a top-level task may potentially do a lot of
//! work before it returns `Poll::Pending`. If a task runs for a long period of
//! time without yielding back to the executor, it can starve other tasks
//! waiting on that executor to execute them, or drive underlying resources.
//! Since Rust does not have a runtime, it is difficult to forcibly preempt a
//! long-running task. Instead, this module provides an opt-in mechanism for
//! futures to collaborate with the executor to avoid starvation.
//!
//! Consider a future like this one:
//!
//! ```
//! # use tokio_stream::{Stream, StreamExt};
//! async fn drop_all<I: Stream + Unpin>(mut input: I) {
//! while let Some(_) = input.next().await {}
//! }
//! ```
//!
//! It may look harmless, but consider what happens under heavy load if the
//! input stream is _always_ ready. If we spawn `drop_all`, the task will never
//! yield, and will starve other tasks and resources on the same executor.
//!
//! To account for this, Tokio has explicit yield points in a number of library
//! functions, which force tasks to return to the executor periodically.
//!
//!
//! #### unconstrained
//!
//! If necessary, [`task::unconstrained`] lets you opt a future out of Tokio's cooperative
//! scheduling. When a future is wrapped with `unconstrained`, it will never be forced to yield to
//! Tokio. For example:
//!
//! ```
//! # #[tokio::main]
//! # async fn main() {
//! use tokio::{task, sync::mpsc};
//!
//! let fut = async {
//! let (tx, mut rx) = mpsc::unbounded_channel();
//!
//! for i in 0..1000 {
//! let _ = tx.send(());
//! // This will always be ready. If coop was in effect, this code would be forced to yield
//! // periodically. However, if left unconstrained, then this code will never yield.
//! rx.recv().await;
//! }
//! };
//!
//! task::unconstrained(fut).await;
//! # }
//! ```
//!
//! [`task::spawn_blocking`]: crate::task::spawn_blocking
//! [`task::block_in_place`]: crate::task::block_in_place
//! [rt-multi-thread]: ../runtime/index.html#threaded-scheduler
//! [`task::yield_now`]: crate::task::yield_now()
//! [`thread::yield_now`]: std::thread::yield_now
//! [`task::unconstrained`]: crate::task::unconstrained()
//! [`poll`]: method@std::future::Future::poll
cfg_rt! {
pub use crate::runtime::task::{JoinError, JoinHandle};
@@ -337,8 +282,16 @@ cfg_rt! {
mod yield_now;
pub use yield_now::yield_now;
mod consume_budget;
pub use consume_budget::consume_budget;
pub mod coop;
#[doc(hidden)]
#[deprecated = "Moved to tokio::task::coop::consume_budget"]
pub use coop::consume_budget;
#[doc(hidden)]
#[deprecated = "Moved to tokio::task::coop::unconstrained"]
pub use coop::unconstrained;
#[doc(hidden)]
#[deprecated = "Moved to tokio::task::coop::Unconstrained"]
pub use coop::Unconstrained;
mod local;
pub use local::{spawn_local, LocalSet, LocalEnterGuard};
@@ -346,9 +299,6 @@ cfg_rt! {
mod task_local;
pub use task_local::LocalKey;
mod unconstrained;
pub use unconstrained::{unconstrained, Unconstrained};
#[doc(inline)]
pub use join_set::JoinSet;
pub use crate::runtime::task::AbortHandle;
@@ -371,3 +321,7 @@ cfg_rt! {
pub use super::task_local::TaskLocalFuture;
}
}
cfg_not_rt! {
pub(crate) mod coop;
}
+2 -2
View File
@@ -407,11 +407,11 @@ impl Sleep {
#[cfg(all(tokio_unstable, feature = "tracing"))]
let coop = ready!(trace_poll_op!(
"poll_elapsed",
crate::runtime::coop::poll_proceed(cx),
crate::task::coop::poll_proceed(cx),
));
#[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
let coop = ready!(crate::runtime::coop::poll_proceed(cx));
let coop = ready!(crate::task::coop::poll_proceed(cx));
let result = me.entry.poll_elapsed(cx).map(move |r| {
coop.made_progress();
+1 -1
View File
@@ -5,7 +5,7 @@
//! [`Timeout`]: struct@Timeout
use crate::{
runtime::coop,
task::coop,
time::{error::Elapsed, sleep_until, Duration, Instant, Sleep},
util::trace,
};
+29
View File
@@ -0,0 +1,29 @@
#[cfg(unix)]
use std::os::fd::AsFd;
#[cfg(unix)]
#[allow(unused_variables)]
#[track_caller]
pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
#[cfg(not(tokio_allow_from_blocking_fd))]
{
let sock = socket2::SockRef::from(s);
debug_assert!(
sock.nonblocking()?,
"Registering a blocking socket with the tokio runtime is unsupported. \
If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your \
RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details."
);
}
Ok(())
}
#[cfg(not(unix))]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S>(s: &S) -> crate::io::Result<()> {
// we cannot retrieve the nonblocking status on windows
// and i dont know how to support wasi yet
Ok(())
}
+6
View File
@@ -5,6 +5,12 @@ cfg_io_driver! {
#[cfg(feature = "rt")]
pub(crate) mod atomic_cell;
#[cfg(feature = "net")]
mod blocking_check;
#[cfg(feature = "net")]
#[allow(unused_imports)]
pub(crate) use blocking_check::check_socket_for_blocking;
pub(crate) mod metric_atomics;
#[cfg(any(feature = "rt", feature = "signal", feature = "process"))]
-2
View File
@@ -1,5 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#[cfg(not(any(feature = "full", target_family = "wasm")))]
compile_error!("run main Tokio tests with `--features full`");
+3
View File
@@ -394,6 +394,9 @@ assert_value!(tokio::sync::broadcast::Receiver<YY>: Send & Sync & Unpin);
assert_value!(tokio::sync::broadcast::Sender<NN>: !Send & !Sync & Unpin);
assert_value!(tokio::sync::broadcast::Sender<YN>: Send & Sync & Unpin);
assert_value!(tokio::sync::broadcast::Sender<YY>: Send & Sync & Unpin);
assert_value!(tokio::sync::broadcast::WeakSender<NN>: !Send & !Sync & Unpin);
assert_value!(tokio::sync::broadcast::WeakSender<YN>: Send & Sync & Unpin);
assert_value!(tokio::sync::broadcast::WeakSender<YY>: Send & Sync & Unpin);
assert_value!(tokio::sync::futures::Notified<'_>: Send & Sync & !Unpin);
assert_value!(tokio::sync::mpsc::OwnedPermit<NN>: !Send & !Sync & Unpin);
assert_value!(tokio::sync::mpsc::OwnedPermit<YN>: Send & Sync & Unpin);
+16 -1
View File
@@ -4,6 +4,9 @@
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::net::UdpSocket;
use tokio::task::coop::{consume_budget, has_budget_remaining};
const BUDGET: usize = 128;
/// Ensure that UDP sockets have functional budgeting
///
@@ -24,7 +27,6 @@ use tokio::net::UdpSocket;
#[tokio::test]
#[cfg_attr(miri, ignore)] // No `socket` on miri.
async fn coop_budget_udp_send_recv() {
const BUDGET: usize = 128;
const N_ITERATIONS: usize = 1024;
const PACKET: &[u8] = b"Hello, world";
@@ -76,3 +78,16 @@ async fn coop_budget_udp_send_recv() {
assert_eq!(N_ITERATIONS / (BUDGET / 2), tracker.load(Ordering::SeqCst));
}
#[tokio::test]
async fn test_has_budget_remaining() {
// At the beginning, budget should be available.
assert!(has_budget_remaining());
// Deplete the budget
for _ in 0..BUDGET {
consume_budget().await;
}
assert!(!has_budget_remaining());
}
-1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![cfg(all(
tokio_unstable,
tokio_taskdump,
+19
View File
@@ -1,6 +1,7 @@
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(target_os = "wasi")))] // WASI does not support all fs operations
use futures::future::FutureExt;
use std::io::prelude::*;
use std::io::IoSlice;
use tempfile::NamedTempFile;
@@ -176,6 +177,24 @@ async fn read_file_from_std() {
assert_eq!(&buf[..n], HELLO);
}
#[tokio::test]
async fn empty_read() {
let mut tempfile = tempfile();
tempfile.write_all(HELLO).unwrap();
let mut file = File::open(tempfile.path()).await.unwrap();
// Perform an empty read and get a length of zero.
assert!(matches!(file.read(&mut []).now_or_never(), Some(Ok(0))));
// Check that we don't get EOF on the next read.
let mut buf = [0; 1024];
let n = file.read(&mut buf).await.unwrap();
assert_eq!(n, HELLO.len());
assert_eq!(&buf[..n], HELLO);
}
fn tempfile() -> NamedTempFile {
NamedTempFile::new().unwrap()
}
+5 -2
View File
@@ -96,7 +96,10 @@ fn panics_when_io_disabled() {
let rt = runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let _ =
tokio::net::TcpListener::from_std(std::net::TcpListener::bind("127.0.0.1:0").unwrap());
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let _ = tokio::net::TcpListener::from_std(listener);
});
}
+6
View File
@@ -13,6 +13,9 @@ fn tcp_doesnt_block() {
let listener = {
let _enter = rt.enter();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
TcpListener::from_std(listener).unwrap()
};
@@ -33,6 +36,9 @@ fn drop_wakes() {
let listener = {
let _enter = rt.enter();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
TcpListener::from_std(listener).unwrap()
};
+24 -1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![cfg(feature = "macros")]
#![allow(clippy::disallowed_names)]
@@ -717,3 +716,27 @@ async fn temporary_lifetime_extension() {
() = &mut std::future::ready(()) => {},
}
}
#[tokio::test]
async fn select_is_budget_aware() {
const BUDGET: usize = 128;
let task = || {
Box::pin(async move {
tokio::select! {
biased;
() = tokio::task::coop::consume_budget() => {},
() = std::future::ready(()) => {}
}
})
};
for _ in 0..BUDGET {
let poll = futures::poll!(&mut task());
assert!(poll.is_ready());
}
let poll = futures::poll!(&mut task());
assert!(poll.is_pending());
}
-1
View File
@@ -1,4 +1,3 @@
#![allow(unknown_lints, unexpected_cfgs)]
#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi doesn't support threading
use tokio::test;
+5 -1
View File
@@ -39,5 +39,9 @@ async fn timeout_value() {
)]
#[cfg_attr(miri, ignore)] // No `socket` in miri.
fn io_panics_when_no_tokio_context() {
let _ = tokio::net::TcpListener::from_std(std::net::TcpListener::bind("127.0.0.1:0").unwrap());
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let _ = tokio::net::TcpListener::from_std(listener);
}

Some files were not shown because too many files have changed in this diff Show More