chore: prepare Tokio v1.51.0 (#8005)

`Chan::recv_many` intends to assert that no slots
have been consumed when exiting with `Ready` via
the `rx_closed` code path.

Instead of asserting no items were added to the
buffer, it asserted buffer emptiness, incorrectly
making assumptions about the provided buffer.

When `recv_many` was called on an empty channel
with idle semaphore after the receiver was closed,
the method would panic.

The branch coverage had been previously missing.

This changeset corrects the assertion
and adds tests covering the code path.

Fixes #7990.

.cirrus.yml

@@ -1,10 +1,10 @@
 only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
 auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
 freebsd_instance:
-  image_family: freebsd-14-2
+  image_family: freebsd-14-3
 env:
   RUST_STABLE: stable
-  RUST_NIGHTLY: nightly-2025-01-25
+  RUST_NIGHTLY: nightly-2025-10-12
   RUSTFLAGS: -D warnings
   # This excludes unstable features like io_uring, which require '--cfg tokio_unstable'.
   TOKIO_STABLE_FEATURES: full,test-util
@@ -25,7 +25,7 @@ task:
       rustc --version
   test_script:
     - . $HOME/.cargo/env
-    - cargo test --all --features $TOKIO_STABLE_FEATURES
+    - cargo test --workspace --features $TOKIO_STABLE_FEATURES
     # Free the disk space before the next build,
     # otherwise cirrus-ci complains about "No space left on device".
     - cargo clean
@@ -68,4 +68,4 @@ task:
       rustc --version
   test_script:
     - . $HOME/.cargo/env
-    - cargo test --all --features $TOKIO_STABLE_FEATURES --target i686-unknown-freebsd
+    - cargo test --workspace --features $TOKIO_STABLE_FEATURES --target i686-unknown-freebsd

.github/labeler.yml

@@ -1,3 +1,6 @@
+R-loom-blocking:
+- tokio/src/runtime/blocking/*
+- tokio/src/runtime/blocking/**/*
 
 R-loom-sync:
 - tokio/src/sync/*
@@ -19,3 +22,7 @@ R-loom-multi-thread:
 - tokio/src/runtime/scheduler/multi_thread/**
 - tokio/src/runtime/task/*
 - tokio/src/runtime/task/**
+
+R-loom-util:
+- tokio-util/src/*
+- tokio-util/src/**/*

.github/workflows/ci.yml

@@ -18,7 +18,7 @@ env:
   rust_stable: stable
   rust_nightly: nightly-2025-10-12
   # Pin a specific miri version
-  rust_miri_nightly: nightly-2025-06-02
+  rust_miri_nightly: nightly-2025-11-13
   rust_clippy: '1.88'
   # When updating this, also update:
   # - README.md
@@ -112,6 +112,16 @@ jobs:
         run: |
           set -euxo pipefail
           cargo nextest run --workspace --features $TOKIO_STABLE_FEATURES
+          # Removing workspace patches to run tests without path dependencies
+          # (if not specified differently in the crate)
+          perl -0 -i -pe 's/\[patch\.crates-io\].+\n\[/[/s' Cargo.toml
+          cargo nextest run \
+            --workspace \
+            --exclude tokio \
+            --exclude examples \
+            --features $TOKIO_STABLE_FEATURES
+          # Cargo nextest does not support doctest, so we run them separately
+          # (see https://github.com/nextest-rs/nextest/issues/16)
           cargo test --doc --workspace --features $TOKIO_STABLE_FEATURES
 
   test-workspace-all-features-panic-abort:
@@ -211,7 +221,13 @@ jobs:
       - uses: Swatinem/rust-cache@v2
 
       - name: Check tests --features ${{ env.TOKIO_STABLE_FEATURES }}
-        run: cargo check --workspace --tests --features $TOKIO_STABLE_FEATURES
+        run: |
+          set -euxo pipefail
+          cargo check --workspace --tests --features $TOKIO_STABLE_FEATURES
+          # Removing the tokio workspace patch to run tests without path dependencies
+          # (if not specified differently in the crate)
+          perl -0 -i -pe 's/\[patch\.crates-io\].+\n\[/[/s' Cargo.toml
+          cargo check --workspace --exclude tokio --tests --features $TOKIO_STABLE_FEATURES
 
   valgrind:
     name: valgrind
@@ -372,7 +388,7 @@ jobs:
           cargo miri nextest run --features full --lib --no-fail-fast
         working-directory: tokio
         env:
-          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance
 
   miri-test:
     name: miri-test
@@ -395,7 +411,7 @@ jobs:
           cargo miri nextest run --features full --test '*' --no-fail-fast
         working-directory: tokio
         env:
-          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance
 
   miri-doc:
     name: miri-doc
@@ -414,7 +430,7 @@ jobs:
           cargo miri test --doc --features $TOKIO_STABLE_FEATURES --no-fail-fast
         working-directory: tokio
         env:
-          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance
 
   asan:
     name: asan
@@ -716,6 +732,10 @@ jobs:
         uses: dtolnay/rust-toolchain@stable
         with:
           toolchain: ${{ env.rust_min }}
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-hack
       - uses: Swatinem/rust-cache@v2
       - name: "cargo check"
         run: |
@@ -728,7 +748,7 @@ jobs:
 
             cargo check -p tokio --features $TOKIO_STABLE_FEATURES
             # Other crates doesn't have unstable features, so we can use --all-features.
-            cargo check -p tokio-macros -p tokio-stream -p tokio-util -p tokio-test --all-features
+            cargo hack check -p tokio-macros -p tokio-stream -p tokio-util -p tokio-test --all-features
           fi
 
   minimal-versions:
@@ -797,10 +817,23 @@ jobs:
           components: clippy
       - uses: Swatinem/rust-cache@v2
       # Run clippy
-      - name: "clippy --all --features ${{ env.TOKIO_STABLE_FEATURES }}"
-        run: cargo clippy --all --tests --no-deps --features $TOKIO_STABLE_FEATURES
-      - name: "clippy --all --all-features --unstable"
-        run: cargo clippy --all --tests --no-deps --all-features
+      - name: "clippy --workspace --features ${{ env.TOKIO_STABLE_FEATURES }}"
+        run: |
+          cargo clippy --workspace --tests --no-deps --features $TOKIO_STABLE_FEATURES
+          # Removing the tokio workspace patch to check without path dependencies
+          # (if not specified differently in the crate)
+          perl -0 -i.bak -pe 's/\[patch\.crates-io\].+\n\[/[/s' Cargo.toml
+          cargo clippy --workspace --exclude tokio --tests --no-deps --features $TOKIO_STABLE_FEATURES
+      - name: "clippy --workspace --all-features --unstable"
+        run: |
+          # Forcing the cargo lock regeneration to apply the tokio patch
+          rm Cargo.lock
+          mv Cargo.toml Cargo.toml.nopatch
+          mv Cargo.toml.bak Cargo.toml
+          cargo clippy --workspace --tests --no-deps --all-features
+          # check without path dependencies
+          mv Cargo.toml.nopatch Cargo.toml
+          cargo clippy --workspace --exclude tokio --tests --no-deps --all-features
         env:
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings
 
@@ -1046,7 +1079,7 @@ jobs:
         run: cargo test -p tokio --target ${{ matrix.target }} --features "sync,macros,io-util,rt,time"
         env:
           CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
           # in order to run doctests for unstable features, we must also pass
           # the unstable cfg to RustDoc
@@ -1056,14 +1089,15 @@ jobs:
         run: cargo test -p tokio-util --target ${{ matrix.target }} --features full
         env:
           CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
+          RUSTDOCFLAGS: -C link-args=--max-memory=67108864
 
       - name: WASI test tokio-stream
-        run: cargo test -p tokio-stream --target ${{ matrix.target }} --features time,net,io-util,sync
+        run: cargo test --manifest-path=tokio-stream/Cargo.toml --target ${{ matrix.target }} --features time,net,io-util,sync
         env:
           CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: test tests-integration --features wasi-rt
@@ -1080,9 +1114,35 @@ jobs:
         if: matrix.target == 'wasm32-wasip1-threads'
         working-directory: tests-integration
         env:
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
+  wasm32-wasip2:
+    name: test tokio for wasm32-wasip2
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: ${{ env.rust_stable }}
+          targets: wasm32-wasip2
+
+      - name: Install cargo-nextest, wasmtime
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-nextest,wasmtime-cli
+
+      - uses: Swatinem/rust-cache@v2
+
+      - name: test tokio --target wasm32-wasip2
+        run: cargo nextest run --target wasm32-wasip2 --features net,macros,rt,io-util
+        working-directory: tokio
+        env:
+          RUSTFLAGS: --cfg tokio_unstable
+          CARGO_TARGET_WASM32_WASIP2_RUNNER: wasmtime run -Sinherit-network
+
   check-external-types:
     name: check-external-types (${{ matrix.os }})
     needs: basics

.github/workflows/loom.yml

@@ -23,6 +23,22 @@ permissions:
   contents: read
 
 jobs:
+  loom-blocking:
+    name: loom tokio::runtime::spawn_blocking
+    # base_ref is null when it's not a pull request
+    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-blocking') || (github.base_ref == null))
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v5
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
+      - name: run tests
+        run: cargo test --lib --release --features full -- --nocapture loom_blocking
+        working-directory: tokio
+
   loom-sync:
     name: loom tokio::sync
     # base_ref is null when it's not a pull request
@@ -52,7 +68,7 @@ jobs:
             toolchain: ${{ env.rust_stable }}
       - uses: Swatinem/rust-cache@v2
       - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture runtime::time::tests
+        run: cargo test --lib --release --features full -- --nocapture runtime::time
         working-directory: tokio
 
   loom-current-thread:
@@ -95,3 +111,19 @@ jobs:
         working-directory: tokio
         env:
           SCOPE: ${{ matrix.scope }}
+
+  loom-util:
+    name: loom tokio-util
+    # base_ref is null when it's not a pull request
+    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-util') || (github.base_ref == null))
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v5
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
+      - name: run tests
+        run: cargo test --lib --release --features full -- --nocapture
+        working-directory: tokio-util

.gitignore

@@ -3,3 +3,4 @@ Cargo.lock
 
 .cargo/config.toml
 .cargo/config
+

CONTRIBUTING.md

@@ -1,7 +1,6 @@
 # Contributing to Tokio
 
-:balloon: Thanks for your help improving the project! We are so happy to have
-you!
+Thanks for your help improving Tokio! We are so happy to have you!
 
 There are opportunities to contribute to Tokio at any level. It doesn't matter if
 you are just getting started with Rust or are the most weathered expert, we can
@@ -9,15 +8,11 @@ use your help.
 
 **No contribution is too small and all contributions are valued.**
 
-This guide will help you get started. **Do not let this guide intimidate you**.
-It should be considered a map to help you navigate the process.
+See the [contributing guidelines] to get started.
 
-The [dev channel][dev] is available for any concerns not covered in this guide, please join
-us!
+[contributing guidelines]: docs/contributing/README.md
 
-[dev]: https://discord.gg/tokio
-
-## Conduct
+## Code of Conduct
 
 The Tokio project adheres to the [Rust Code of Conduct][coc]. This describes
 the _minimum_ behavior expected from all contributors. Instances of violations of the
@@ -26,709 +21,34 @@ Code of Conduct can be reported by contacting the project team at
 
 [coc]: https://github.com/rust-lang/rust/blob/master/CODE_OF_CONDUCT.md
 
-## Contributing in Issues
+## Need Help?
 
-For any issue, there are fundamentally three ways an individual can contribute:
+Reach out to us on the [Discord server] for any concern not covered in this guide.
 
-1. By opening the issue for discussion: For instance, if you believe that you
-   have discovered a bug in Tokio, creating a new issue in [the tokio-rs/tokio
-   issue tracker][issue] is the way to report it.
-
-2. By helping to triage the issue: This can be done by providing
-   supporting details (a test case that demonstrates a bug), providing
-   suggestions on how to address the issue, or ensuring that the issue is tagged
-   correctly.
-
-3. By helping to resolve the issue: Typically this is done either in the form of
-   demonstrating that the issue reported is not a problem after all, or more
-   often, by opening a Pull Request that changes some bit of something in
-   Tokio in a concrete and reviewable manner.
-
-[issue]: https://github.com/tokio-rs/tokio/issues
-
-**Anybody can participate in any stage of contribution**. We urge you to
-participate in the discussion around bugs and participate in reviewing PRs.
-
-### Asking for General Help
-
-If you have reviewed existing documentation and still have questions or are
-having problems, you can [open a discussion] asking for help.
-
-In exchange for receiving help, we ask that you contribute back a documentation
-PR that helps others avoid the problems that you encountered.
-
-[open a discussion]: https://github.com/tokio-rs/tokio/discussions/new
-
-### Submitting a Bug Report
-
-When opening a new issue in the Tokio issue tracker, you will be presented
-with a basic template that should be filled in. If you believe that you have
-uncovered a bug, please fill out this form, following the template to the best
-of your ability. Do not worry if you cannot answer every detail, just fill in
-what you can.
-
-The two most important pieces of information we need in order to properly
-evaluate the report is a description of the behavior you are seeing and a simple
-test case we can use to recreate the problem on our own. If we cannot recreate
-the issue, it becomes impossible for us to fix.
-
-In order to rule out the possibility of bugs introduced by userland code, test
-cases should be limited, as much as possible, to using only Tokio APIs.
-
-See [How to create a Minimal, Complete, and Verifiable example][mcve].
-
-[mcve]: https://stackoverflow.com/help/mcve
-
-### Triaging a Bug Report
-
-Once an issue has been opened, it is not uncommon for there to be discussion
-around it. Some contributors may have differing opinions about the issue,
-including whether the behavior being seen is a bug or a feature. This discussion
-is part of the process and should be kept focused, helpful, and professional.
-
-Short, clipped responses—that provide neither additional context nor supporting
-detail—are not helpful or professional. To many, such responses are simply
-annoying and unfriendly.
-
-Contributors are encouraged to help one another make forward progress as much as
-possible, empowering one another to solve issues collaboratively. If you choose
-to comment on an issue that you feel either is not a problem that needs to be
-fixed, or if you encounter information in an issue that you feel is incorrect,
-explain why you feel that way with additional supporting context, and be willing
-to be convinced that you may be wrong. By doing so, we can often reach the
-correct outcome much faster.
-
-### Resolving a Bug Report
-
-In the majority of cases, issues are resolved by opening a Pull Request. The
-process for opening and reviewing a Pull Request is similar to that of opening
-and triaging issues, but carries with it a necessary review and approval
-workflow that ensures that the proposed changes meet the minimal quality and
-functional guidelines of the Tokio project.
-
-## Pull Requests
-
-Pull Requests are the way concrete changes are made to the code, documentation,
-and dependencies in the Tokio repository.
-
-Even tiny pull requests (e.g., one character pull request fixing a typo in API
-documentation) are greatly appreciated. Before making a large change, it is
-usually a good idea to first open an issue describing the change to solicit
-feedback and guidance. This will increase the likelihood of the PR getting
-merged.
-
-### Cargo Commands
-
-Due to the extensive use of features in Tokio, you will often need to add extra
-arguments to many common cargo commands. This section lists some commonly needed
-commands.
-
-Some commands just need the `--all-features` argument:
-
-```
-cargo build --all-features
-cargo check --all-features
-cargo test --all-features
-```
-
-Ideally, you should use the same version of clippy as the one used in CI
-(defined by `env.rust_clippy` in [ci.yml][ci.yml]), because newer versions
-might have new lints:
-
-[ci.yml]: .github/workflows/ci.yml
-
-<!--
-When updating this, also update:
-- .github/workflows/ci.yml
-- README.md
-- tokio/README.md
-- tokio/Cargo.toml
-- tokio-util/Cargo.toml
-- tokio-test/Cargo.toml
-- tokio-stream/Cargo.toml
--->
-
-```
-cargo +1.88 clippy --all --tests --all-features
-```
-
-When building documentation, a simple `cargo doc` is not sufficient. To produce
-documentation equivalent to what will be produced in docs.rs's builds of Tokio's
-docs, please use:
-
-```
-RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg docsrs --cfg tokio_unstable" cargo +nightly doc --all-features [--open]
-```
-
-This turns on indicators to display the Cargo features required for
-conditionally compiled APIs in Tokio, and it enables documentation of unstable
-Tokio features. Notice that it is necessary to pass cfg flags to both RustDoc
-*and* rustc.
-
-There is a more concise way to build docs.rs-equivalent docs by using [`cargo
-docs-rs`], which reads the above documentation flags out of Tokio's Cargo.toml
-as docs.rs itself does.
-
-[`cargo docs-rs`]: https://github.com/dtolnay/cargo-docs-rs
-
-```
-cargo install --locked cargo-docs-rs
-cargo +nightly docs-rs [--open]
-```
-
-The `cargo fmt` command does not work on the Tokio codebase. You can use the
-command below instead:
-
-```
-# Mac or Linux
-rustfmt --check --edition 2021 $(git ls-files '*.rs')
-
-# Powershell
-Get-ChildItem . -Filter "*.rs" -Recurse | foreach { rustfmt --check --edition 2021 $_.FullName }
-```
-The `--check` argument prints the things that need to be fixed. If you remove
-it, `rustfmt` will update your files locally instead.
-
-You can run loom tests with
-```
-cd tokio # tokio crate in workspace
-LOOM_MAX_PREEMPTIONS=1 LOOM_MAX_BRANCHES=10000 RUSTFLAGS="--cfg loom -C debug_assertions" \
-    cargo test --lib --release --features full -- --test-threads=1 --nocapture
-```
-Additionally, you can also add `--cfg tokio_unstable` to the `RUSTFLAGS` environment variable to
-run loom tests that test unstable features.
-
-You can run miri tests with
-```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields" \
-    cargo +nightly miri test --features full --lib --tests
-```
-
-### Performing spellcheck on tokio codebase
-
-You can perform spell-check on tokio codebase. For details of how to use the spellcheck tool, feel free to visit
-https://github.com/drahnr/cargo-spellcheck
-```
-# First install the spell-check plugin
-cargo install --locked cargo-spellcheck
-
-# Then run the cargo spell check command
-cargo spellcheck check
-```
-
-if the command rejects a word, you should backtick the rejected word if it's code related. If not, the
-rejected word should be put into `spellcheck.dic` file.
-
-Note that when you add a word into the file, you should also update the first line which tells the spellcheck tool
-the total number of words included in the file
-
-### Tests
-
-If the change being proposed alters code (as opposed to only documentation for
-example), it is either adding new functionality to Tokio or it is fixing
-existing, broken functionality. In both of these cases, the pull request should
-include one or more tests to ensure that Tokio does not regress in the future.
-There are two ways to write tests: [integration tests][integration-tests]
-and [documentation tests][documentation-tests].
-(Tokio avoids [unit tests][unit-tests] as much as possible).
-
-Tokio uses [conditional compilation attributes][conditional-compilation]
-throughout the codebase, to modify rustc's behavior. Code marked with such
-attributes can be enabled using RUSTFLAGS and RUSTDOCFLAGS environment
-variables. One of the most prevalent flags passed in these variables is
-the `--cfg` option. To run tests in a particular file, check first what
-options #![cfg] declaration defines for that file.
-
-For instance, to run a test marked with the 'tokio_unstable' cfg option,
-you must pass this flag to the compiler when running the test.
-```
-$ RUSTFLAGS="--cfg tokio_unstable" cargo test -p tokio --all-features --test rt_metrics
-```
-
-#### Integration tests
-
-Integration tests go in the same crate as the code they are testing. Each sub
-crate should have a `dev-dependency` on `tokio` itself. This makes all Tokio
-utilities available to use in tests, no matter the crate being tested.
-
-The best strategy for writing a new integration test is to look at existing
-integration tests in the crate and follow the style.
-
-#### Fuzz tests
-
-Some of our crates include a set of fuzz tests, this will be marked by a
-directory `fuzz`. It is a good idea to run fuzz tests after each change.
-To get started with fuzz testing you'll need to install
-[cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz).
-
-`cargo install --locked cargo-fuzz`
-
-To list the available fuzzing harnesses you can run;
-
-```bash
-$ cd tokio
-$ cargo fuzz list
-fuzz_linked_list
-```
-
-Running a fuzz test is as simple as;
-
-`cargo fuzz run fuzz_linked_list`
-
-**NOTE**: Keep in mind that by default when running a fuzz test the fuzz
-harness will run forever and will only exit if you `ctrl-c` or it finds
-a bug.
-
-#### Documentation tests
-
-Ideally, every API has at least one [documentation test] that demonstrates how to
-use the API. Documentation tests are run with `cargo test --doc`. This ensures
-that the example is correct and provides additional test coverage.
-
-The trick to documentation tests is striking a balance between being succinct
-for a reader to understand and actually testing the API.
-
-Same as with integration tests, when writing a documentation test, the full
-`tokio` crate is available. This is especially useful for getting access to the
-runtime to run the example.
-
-The documentation tests will be visible from both the crate specific
-documentation **and** the `tokio` facade documentation via the re-export. The
-example should be written from the point of view of a user that is using the
-`tokio` crate. As such, the example should use the API via the facade and not by
-directly referencing the crate.
-
-The type level example for `tokio_timer::Timeout` provides a good example of a
-documentation test:
-
-```
-/// // import the `timeout` function, usually this is done
-/// // with `use tokio::prelude::*`
-/// use tokio::prelude::FutureExt;
-/// use futures::Stream;
-/// use futures::sync::mpsc;
-/// use std::time::Duration;
-///
-/// # fn main() {
-/// let (tx, rx) = mpsc::unbounded();
-/// # tx.unbounded_send(()).unwrap();
-/// # drop(tx);
-///
-/// let process = rx.for_each(|item| {
-///     // do something with `item`
-/// # drop(item);
-/// # Ok(())
-/// });
-///
-/// # tokio::runtime::current_thread::block_on_all(
-/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
-/// process.timeout(Duration::from_millis(10))
-/// # ).unwrap();
-/// # }
-```
-
-Given that this is a *type* level documentation test and the primary way users
-of `tokio` will create an instance of `Timeout` is by using
-`FutureExt::timeout`, this is how the documentation test is structured.
-
-Lines that start with `/// #` are removed when the documentation is generated.
-They are only there to get the test to run. The `block_on_all` function is the
-easiest way to execute a future from a test.
-
-If this were a documentation test for the `Timeout::new` function, then the
-example would explicitly use `Timeout::new`. For example:
-
-```
-/// use tokio::timer::Timeout;
-/// use futures::Future;
-/// use futures::sync::oneshot;
-/// use std::time::Duration;
-///
-/// # fn main() {
-/// let (tx, rx) = oneshot::channel();
-/// # tx.send(()).unwrap();
-///
-/// # tokio::runtime::current_thread::block_on_all(
-/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
-/// Timeout::new(rx, Duration::from_millis(10))
-/// # ).unwrap();
-/// # }
-```
-
-### Benchmarks
-
-You can run benchmarks locally for the changes you've made to the tokio codebase.
-Tokio currently uses [Criterion](https://github.com/bheisler/criterion.rs) as its benchmarking tool. To run a benchmark
-against the changes you have made, for example, you can run;
-
-```bash
-cd benches
-
-# Run all benchmarks.
-cargo bench
-
-# Run all tests in the `benches/fs.rs` file
-cargo bench --bench fs
-
-# Run the `async_read_buf` benchmark in `benches/fs.rs` specifically.
-cargo bench async_read_buf
-
-# After running benches, you can check the statistics under `tokio/target/criterion/`
-```
-
-You can also refer to Criterion docs for additional options and details.
-
-### Commits
-
-It is a recommended best practice to keep your changes as logically grouped as
-possible within individual commits. There is no limit to the number of commits
-any single Pull Request may have, and many contributors find it easier to review
-changes that are split across multiple commits.
-
-That said, if you have a number of commits that are "checkpoints" and don't
-represent a single logical change, please squash those together.
-
-Note that multiple commits often get squashed when they are landed (see the
-notes about [commit squashing](#commit-squashing)).
-
-#### Commit message guidelines
-
-A good commit message should describe what changed and why.
-
-1. The first line should:
-
-  * contain a short description of the change (preferably 50 characters or less,
-    and no more than 72 characters)
-  * be entirely in lowercase with the exception of proper nouns, acronyms, and
-    the words that refer to code, like function/variable names
-  * start with an imperative verb
-  * not have a period at the end
-  * be prefixed with the name of the module being changed; usually this is the
-    same as the M-* label on the PR
-
-  Examples:
-
-  * time: introduce `Timeout` and deprecate `Deadline`
-  * codec: export `Encoder`, `Decoder`, `Framed*`
-  * ci: fix the FreeBSD ci configuration
-
-2. Keep the second line blank.
-3. Wrap all other lines at 72 columns (except for long URLs).
-4. If your patch fixes an open issue, you can add a reference to it at the end
-   of the log. Use the `Fixes: #` prefix and the issue number. For other
-   references use `Refs: #`. `Refs` may include multiple issues, separated by a
-   comma.
-
-   Examples:
-
-   - `Fixes: #1337`
-   - `Refs: #1234`
-
-Sample complete commit message:
-
-```txt
-module: explain the commit in one line
-
-Body of commit message is a few lines of text, explaining things
-in more detail, possibly giving some background about the issue
-being fixed, etc.
-
-The body of the commit message can be several paragraphs, and
-please do proper word-wrap and keep columns shorter than about
-72 characters or so. That way, `git log` will show things
-nicely even when it is indented.
-
-Fixes: #1337
-Refs: #453, #154
-```
-
-### Opening the Pull Request
-
-From within GitHub, opening a new Pull Request will present you with a
-[template] that should be filled out. Please try to do your best at filling out
-the details, but feel free to skip parts if you're not sure what to put.
-
-[template]: .github/PULL_REQUEST_TEMPLATE.md
-
-### Discuss and update
-
-You will probably get feedback or requests for changes to your Pull Request.
-This is a big part of the submission process so don't be discouraged! Some
-contributors may sign off on the Pull Request right away, others may have
-more detailed comments or feedback. This is a necessary part of the process
-in order to evaluate whether the changes are correct and necessary.
-
-**Any community member can review a PR and you might get conflicting feedback**.
-Keep an eye out for comments from code owners to provide guidance on conflicting
-feedback.
-
-**Once the PR is open, do not rebase the commits**. See [Commit Squashing](#commit-squashing) for
-more details.
-
-### Commit Squashing
-
-In most cases, **do not squash commits that you add to your Pull Request during
-the review process**. When the commits in your Pull Request land, they may be
-squashed into one commit per logical change. Metadata will be added to the
-commit message (including links to the Pull Request, links to relevant issues,
-and the names of the reviewers). The commit history of your Pull Request,
-however, will stay intact on the Pull Request page.
-
-## Reviewing Pull Requests
-
-**Any Tokio community member is welcome to review any pull request**.
-
-All Tokio contributors who choose to review and provide feedback on Pull
-Requests have a responsibility to both the project and the individual making the
-contribution. Reviews and feedback must be helpful, insightful, and geared
-towards improving the contribution as opposed to simply blocking it. If there
-are reasons why you feel the PR should not land, explain what those are. Do not
-expect to be able to block a Pull Request from advancing simply because you say
-"No" without giving an explanation. Be open to having your mind changed. Be open
-to working with the contributor to make the Pull Request better.
-
-Reviews that are dismissive or disrespectful of the contributor or any other
-reviewers are strictly counter to the Code of Conduct.
-
-When reviewing a Pull Request, the primary goals are for the codebase to improve
-and for the person submitting the request to succeed. **Even if a Pull Request
-does not land, the submitters should come away from the experience feeling like
-their effort was not wasted or unappreciated**. Every Pull Request from a new
-contributor is an opportunity to grow the community.
-
-### Review a bit at a time.
-
-Do not overwhelm new contributors.
-
-It is tempting to micro-optimize and make everything about relative performance,
-perfect grammar, or exact style matches. Do not succumb to that temptation.
-
-Focus first on the most significant aspects of the change:
-
-1. Does this change make sense for Tokio?
-2. Does this change make Tokio better, even if only incrementally?
-3. Are there clear bugs or larger scale issues that need attending to?
-4. Is the commit message readable and correct? If it contains a breaking change
-   is it clear enough?
-
-Note that only **incremental** improvement is needed to land a PR. This means
-that the PR does not need to be perfect, only better than the status quo. Follow
-up PRs may be opened to continue iterating.
-
-When changes are necessary, *request* them, do not *demand* them, and **do not
-assume that the submitter already knows how to add a test or run a benchmark**.
-
-Specific performance optimization techniques, coding styles and conventions
-change over time. The first impression you give to a new contributor never does.
-
-Nits (requests for small changes that are not essential) are fine, but try to
-avoid stalling the Pull Request. Most nits can typically be fixed by the Tokio
-Collaborator landing the Pull Request but they can also be an opportunity for
-the contributor to learn a bit more about the project.
-
-It is always good to clearly indicate nits when you comment: e.g.
-`Nit: change foo() to bar(). But this is not blocking.`
-
-If your comments were addressed but were not folded automatically after new
-commits or if they proved to be mistaken, please, [hide them][hiding-a-comment]
-with the appropriate reason to keep the conversation flow concise and relevant.
-
-### Be aware of the person behind the code
-
-Be aware that *how* you communicate requests and reviews in your feedback can
-have a significant impact on the success of the Pull Request. Yes, we may land
-a particular change that makes Tokio better, but the individual might just not
-want to have anything to do with Tokio ever again. The goal is not just having
-good code.
-
-### Abandoned or Stalled Pull Requests
-
-If a Pull Request appears to be abandoned or stalled, it is polite to first
-check with the contributor to see if they intend to continue the work before
-checking if they would mind if you took it over (especially if it just has nits
-left). When doing so, it is courteous to give the original contributor credit
-for the work they started (either by preserving their name and email address in
-the commit log, or by using an `Author: ` meta-data tag in the commit.
-
-_Adapted from the [Node.js contributing guide][node]_.
-
-[node]: https://github.com/nodejs/node/blob/master/CONTRIBUTING.md
-[hiding-a-comment]: https://help.github.com/articles/managing-disruptive-comments/#hiding-a-comment
-[documentation test]: https://doc.rust-lang.org/rustdoc/documentation-tests.html
-
-## Keeping track of issues and PRs
-
-The Tokio GitHub repository has a lot of issues and PRs to keep track of. This
-section explains the meaning of various labels, as well as our [GitHub
-project][project]. The section is primarily targeted at maintainers.  Most
-contributors aren't able to set these labels.
-
-### Area
-
-The area label describes the crates relevant to this issue or PR.
-
- - **A-tokio** This issue concerns the main Tokio crate.
- - **A-tokio-util** This issue concerns the `tokio-util` crate.
- - **A-tokio-tls** This issue concerns the `tokio-tls` crate. Only used for
-   older issues, as the crate has been moved to another repository.
- - **A-tokio-test** The issue concerns the `tokio-test` crate.
- - **A-tokio-macros** This issue concerns the `tokio-macros` crate. Should only
-   be used for the procedural macros, and not `join!` or `select!`.
- - **A-ci** This issue concerns our GitHub Actions setup.
-
-### Category
-
- - **C-bug** This is a bug-report. Bug-fix PRs use `C-enhancement` instead.
- - **C-enhancement** This is a PR that adds a new features.
- - **C-maintenance** This is an issue or PR about stuff such as documentation,
-   GitHub Actions or code quality.
- - **C-feature-request** This is a feature request. Implementations of feature
-   requests use `C-enhancement` instead.
- - **C-feature-accepted** If you submit a PR for this feature request, we wont
-   close it with the reason "we don't want this". Issues with this label should
-   also have the `C-feature-request` label.
- - **C-musing** Stuff like tracking issues or roadmaps. "musings about a better
-   world"
- - **C-proposal** A proposal of some kind, and a request for comments.
- - **C-question** A user question. Large overlap with GitHub discussions.
- - **C-request** A non-feature request, e.g. "please add deprecation notices to
-   `-alpha.*` versions of crates"
-
-### Calls for participation
-
- - **E-help-wanted** Stuff where we want help. Often seen together with `C-bug`
-   or `C-feature-accepted`.
- - **E-easy** This is easy, ranging from quick documentation fixes to stuff you
-   can do after reading the tutorial on our website.
- - **E-medium** This is not `E-easy` or `E-hard`.
- - **E-hard** This either involves very tricky code, is something we don't know
-   how to solve, or is difficult for some other reason.
- - **E-needs-mvce** This bug is missing a minimal complete and verifiable
-   example.
-
-The "E-" prefix is the same as used in the Rust compiler repository. Some
-issues are missing a difficulty rating, but feel free to ask on our Discord
-server if you want to know how difficult an issue likely is.
-
-### Module
-
-The module label provides a more fine grained categorization than **Area**.
-
- - **M-blocking** Things relevant to `spawn_blocking`, `block_in_place`.
- - **M-codec** The `tokio_util::codec` module.
- - **M-compat** The `tokio_util::compat` module.
- - **M-coop** Things relevant to coop.
- - **M-fs** The `tokio::fs` module.
- - **M-io** The `tokio::io` module.
- - **M-macros** Issues about any kind of macro.
- - **M-net** The `tokio::net` module.
- - **M-process** The `tokio::process` module.
- - **M-runtime** The `tokio::runtime` module.
- - **M-signal** The `tokio::signal` module.
- - **M-sync** The `tokio::sync` module.
- - **M-task** The `tokio::task` module.
- - **M-time** The `tokio::time` module.
- - **M-tracing** Tracing support in Tokio.
-
-### Topic
-
-Some extra information.
-
- - **T-docs** This is about documentation.
- - **T-performance** This is about performance.
- - **T-v0.1.x** This is about old Tokio.
-
-Any label not listed here is not in active use.
-
-[project]: https://github.com/orgs/tokio-rs/projects/1
+[Discord server]: https://discord.gg/tokio
 
 ## LTS guarantees
 
-Tokio ≥1.0.0 comes with LTS guarantees:
-
- * A minimum of 5 years of maintenance.
- * A minimum of 3 years before a hypothetical 2.0 release.
+In Tokio ≥1.0.0, each LTS release comes with the guarantee of at least one year of
+backported fixes.
 
 The goal of these guarantees is to provide stability to the ecosystem.
 
 ## Minimum Supported Rust Version (MSRV)
 
- * All Tokio ≥1.0.0 releases will support at least a 6-month old Rust
-   compiler release.
- * The MSRV will only be increased on 1.x releases.
+* All Tokio ≥1.0.0 releases will support at least a 6-month old Rust
+  compiler release.
+* The MSRV will only be increased on 1.x releases.
 
 ## Versioning Policy
 
 With Tokio ≥1.0.0:
 
- * Patch (1.\_.x) releases _should only_ contain bug fixes or documentation
-   changes. Besides this, these releases should not substantially change
-   runtime behavior.
- * Minor (1.x) releases may contain new functionality, MSRV increases (see
-   above), minor dependency updates, deprecations, and larger internal
-   implementation changes.
+* Patch (1.\_.x) releases _should only_ contain bug fixes or documentation
+  changes. Besides this, these releases should not substantially change
+  runtime behavior.
+* Minor (1.x) releases may contain new functionality, MSRV increases (see
+  above), minor dependency updates, deprecations, and larger internal
+  implementation changes.
 
 This is as defined by [Semantic Versioning 2.0](https://semver.org/).
-
-## Releasing
-
-Since the Tokio project consists of a number of crates, many of which depend on
-each other, releasing new versions to crates.io can involve some complexities.
-When releasing a new version of a crate, follow these steps:
-
-1. **Ensure that the release crate has no path dependencies.** When the HEAD
-   version of a Tokio crate requires unreleased changes in another Tokio crate,
-   the crates.io dependency on the second crate will be replaced with a path
-   dependency. Crates with path dependencies cannot be published, so before
-   publishing the dependent crate, any path dependencies must also be published.
-   This should be done through a form of depth-first tree traversal:
-
-   1. Starting with the first path dependency in the crate to be released,
-      inspect the `Cargo.toml` for the dependency. If the dependency has any
-      path dependencies of its own, repeat this step with the first such
-      dependency.
-   2. Begin the release process for the path dependency.
-   3. Once the path dependency has been published to crates.io, update the
-      dependent crate to depend on the crates.io version.
-   4. When all path dependencies have been published, the dependent crate may
-      be published.
-
-   To verify that a crate is ready to publish, run:
-
-   ```bash
-   bin/publish --dry-run <CRATE NAME> <CRATE VERSION>
-   ```
-
-2. **Update Cargo metadata.** After releasing any path dependencies, update the
-   `version` field in `Cargo.toml` to the new version, and the `documentation`
-   field to the docs.rs URL of the new version.
-3. **Update other documentation links.** Update the "Documentation" link in the
-   crate's `README.md` to point to the docs.rs URL of the new version.
-4. **Update the changelog for the crate.** Each crate in the Tokio repository
-   has its own `CHANGELOG.md` in that crate's subdirectory. Any changes to that
-   crate since the last release should be added to the changelog. Change
-   descriptions may be taken from the Git history, but should be edited to
-   ensure a consistent format, based on [Keep A Changelog][keep-a-changelog].
-   Other entries in that crate's changelog may also be used for reference.
-5. **Perform a final audit for breaking changes.** Compare the HEAD version of
-   crate with the Git tag for the most recent release version. If there are any
-   breaking API changes, determine if those changes can be made without breaking
-   existing APIs. If so, resolve those issues. Otherwise, if it is necessary to
-   make a breaking release, update the version numbers to reflect this.
-6. **Open a pull request with your changes.** Once that pull request has been
-   approved by a maintainer and the pull request has been merged, continue to
-   the next step.
-7. **Release the crate.** Run the following command:
-
-   ```bash
-   bin/publish <NAME OF CRATE> <VERSION>
-   ```
-
-   Your editor and prompt you to edit a message for the tag. Copy the changelog
-   entry for that release version into your editor and close the window.
-
-[keep-a-changelog]: https://github.com/olivierlacan/keep-a-changelog/blob/master/CHANGELOG.md
-[unit-tests]: https://doc.rust-lang.org/rust-by-example/testing/unit_testing.html
-[integration-tests]: https://doc.rust-lang.org/rust-by-example/testing/integration_testing.html
-[documentation-tests]: https://doc.rust-lang.org/rust-by-example/testing/doc_testing.html
-[conditional-compilation]: https://doc.rust-lang.org/reference/conditional-compilation.html

Cargo.toml

@@ -15,6 +15,13 @@ members = [
   "tests-integration",
 ]
 
+[patch.crates-io]
+tokio = { path = "tokio" }
+tokio-macros = { path = "tokio-macros" }
+tokio-stream = { path = "tokio-stream" }
+tokio-test = { path = "tokio-test" }
+tokio-util = { path = "tokio-util" }
+
 [workspace.metadata.spellcheck]
 config = "spellcheck.toml"
 

README.md

@@ -1,3 +1,7 @@
+*[TokioConf 2026 program and tickets are now available!](https://tokioconf.com)*
+
+---
+
 # Tokio
 
 A runtime for writing reliable, asynchronous, and slim applications with
@@ -52,11 +56,11 @@ an asynchronous application.
 
 A basic TCP echo server with Tokio.
 
-Make sure you activated the full features of the tokio crate on Cargo.toml:
+Make sure you enable the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.48.0", features = ["full"] }
+tokio = { version = "1.51.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -103,7 +107,7 @@ More examples can be found [here][examples]. For a larger "real world" example,
 [examples]: https://github.com/tokio-rs/tokio/tree/master/examples
 [mini-redis]: https://github.com/tokio-rs/mini-redis/
 
-To see a list of the available features flags that can be enabled, check our
+To see a list of the available feature flags that can be enabled, check our
 [docs][feature-flag-docs].
 
 ## Getting Help
@@ -125,7 +129,7 @@ question. You can also ask your question on [the discussions page][discussions].
 you! We have a [contributing guide][guide] to help you get involved in the Tokio
 project.
 
-[guide]: https://github.com/tokio-rs/tokio/blob/master/CONTRIBUTING.md
+[guide]: https://github.com/tokio-rs/tokio/blob/master/docs/contributing/README.md
 
 ## Related Projects
 
@@ -217,18 +221,18 @@ 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.43.x` - LTS release until March 2026. (MSRV 1.70)
  * `1.47.x` - LTS release until September 2026. (MSRV 1.70)
+ * `1.51.x` - LTS release until March 2027. (MSRV 1.71)
 
 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
 use an LTS release.
 
 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.43.x` patch release, you
+example, to specify that you wish to use the newest `1.47.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.43", features = [...] }
+tokio = { version = "~1.47", features = [...] }
 ```
 
 ### Previous LTS releases
@@ -241,6 +245,7 @@ tokio = { version = "~1.43", features = [...] }
  * `1.32.x` - LTS release until September 2024.
  * `1.36.x` - LTS release until March 2025.
  * `1.38.x` - LTS release until July 2025.
+ * `1.43.x` - LTS release until March 2026.
 
 ## License
 
@@ -251,5 +256,5 @@ This project is licensed under the [MIT license].
 ### Contribution
 
 Unless you explicitly state otherwise, any contribution intentionally submitted
-for inclusion in Tokio by you, shall be licensed as MIT, without any additional
+for inclusion in Tokio by you shall be licensed as MIT, without any additional
 terms or conditions.

benches/Cargo.toml

@@ -96,5 +96,15 @@ name = "time_timeout"
 path = "time_timeout.rs"
 harness = false
 
+[[bench]]
+name = "spawn_blocking"
+path = "spawn_blocking.rs"
+harness = false
+
+[[bench]]
+name = "remote_spawn"
+path = "remote_spawn.rs"
+harness = false
+
 [lints]
 workspace = true

benches/remote_spawn.rs

@@ -0,0 +1,103 @@
+//! Benchmark remote task spawning (push_remote_task) at different concurrency
+//! levels on the multi-threaded scheduler.
+//!
+//! This measures contention on the scheduler's inject queue mutex when multiple
+//! external (non-worker) threads spawn tasks into the tokio runtime simultaneously.
+//! Every rt.spawn() from an external thread unconditionally goes through
+//! push_remote_task, making this a direct measurement of inject queue contention.
+//!
+//! For each parallelism level N (1, 2, 4, 8, 16, 32, 64, capped at available parallelism):
+//! - Spawns N std::threads (external to the runtime)
+//! - Each thread spawns TOTAL_TASKS / N tasks into the runtime via rt.spawn()
+//! - All threads are synchronized with a barrier to maximize contention
+//! - Tasks are trivial no-ops to isolate the push overhead
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use std::sync::Barrier;
+use tokio::runtime::{self, Runtime};
+
+/// Total number of tasks spawned across all threads per iteration.
+/// Must be divisible by the largest parallelism level (64).
+const TOTAL_TASKS: usize = 12_800;
+const _: () = assert!(
+    TOTAL_TASKS.is_multiple_of(64),
+    "TOTAL_TASKS must be divisible by 64"
+);
+
+fn remote_spawn_contention(c: &mut Criterion) {
+    let parallelism_levels = parallelism_levels();
+    let mut group = c.benchmark_group("remote_spawn");
+
+    for num_threads in &parallelism_levels {
+        let num_threads = *num_threads;
+        group.bench_with_input(
+            BenchmarkId::new("threads", num_threads),
+            &num_threads,
+            |b, &num_threads| {
+                let rt = rt();
+                let tasks_per_thread = TOTAL_TASKS / num_threads;
+                let barrier = Barrier::new(num_threads);
+
+                b.iter_custom(|iters| {
+                    let mut total_duration = std::time::Duration::ZERO;
+                    for _ in 0..iters {
+                        let start = std::time::Instant::now();
+
+                        let all_handles = std::thread::scope(|s| {
+                            let handles: Vec<_> = (0..num_threads)
+                                .map(|_| {
+                                    let barrier = &barrier;
+                                    let rt = &rt;
+                                    s.spawn(move || {
+                                        let mut join_handles = Vec::with_capacity(tasks_per_thread);
+                                        barrier.wait();
+
+                                        for _ in 0..tasks_per_thread {
+                                            join_handles.push(rt.spawn(async {}));
+                                        }
+                                        join_handles
+                                    })
+                                })
+                                .collect();
+
+                            handles
+                                .into_iter()
+                                .flat_map(|h| h.join().unwrap())
+                                .collect::<Vec<_>>()
+                        });
+
+                        total_duration += start.elapsed();
+
+                        rt.block_on(async {
+                            for h in all_handles {
+                                h.await.unwrap();
+                            }
+                        });
+                    }
+                    total_duration
+                });
+            },
+        );
+    }
+
+    group.finish();
+}
+
+fn parallelism_levels() -> Vec<usize> {
+    let max_parallelism = std::thread::available_parallelism()
+        .map(|p| p.get())
+        .unwrap_or(1);
+
+    [1, 2, 4, 8, 16, 32, 64]
+        .into_iter()
+        .filter(|&n| n <= max_parallelism)
+        .collect()
+}
+
+fn rt() -> Runtime {
+    runtime::Builder::new_multi_thread().build().unwrap()
+}
+
+criterion_group!(remote_spawn_benches, remote_spawn_contention);
+
+criterion_main!(remote_spawn_benches);

benches/spawn_blocking.rs

@@ -0,0 +1,73 @@
+//! Benchmark spawn_blocking at different concurrency levels on the multi-threaded scheduler.
+//!
+//! For each parallelism level N (1, 2, 4, 8, 16, 32, 64, capped at available parallelism):
+//! - Spawns N regular async tasks
+//! - Each task spawns M batches of B spawn_blocking tasks (no-ops)
+//! - Each batch is awaited to completion before starting the next
+
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+use tokio::runtime::{self, Runtime};
+use tokio::task::JoinSet;
+
+/// Number of batches per task
+const NUM_BATCHES: usize = 100;
+/// Number of spawn_blocking calls per batch
+const BATCH_SIZE: usize = 16;
+
+fn spawn_blocking_concurrency(c: &mut Criterion) {
+    let max_parallelism = std::thread::available_parallelism()
+        .map(|p| p.get())
+        .unwrap_or(1);
+
+    let parallelism_levels: Vec<usize> = [1, 2, 4, 8, 16, 32, 64]
+        .into_iter()
+        .filter(|&n| n <= max_parallelism)
+        .collect();
+
+    let mut group = c.benchmark_group("spawn_blocking");
+
+    for num_tasks in parallelism_levels {
+        group.bench_with_input(
+            BenchmarkId::new("concurrency", num_tasks),
+            &num_tasks,
+            |b, &num_tasks| {
+                let rt = rt();
+
+                b.iter(|| {
+                    rt.block_on(async {
+                        let mut tasks = JoinSet::new();
+
+                        for _ in 0..num_tasks {
+                            tasks.spawn(async {
+                                for _ in 0..NUM_BATCHES {
+                                    let mut batch = JoinSet::new();
+
+                                    for _ in 0..BATCH_SIZE {
+                                        batch.spawn_blocking(|| black_box(0));
+                                    }
+
+                                    batch.join_all().await;
+                                }
+                            });
+                        }
+
+                        tasks.join_all().await;
+                    });
+                });
+            },
+        );
+    }
+
+    group.finish();
+}
+
+fn rt() -> Runtime {
+    runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+}
+
+criterion_group!(spawn_blocking_benches, spawn_blocking_concurrency);
+
+criterion_main!(spawn_blocking_benches);

docs/contributing/README.md

@@ -0,0 +1,46 @@
+# Contributing
+
+This guide will help you get started. **Do not let this guide intimidate you**.
+It should be considered a map to help you navigate the process.
+
+## Quick start
+
+If you are unsure where to begin, use the following guides:
+
+- Want to report or triage a bug? Start with [Contributing in Issues](contributing-in-issues.md).
+- Looking for something to work on? Filter issues by [`E-help-wanted`](https://github.com/tokio-rs/tokio/labels/E-help-wanted).
+- Planning to submit a PR? Read [Pull Requests](pull-requests.md) for the full workflow and required checks.
+- Want to understand what the labels on issues mean? See [Keeping track of issues and PRs](keeping-track-of-issues-and-prs.md).
+- Interested in code review? See [Reviewing Pull Requests](reviewing-pull-requests.md).
+
+## Table of Contents
+
+- [Contributing in Issues](contributing-in-issues.md)
+    - [Asking for General Help](contributing-in-issues.md#asking-for-general-help)
+    - [Submitting a Bug Report](contributing-in-issues.md#submitting-a-bug-report)
+    - [Triaging a Bug Report](contributing-in-issues.md#triaging-a-bug-report)
+    - [Resolving a Bug Report](contributing-in-issues.md#resolving-a-bug-report)
+- [Pull Requests](pull-requests.md)
+    - [Cargo Commands](pull-requests.md#cargo-commands)
+    - [Performing spellcheck on tokio codebase](pull-requests.md#performing-spellcheck-on-tokio-codebase)
+    - [Tests](pull-requests.md#tests)
+        - [Integration tests](pull-requests.md#integration-tests)
+        - [Fuzz tests](pull-requests.md#fuzz-tests)
+        - [Documentation tests](pull-requests.md#documentation-tests)
+    - [Benchmarks](pull-requests.md#benchmarks)
+    - [Commits](pull-requests.md#commits)
+        - [Commit message guidelines](pull-requests.md#commit-message-guidelines)
+    - [Opening the Pull Request](pull-requests.md#opening-the-pull-request)
+    - [Discuss and update](pull-requests.md#discuss-and-update)
+    - [Commit Squashing](pull-requests.md#commit-squashing)
+- [Reviewing Pull Requests](reviewing-pull-requests.md)
+    - [Review a bit at a time](reviewing-pull-requests.md#review-a-bit-at-a-time)
+    - [Be aware of the person behind the code](reviewing-pull-requests.md#be-aware-of-the-person-behind-the-code)
+    - [Abandoned or Stalled Pull Requests](reviewing-pull-requests.md#abandoned-or-stalled-pull-requests)
+- [How to specify crates dependencies versions](how-to-specify-crates-dependencies-versions.md)
+- [Keeping track of issues and PRs](keeping-track-of-issues-and-prs.md)
+    - [Area](keeping-track-of-issues-and-prs.md#area)
+    - [Category](keeping-track-of-issues-and-prs.md#category)
+    - [Calls for participation](keeping-track-of-issues-and-prs.md#calls-for-participation)
+    - [Module](keeping-track-of-issues-and-prs.md#module)
+    - [Topic](keeping-track-of-issues-and-prs.md#topic)

docs/contributing/contributing-in-issues.md

@@ -0,0 +1,79 @@
+## Contributing in Issues
+
+For any issue, there are fundamentally three ways an individual can contribute:
+
+1. By opening the issue for discussion: For instance, if you believe that you
+   have discovered a bug in Tokio, creating a new issue in [the tokio-rs/tokio
+   issue tracker][issue] is the way to report it.
+
+2. By helping to triage the issue: This can be done by providing
+   supporting details (a test case that demonstrates a bug), providing
+   suggestions on how to address the issue, or ensuring that the issue is tagged
+   correctly.
+
+3. By helping to resolve the issue: Typically this is done either in the form of
+   demonstrating that the issue reported is not a problem after all, or more
+   often, by opening a Pull Request that changes some bit of something in
+   Tokio in a concrete and reviewable manner.
+
+[issue]: https://github.com/tokio-rs/tokio/issues
+
+**Anybody can participate in any stage of contribution**. We urge you to
+participate in the discussion around bugs and participate in reviewing PRs.
+
+### Asking for General Help
+
+If you have reviewed existing documentation and still have questions or are
+having problems, you can [open a discussion] asking for help.
+
+In exchange for receiving help, we ask that you contribute back a documentation
+PR that helps others avoid the problems that you encountered.
+
+[open a discussion]: https://github.com/tokio-rs/tokio/discussions/new/choose
+
+### Submitting a Bug Report
+
+When opening a new issue in the Tokio issue tracker, you will be presented
+with a basic template that should be filled in. If you believe that you have
+uncovered a bug, please fill out this form, following the template to the best
+of your ability. Do not worry if you cannot answer every detail, just fill in
+what you can.
+
+The two most important pieces of information we need in order to properly
+evaluate the report is a description of the behavior you are seeing and a simple
+test case we can use to recreate the problem on our own. If we cannot recreate
+the issue, it becomes impossible for us to fix.
+
+In order to rule out the possibility of bugs introduced by userland code, test
+cases should be limited, as much as possible, to using only Tokio APIs.
+
+See [How to create a Minimal, Complete, and Verifiable example][mcve].
+
+[mcve]: https://stackoverflow.com/help/mcve
+
+### Triaging a Bug Report
+
+Once an issue has been opened, it is not uncommon for there to be discussion
+around it. Some contributors may have differing opinions about the issue,
+including whether the behavior being seen is a bug or a feature. This discussion
+is part of the process and should be kept focused, helpful, and professional.
+
+Short, clipped responses—that provide neither additional context nor supporting
+detail—are not helpful or professional. To many, such responses are simply
+annoying and unfriendly.
+
+Contributors are encouraged to help one another make forward progress as much as
+possible, empowering one another to solve issues collaboratively. If you choose
+to comment on an issue that you feel either is not a problem that needs to be
+fixed, or if you encounter information in an issue that you feel is incorrect,
+explain why you feel that way with additional supporting context, and be willing
+to be convinced that you may be wrong. By doing so, we can often reach the
+correct outcome much faster.
+
+### Resolving a Bug Report
+
+In the majority of cases, issues are resolved by opening a Pull Request. The
+process for opening and reviewing a Pull Request is similar to that of opening
+and triaging issues, but carries with it a necessary review and approval
+workflow that ensures that the proposed changes meet the minimal quality and
+functional guidelines of the Tokio project.

docs/contributing/how-to-specify-crates-dependencies-versions.md

@@ -0,0 +1,31 @@
+# How to specify crates dependencies versions
+
+Each crate (e.g., `tokio-util`, `tokio-stream`, etc.) should specify dependencies
+according to these rules:
+
+1. The listed version should be the oldest version that the crate works with
+(e.g., if `tokio-util` works with `tokio` version `1.44` but not `1.43`, then
+`tokio-util` should specify version `1.44` for its `tokio` dependency).
+We don't require users to use the latest version unnecessarily.
+2. When a crate starts using a newer feature in a dependency, the version
+should be bumped to the version that introduced it.
+3. If a crate depends on an unreleased feature in a dependency, it may use
+`path = ` dependency to specify this. Since path dependencies must be removed
+during the release of the crate, this ensures that it can't be released until
+the dependency has a new version.
+
+Consider the following example from `tokio-stream`:
+
+```toml
+[dependencies]
+futures-core = { version = "0.3.0" }
+pin-project-lite = "0.2.11"
+tokio = { path = "../tokio", features = ["sync"] }
+```
+
+In this case, local development of `tokio-stream` uses the local version
+of `tokio` via the `path` dependency. This means that it's currently not
+possible to release `tokio-stream`. Once a new version of `tokio` is
+released, the path dependency will be removed from `tokio-stream`.
+As mentioned before, this version should only be bumped when adding a new
+feature in the crate that relies on a newer version.

docs/contributing/keeping-track-of-issues-and-prs.md

@@ -0,0 +1,89 @@
+## Keeping track of issues and PRs
+
+The Tokio GitHub repository has a lot of issues and PRs to keep track of. This
+section explains the meaning of various labels, as well as our [GitHub
+project][project]. The section is primarily targeted at maintainers.  Most
+contributors aren't able to set these labels.
+
+### Area
+
+The area label describes the crates relevant to this issue or PR.
+
+- **A-ci** This issue concerns our GitHub Actions setup.
+- **A-tokio** This issue concerns the main Tokio crate.
+- **A-readme** This issue is related to documentation such as README.md.
+- **A-benches** This issue concerns the benchmarks.
+- **A-examples** This issue concerns the examples.
+- **A-tokio-test** The issue concerns the `tokio-test` crate.
+- **A-tokio-util** This issue concerns the `tokio-util` crate.
+- **A-tokio-macros** This issue concerns the `tokio-macros` crate. Should only
+  be used for the procedural macros, and not `join!` or `select!`.
+- **A-tokio-stream** This issue concerns the `tokio-stream` crate.
+
+### Category
+
+- **C-bug** This is a bug-report. Bug-fix PRs use `C-enhancement` instead.
+- **C-enhancement** This is a PR that adds a new features.
+- **C-maintenance** This is an issue or PR about stuff such as documentation,
+  GitHub Actions or code quality.
+- **C-feature-request** This is a feature request. Implementations of feature
+  requests use `C-enhancement` instead.
+- **C-feature-accepted** If you submit a PR for this feature request, we won't
+  close it with the reason "we don't want this". Issues with this label should
+  also have the `C-feature-request` label.
+- **C-musing** Stuff like tracking issues or roadmaps. "musings about a better
+  world"
+- **C-proposal** A proposal of some kind, and a request for comments.
+- **C-question** A user question. Large overlap with GitHub discussions.
+- **C-request** A non-feature request, e.g. "please add deprecation notices to
+  `-alpha.*` versions of crates"
+
+### Calls for participation
+
+- **E-help-wanted** Stuff where we want help. Often seen together with `C-bug`
+  or `C-feature-accepted`.
+- **E-easy** This is easy, ranging from quick documentation fixes to stuff you
+  can do after reading the tutorial on our website.
+- **E-medium** This is not `E-easy` or `E-hard`.
+- **E-hard** This either involves very tricky code, is something we don't know
+  how to solve, or is challenging for some other reason.
+- **E-needs-mvce** This bug is missing a minimal complete and verifiable
+  example.
+
+The "E-" prefix is the same as used in the Rust compiler repository. Some
+issues are missing a difficulty rating, but feel free to ask on our Discord
+server if you want to know how challenging an issue likely is.
+
+### Module
+
+The module label provides a more fine grained categorization than **Area**.
+
+- **M-blocking** Things relevant to `spawn_blocking`, `block_in_place`.
+- **M-codec** The `tokio_util::codec` module.
+- **M-compat** The `tokio_util::compat` module.
+- **M-coop** Things relevant to coop.
+- **M-fs** The `tokio::fs` module.
+- **M-io** The `tokio::io` module.
+- **M-macros** Issues about any kind of macro.
+- **M-metrics** Things relevant to `tokio::runtime::metrics`.
+- **M-net** The `tokio::net` module.
+- **M-process** The `tokio::process` module.
+- **M-runtime** The `tokio::runtime` module.
+- **M-signal** The `tokio::signal` module.
+- **M-sync** The `tokio::sync` module.
+- **M-task** The `tokio::task` module.
+- **M-time** The `tokio::time` module.
+- **M-tracing** Tracing support in Tokio.
+- **M-taskdump** Things relevant to taskdump.
+
+### Topic
+
+Some extra information.
+
+- **T-docs** This is about documentation.
+- **T-performance** This is about performance.
+- **T-v0.1.x** This is about old Tokio.
+
+Any label not listed here is not in active use.
+
+[project]: https://github.com/orgs/tokio-rs/projects/1

docs/contributing/pull-requests.md

@@ -0,0 +1,346 @@
+## Pull Requests
+
+Pull Requests are the way concrete changes are made to the code, documentation,
+and dependencies in the Tokio repository.
+
+Even tiny pull requests (e.g., one-character pull request fixing a typo in API
+documentation) are greatly appreciated. Before making a large change, it is
+usually a good idea to first open an issue describing the change to solicit
+feedback and guidance. This will increase the likelihood of the PR getting
+merged.
+
+### Cargo Commands
+
+Due to the extensive use of features in Tokio, you will often need to add extra
+arguments to many common cargo commands. This section lists some commonly needed
+commands.
+
+Some commands just need the `--all-features` argument:
+
+```
+cargo build --all-features
+cargo check --all-features
+cargo test --all-features
+```
+
+**NOTE**: there are some features that are not supported in every system, so you might
+need to specify which features you want to pass to cargo (e.g., `cargo check --features=full,io-uring`)
+
+Ideally, you should use the same version of clippy as the one used in CI
+(defined by `env.rust_clippy` in [ci.yml][ci.yml]), because newer versions
+might have new lints:
+
+[ci.yml]: ../../.github/workflows/ci.yml
+
+<!--
+When updating this, also update:
+- .github/workflows/ci.yml
+- README.md
+- tokio/README.md
+- tokio/Cargo.toml
+- tokio-util/Cargo.toml
+- tokio-test/Cargo.toml
+- tokio-stream/Cargo.toml
+-->
+
+```
+cargo +1.88 clippy --all --tests --all-features
+```
+
+When building documentation, a simple `cargo doc` is not sufficient. To produce
+documentation equivalent to what will be produced in docs.rs's builds of Tokio's
+docs, please use:
+
+```
+RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg docsrs --cfg tokio_unstable" cargo +nightly doc --all-features [--open]
+```
+
+This turns on indicators to display the Cargo features required for
+conditionally compiled APIs in Tokio, and it enables documentation of unstable
+Tokio features. Notice that it is necessary to pass cfg flags to both RustDoc
+*and* rustc.
+
+There is a more concise way to build docs.rs-equivalent docs by using [`cargo
+docs-rs`], which reads the above documentation flags out of Tokio's Cargo.toml
+as docs.rs itself does.
+
+[`cargo docs-rs`]: https://github.com/dtolnay/cargo-docs-rs
+
+```
+cargo install --locked cargo-docs-rs
+cargo +nightly docs-rs [--open]
+```
+
+The `cargo fmt` command does not work on the Tokio codebase. You can use the
+command below instead:
+
+```
+# Mac or Linux
+rustfmt --check --edition 2021 $(git ls-files '*.rs')
+
+# Powershell
+Get-ChildItem . -Filter "*.rs" -Recurse | foreach { rustfmt --check --edition 2021 $_.FullName }
+```
+The `--check` argument prints the things that need to be fixed. If you remove
+it, `rustfmt` will update your files locally instead.
+
+You can run loom tests with
+```
+cd tokio # tokio crate in workspace
+LOOM_MAX_PREEMPTIONS=1 LOOM_MAX_BRANCHES=10000 RUSTFLAGS="--cfg loom -C debug_assertions" \
+    cargo test --lib --release --features full -- --test-threads=1 --nocapture
+```
+Additionally, you can also add `--cfg tokio_unstable` to the `RUSTFLAGS` environment variable to
+run loom tests that test unstable features.
+
+You can run miri tests with
+```
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-strict-provenance" \
+    cargo +nightly miri test --features full --lib --tests
+```
+
+### Performing spellcheck on tokio codebase
+
+You can perform a spell-check on the Tokio codebase. For details of how to use the spellcheck tool, feel free to visit
+https://github.com/drahnr/cargo-spellcheck
+```
+# First install the spell-check plugin
+cargo install --locked cargo-spellcheck
+
+# Then run the cargo spell check command
+cargo spellcheck check
+```
+
+If the command rejects a word, you should backtick the rejected word if it's code related. If not, the
+rejected word should be put into `spellcheck.dic` file.
+
+Note that when you add a word into the file, you should also update the first line which tells the spellcheck tool
+the total number of words included in the file
+
+### Tests
+
+If the change being proposed alters code (as opposed to only documentation for
+example), it is either adding new functionality to Tokio or it is fixing
+existing, broken functionality. In both of these cases, the pull request should
+include one or more tests to ensure that Tokio does not regress in the future.
+There are two ways to write tests: [integration tests][integration-tests]
+and [documentation tests][documentation-tests].
+(Tokio avoids [unit tests][unit-tests] as much as possible).
+
+Tokio uses [conditional compilation attributes][conditional-compilation]
+throughout the codebase, to modify rustc's behavior. Code marked with such
+attributes can be enabled using `RUSTFLAGS` and `RUSTDOCFLAGS` environment
+variables. One of the most prevalent flags passed in these variables is
+the `--cfg` option. To run tests in a particular file, check first what
+options #![cfg] declaration defines for that file.
+
+For instance, to run a test marked with the 'tokio_unstable' cfg option,
+you must pass this flag to the compiler when running the test.
+```
+$ RUSTFLAGS="--cfg tokio_unstable" cargo test -p tokio --all-features --test rt_metrics
+```
+
+#### Integration tests
+
+Integration tests go in the same crate as the code they are testing. Each sub
+crate should have a `dev-dependency` on `tokio` itself. This makes all Tokio
+utilities available to use in tests, no matter the crate being tested.
+
+The best strategy for writing a new integration test is to look at existing
+integration tests in the crate and follow the style.
+
+#### Fuzz tests
+
+Some of our crates include a set of fuzz tests, this will be marked by a
+directory `fuzz`. It is a good idea to run fuzz tests after each change.
+To get started with fuzz testing you'll need to install
+[cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz).
+
+`cargo install --locked cargo-fuzz`
+
+To list the available fuzzing harnesses you can run;
+
+```bash
+$ cd tokio
+$ cargo fuzz list
+fuzz_linked_list
+```
+
+Running a fuzz test is as simple as;
+
+`cargo fuzz run fuzz_linked_list`
+
+**NOTE**: Keep in mind that by default when running a fuzz test the fuzz
+harness will run forever and will only exit if you `ctrl-c` or it finds
+a bug.
+
+#### Documentation tests
+
+Ideally, every API has at least one [documentation test] that demonstrates how to
+use the API. Documentation tests are run with `cargo test --doc`. This ensures
+that the example is correct and provides additional test coverage.
+
+The trick to documentation tests is striking a balance between being succinct
+for a reader to understand and actually testing the API.
+
+Same as with integration tests, when writing a documentation test, the full
+`tokio` crate is available. This is especially useful for getting access to the
+runtime to run the example.
+
+The documentation tests will be visible from both the crate-specific
+documentation **and** the `tokio` facade documentation via the re-export. The
+example should be written from the point of view of a user that is using the
+`tokio` crate. As such, the example should use the API via the facade and not by
+directly referencing the crate.
+
+The type level example for `tokio::time::timeout` provides a good example of a
+documentation test:
+
+```
+/// Create a new `Timeout` set to expire in 10 milliseconds.
+///
+/// ```rust
+/// use tokio::time::timeout;
+/// use tokio::sync::oneshot;
+///
+/// use std::time::Duration;
+///
+/// # async fn dox() {
+/// let (tx, rx) = oneshot::channel();
+/// # tx.send(()).unwrap();
+///
+/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
+/// if let Err(_) = timeout(Duration::from_millis(10), rx).await {
+///     println!("did not receive value within 10 ms");
+/// }
+/// # }
+/// ```
+```
+
+Lines that start with `/// #` are removed when the documentation is generated.
+
+### Benchmarks
+
+You can run benchmarks locally for the changes you've made to the tokio codebase.
+Tokio currently uses [Criterion](https://github.com/bheisler/criterion.rs) as its benchmarking tool. To run a benchmark
+against the changes you have made, for example, you can run;
+
+```bash
+cd benches
+
+# Run all benchmarks.
+cargo bench
+
+# Run all tests in the `benches/fs.rs` file
+cargo bench --bench fs
+
+# Run the `async_read_buf` benchmark in `benches/fs.rs` specifically.
+cargo bench async_read_buf
+
+# After running benches, you can check the statistics under `tokio/target/criterion/`
+```
+
+You can also refer to [Criterion] docs for additional options and details.
+
+[Criterion]: https://docs.rs/criterion/latest/criterion/
+
+### Commits
+
+It is a recommended best practice to keep your changes as logically grouped as
+possible within individual commits. There is no limit to the number of commits
+any single Pull Request may have, and many contributors find it easier to review
+changes that are split across multiple commits.
+
+That said, if you have a number of commits that are "checkpoints" and don't
+represent a single logical change, please squash those together.
+
+Note that multiple commits often get squashed when they are landed (see the
+notes about [commit squashing](#commit-squashing)).
+
+#### Commit message guidelines
+
+A good commit message should describe what changed and why.
+
+1. The first line should:
+
+* contain a short description of the change (preferably 50 characters or less,
+  and no more than 72 characters)
+* be entirely in lowercase with the exception of proper nouns, acronyms, and
+  the words that refer to code, like function/variable names
+* start with an imperative verb
+* not have a period at the end
+* be prefixed with the name of the module being changed; usually this is the
+  same as the M-* label on the PR
+
+Examples:
+
+* time: introduce `Timeout` and deprecate `Deadline`
+* codec: export `Encoder`, `Decoder`, `Framed*`
+* ci: fix the FreeBSD ci configuration
+
+2. Keep the second line blank.
+3. Wrap all other lines at 72 columns (except for long URLs).
+4. If your patch fixes an open issue, you can add a reference to it at the end
+   of the log. Use the `Fixes: #` prefix and the issue number. For other
+   references use `Refs: #`. `Refs` may include multiple issues, separated by a
+   comma.
+
+   Examples:
+
+    - `Fixes: #1337`
+    - `Refs: #1234`
+
+Sample complete commit message:
+
+```txt
+module: explain the commit in one line
+
+Body of commit message is a few lines of text, explaining things
+in more detail, possibly giving some background about the issue
+being fixed, etc.
+
+The body of the commit message can be several paragraphs, and
+please do proper word-wrap and keep columns shorter than about
+72 characters or so. That way, `git log` will show things
+nicely even when it is indented.
+
+Fixes: #1337
+Refs: #453, #154
+```
+
+### Opening the Pull Request
+
+From within GitHub, opening a new Pull Request will present you with a
+[template] that should be filled out. Please try to do your best at filling out
+the details, but feel free to skip parts if you're not sure what to put.
+
+[template]: ../../.github/PULL_REQUEST_TEMPLATE.md
+
+### Discuss and update
+
+You will probably get feedback or requests for changes to your Pull Request.
+This is a big part of the submission process so don't be discouraged! Some
+contributors may sign off on the Pull Request right away, others may have
+more detailed comments or feedback. This is a necessary part of the process
+in order to evaluate whether the changes are correct and necessary.
+
+**Any community member can review a PR and you might get conflicting feedback**.
+Keep an eye out for comments from code owners to provide guidance on conflicting
+feedback.
+
+**Once the PR is open, do not rebase the commits**. See [Commit Squashing](#commit-squashing) for
+more details.
+
+### Commit Squashing
+
+In most cases, **do not squash commits that you add to your Pull Request during
+the review process**. When the commits in your Pull Request land, they may be
+squashed into one commit per logical change. Metadata will be added to the
+commit message (including links to the Pull Request, links to relevant issues,
+and the names of the reviewers). The commit history of your Pull Request,
+however, will stay intact on the Pull Request page.
+
+[integration-tests]: https://doc.rust-lang.org/rust-by-example/testing/integration_testing.html
+[unit-tests]: https://doc.rust-lang.org/rust-by-example/testing/unit_testing.html
+[documentation-tests]: https://doc.rust-lang.org/rust-by-example/testing/doc_testing.html
+[conditional-compilation]: https://doc.rust-lang.org/reference/conditional-compilation.html

docs/contributing/reviewing-pull-requests.md

@@ -0,0 +1,80 @@
+## Reviewing Pull Requests
+
+**Any Tokio community member is welcome to review any pull request**.
+
+All Tokio contributors who choose to review and provide feedback on Pull
+Requests have a responsibility to both the project and the individual making the
+contribution. Reviews and feedback must be helpful, insightful, and geared
+towards improving the contribution as opposed to simply blocking it. If there
+are reasons why you feel the PR should not land, explain what those are. Do not
+expect to be able to block a Pull Request from advancing simply because you say
+"No" without giving an explanation. Be open to having your mind changed. Be open
+to working with the contributor to make the Pull Request better.
+
+Reviews that are dismissive or disrespectful of the contributor or any other
+reviewers are strictly counter to the Code of Conduct.
+
+When reviewing a Pull Request, the primary goals are for the codebase to improve
+and for the person submitting the request to succeed. **Even if a Pull Request
+does not land, the submitters should come away from the experience feeling like
+their effort was not wasted or unappreciated**. Every Pull Request from a new
+contributor is an opportunity to grow the community.
+
+### Review a bit at a time
+
+Do not overwhelm new contributors.
+
+It is tempting to micro-optimize and make everything about relative performance,
+perfect grammar, or exact style matches. Do not succumb to that temptation.
+
+Focus first on the most significant aspects of the change:
+
+1. Does this change make sense for Tokio?
+2. Does this change make Tokio better, even if only incrementally?
+3. Are there clear bugs or larger scale issues that need attending to?
+4. Is the commit message readable and correct? If it contains a breaking change
+   is it clear enough?
+
+Note that only **incremental** improvement is needed to land a PR. This means
+that the PR does not need to be perfect, only better than the status quo. Follow
+up PRs may be opened to continue iterating.
+
+When changes are necessary, *request* them, do not *demand* them, and **do not
+assume that the submitter already knows how to add a test or run a benchmark**.
+
+Specific performance optimization techniques, coding styles and conventions
+change over time. The first impression you give to a new contributor never does.
+
+Nits (requests for small changes that are not essential) are fine, but try to
+avoid stalling the Pull Request. Most nits can typically be fixed by the Tokio
+Collaborator landing the Pull Request but they can also be an opportunity for
+the contributor to learn a bit more about the project.
+
+It is always good to clearly indicate nits when you comment: e.g.
+`Nit: change foo() to bar(). But this is not blocking.`
+
+If your comments were addressed but were not folded automatically after new
+commits or if they proved to be mistaken, please, [hide them][hiding-a-comment]
+with the appropriate reason to keep the conversation flow concise and relevant.
+
+### Be aware of the person behind the code
+
+Be aware that *how* you communicate requests and reviews in your feedback can
+have a significant impact on the success of the Pull Request. Yes, we may land
+a particular change that makes Tokio better, but the individual might just not
+want to have anything to do with Tokio ever again. The goal is not just having
+good code.
+
+### Abandoned or Stalled Pull Requests
+
+If a Pull Request appears to be abandoned or stalled, it is polite to first
+check with the contributor to see if they intend to continue the work before
+checking if they would mind if you took it over (especially if it just has nits
+left). When doing so, it is courteous to give the original contributor credit
+for the work they started (either by preserving their name and email address in
+the commit log, or by using an `Author: ` meta-data tag in the commit.
+
+_Adapted from the [Node.js contributing guide][node]_.
+
+[node]: https://github.com/nodejs/node/blob/master/CONTRIBUTING.md
+[hiding-a-comment]: https://help.github.com/articles/managing-disruptive-comments/#hiding-a-comment

examples/Cargo.toml

@@ -24,6 +24,9 @@ httparse = "1.0"
 httpdate = "1.0"
 once_cell = "1.5.2"
 
+[target.'cfg(target_os = "linux")'.dev-dependencies]
+libc = "0.2"
+
 [target.'cfg(all(tokio_unstable, target_os = "linux"))'.dev-dependencies]
 tokio = { version = "1.0.0", path = "../tokio", features = ["full", "tracing", "taskdump"] }
 
@@ -46,6 +49,10 @@ path = "connect-udp.rs"
 name = "echo-tcp"
 path = "echo-tcp.rs"
 
+[[example]]
+name = "graceful-shutdown"
+path = "graceful-shutdown.rs"
+
 [[example]]
 name = "echo-udp"
 path = "echo-udp.rs"
@@ -102,5 +109,9 @@ path = "named-pipe-multi-client.rs"
 name = "dump"
 path = "dump.rs"
 
+[[example]]
+name = "prewarm-fd-table"
+path = "prewarm-fd-table.rs"
+
 [lints]
 workspace = true

examples/chat.rs

@@ -39,6 +39,8 @@ use std::io;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
+const DEFAULT_ADDR: &str = "127.0.0.1:6142";
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     use tracing_subscriber::{fmt::format::FmtSpan, EnvFilter};
@@ -70,14 +72,14 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let addr = env::args()
         .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:6142".to_string());
+        .unwrap_or_else(|| DEFAULT_ADDR.to_string());
 
     // Bind a TCP listener to the socket address.
     //
     // Note that this is the Tokio TcpListener, which is fully async.
     let listener = TcpListener::bind(&addr).await?;
 
-    tracing::info!("server running on {}", addr);
+    tracing::info!("server running on {addr}");
 
     loop {
         // Asynchronously wait for an inbound TcpStream.
@@ -88,9 +90,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
         // Spawn our handler to be run asynchronously.
         tokio::spawn(async move {
-            tracing::debug!("accepted connection");
+            tracing::debug!("accepted connection from {addr}");
             if let Err(e) = process(state, stream, addr).await {
-                tracing::info!("an error occurred; error = {:?}", e);
+                tracing::warn!("Connection from {addr} failed: {e:?}");
             }
         });
     }
@@ -138,12 +140,24 @@ impl Shared {
 
     /// Send a `LineCodec` encoded message to every peer, except
     /// for the sender.
+    ///
+    /// This function also cleans up disconnected peers automatically.
     async fn broadcast(&mut self, sender: SocketAddr, message: &str) {
-        for peer in self.peers.iter_mut() {
-            if *peer.0 != sender {
-                let _ = peer.1.send(message.into());
+        let mut failed_peers = Vec::new();
+        let message = message.to_string(); // Clone once for all sends
+
+        for (addr, tx) in self.peers.iter() {
+            if *addr != sender && tx.send(message.clone()).is_err() {
+                // Receiver has been dropped, mark for removal
+                failed_peers.push(*addr);
             }
         }
+
+        // Clean up disconnected peers
+        for addr in failed_peers {
+            self.peers.remove(&addr);
+            tracing::debug!("Removed disconnected peer: {addr}");
+        }
     }
 }
 
@@ -178,13 +192,10 @@ async fn process(
     lines.send("Please enter your username:").await?;
 
     // Read the first line from the `LineCodec` stream to get the username.
-    let username = match lines.next().await {
-        Some(Ok(line)) => line,
+    let Some(Ok(username)) = lines.next().await else {
         // We didn't get a line so we return early here.
-        _ => {
-            tracing::error!("Failed to get username from {}. Client disconnected.", addr);
-            return Ok(());
-        }
+        tracing::error!("Failed to get username from {addr}. Client disconnected.");
+        return Ok(());
     };
 
     // Register our peer with state which internally sets up some channels.
@@ -194,7 +205,7 @@ async fn process(
     {
         let mut state = state.lock().await;
         let msg = format!("{username} has joined the chat");
-        tracing::info!("{}", msg);
+        tracing::info!("{msg}");
         state.broadcast(addr, &msg).await;
     }
 
@@ -203,7 +214,10 @@ async fn process(
         tokio::select! {
             // A message was received from a peer. Send it to the current user.
             Some(msg) = peer.rx.recv() => {
-                peer.lines.send(&msg).await?;
+                if let Err(e) = peer.lines.send(&msg).await {
+                    tracing::error!("Failed to send message to {username}: {e:?}");
+                    break;
+                }
             }
             result = peer.lines.next() => match result {
                 // A message was received from the current user, we should
@@ -217,10 +231,9 @@ async fn process(
                 // An error occurred.
                 Some(Err(e)) => {
                     tracing::error!(
-                        "an error occurred while processing messages for {}; error = {:?}",
-                        username,
-                        e
+                        "an error occurred while processing messages for {username}; error = {e:?}"
                     );
+                    break;
                 }
                 // The stream has been exhausted.
                 None => break,
@@ -235,7 +248,7 @@ async fn process(
         state.peers.remove(&addr);
 
         let msg = format!("{username} has left the chat");
-        tracing::info!("{}", msg);
+        tracing::info!("{msg}");
         state.broadcast(addr, &msg).await;
     }
 

examples/connect-tcp.rs

@@ -58,14 +58,16 @@ pub async fn connect(
             //BytesMut into Bytes
             Ok(i) => future::ready(Some(i.freeze())),
             Err(e) => {
-                println!("failed to read from socket; error={e}");
+                eprintln!("failed to read from socket; error={e}");
                 future::ready(None)
             }
         })
         .map(Ok);
 
-    match future::join(sink.send_all(&mut stdin), stdout.send_all(&mut stream)).await {
-        (Err(e), _) | (_, Err(e)) => Err(e.into()),
-        _ => Ok(()),
+    tokio::select! {
+        r = sink.send_all(&mut stdin) => r?,
+        r = stdout.send_all(&mut stream) => r?,
     }
+
+    Ok(())
 }

examples/connect-udp.rs

@@ -59,7 +59,10 @@ pub async fn connect(
     let socket = UdpSocket::bind(&bind_addr).await?;
     socket.connect(addr).await?;
 
-    tokio::try_join!(send(stdin, &socket), recv(stdout, &socket))?;
+    tokio::select! {
+        r = send(stdin, &socket) => r?,
+        r = recv(stdout, &socket) => r?,
+    }
 
     Ok(())
 }

examples/echo-tcp.rs

@@ -27,6 +27,9 @@ use tokio::net::TcpListener;
 use std::env;
 use std::error::Error;
 
+const DEFAULT_ADDR: &str = "127.0.0.1:8080";
+const BUFFER_SIZE: usize = 4096;
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     // Allow passing an address to listen on as the first argument of this
@@ -34,7 +37,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     // 127.0.0.1:8080 for connections.
     let addr = env::args()
         .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:8080".to_string());
+        .unwrap_or_else(|| DEFAULT_ADDR.to_string());
 
     // Next up we create a TCP listener which will listen for incoming
     // connections. This TCP listener is bound to the address we determined
@@ -44,7 +47,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     loop {
         // Asynchronously wait for an inbound socket.
-        let (mut socket, _) = listener.accept().await?;
+        let (mut socket, addr) = listener.accept().await?;
 
         // And this is where much of the magic of this server happens. We
         // crucially want all clients to make progress concurrently, rather than
@@ -55,23 +58,27 @@ async fn main() -> Result<(), Box<dyn Error>> {
         // which will allow all of our clients to be processed concurrently.
 
         tokio::spawn(async move {
-            let mut buf = vec![0; 1024];
+            let mut buf = vec![0; BUFFER_SIZE];
 
             // In a loop, read data from the socket and write the data back.
             loop {
-                let n = socket
-                    .read(&mut buf)
-                    .await
-                    .expect("failed to read data from socket");
-
-                if n == 0 {
-                    return;
+                match socket.read(&mut buf).await {
+                    Ok(0) => {
+                        // Connection closed by peer
+                        return;
+                    }
+                    Ok(n) => {
+                        // Write the data back. If writing fails, log the error and exit.
+                        if let Err(e) = socket.write_all(&buf[0..n]).await {
+                            eprintln!("Failed to write to socket {}: {}", addr, e);
+                            return;
+                        }
+                    }
+                    Err(e) => {
+                        eprintln!("Failed to read from socket {}: {}", addr, e);
+                        return;
+                    }
                 }
-
-                socket
-                    .write_all(&buf[0..n])
-                    .await
-                    .expect("failed to write data to socket");
             }
         });
     }

examples/graceful-shutdown.rs

@@ -0,0 +1,120 @@
+//! Graceful shutdown example.
+//!
+//! This example follows the same approach described in the
+//! [Graceful Shutdown tutorial](https://tokio.rs/tokio/topics/shutdown):
+//!
+//! - A [`CancellationToken`] tells tasks to stop accepting new work.
+//! - A [`TaskTracker`] waits for in-flight work to complete.
+//!
+//! It runs a TCP echo server on `127.0.0.1:6142`. When Ctrl+C is
+//! pressed, the server stops accepting connections and waits for all
+//! active connections to finish before exiting.
+//!
+//! Start the server:
+//!
+//!     cargo run --example graceful-shutdown
+//!
+//! Then connect with:
+//!
+//!     nc 127.0.0.1 6142
+//!
+//! Press Ctrl+C on the server to trigger a graceful shutdown.
+
+#![warn(rust_2018_idioms)]
+
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::time::{self, Duration};
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+
+use std::error::Error;
+use std::net::SocketAddr;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let listener = TcpListener::bind("127.0.0.1:6142").await?;
+    println!("listening on 127.0.0.1:6142");
+
+    let token = CancellationToken::new();
+    let tracker = TaskTracker::new();
+
+    loop {
+        tokio::select! {
+            result = listener.accept() => {
+                let (socket, addr) = match result {
+                    Ok(conn) => conn,
+                    Err(e) => {
+                        // Transient errors (e.g. fd exhaustion) are recoverable,
+                        // so we log and continue. A production server might add a
+                        // backoff or break on fatal errors to avoid a busy loop.
+                        eprintln!("failed to accept: {e}");
+                        continue;
+                    }
+                };
+                println!("accepted connection from {addr}");
+
+                let token = token.clone();
+                tracker.spawn(handle_connection(socket, addr, token));
+            }
+
+            _ = tokio::signal::ctrl_c() => {
+                println!("\nshutdown signal received, waiting for connections to finish");
+                break;
+            }
+        }
+    }
+
+    // Signal all tasks to stop and wait for them to complete.
+    token.cancel();
+    tracker.close();
+    tracker.wait().await;
+
+    println!("shutdown complete");
+    Ok(())
+}
+
+async fn handle_connection(mut socket: TcpStream, addr: SocketAddr, token: CancellationToken) {
+    tokio::select! {
+        _ = echo(&mut socket) => {}
+        _ = token.cancelled() => {
+            notify_shutdown(&mut socket).await;
+        }
+    }
+
+    println!("connection from {addr} closed");
+}
+
+/// Reads lines from the client and writes them back.
+///
+/// Called for every accepted connection. Runs until the client disconnects
+/// or a read/write error occurs.
+async fn echo(socket: &mut TcpStream) {
+    let (reader, mut writer) = socket.split();
+    let mut reader = BufReader::new(reader);
+    let mut line = String::new();
+
+    loop {
+        match reader.read_line(&mut line).await {
+            Ok(0) | Err(_) => return,
+            Ok(_) => {
+                if writer.write_all(line.as_bytes()).await.is_err() {
+                    return;
+                }
+                line.clear();
+            }
+        }
+    }
+}
+
+/// Sends a shutdown notice to the client before closing the connection.
+///
+/// Called when the cancellation token fires. Uses a timeout so that a
+/// slow or unresponsive client cannot hold up the server shutdown.
+async fn notify_shutdown(socket: &mut TcpStream) {
+    let _ = time::timeout(
+        Duration::from_secs(1),
+        socket.write_all(b"server shutting down\n"),
+    )
+    .await;
+}

examples/prewarm-fd-table.rs

@@ -0,0 +1,86 @@
+//! Demonstrates pre-warming the Linux file descriptor table to avoid latency
+//! spikes caused by file descriptor table growth in multi-threaded processes.
+//!
+//! On Linux, the kernel's FD table is grown lazily and protected by RCU
+//! synchronization. In multi-threaded processes, when a syscall like `socket()`
+//! triggers a table resize, the calling thread blocks until all RCU readers
+//! quiesce. This can cause stalls of tens of milliseconds on tokio worker threads,
+//! blocking the entire event loop (not just one task).
+//!
+//! The workaround is to force the kernel to expand the FD table once per process
+//! (before any runtime starts), by duplicating an FD to a high slot and then
+//! closing it. The kernel never shrinks the FD table during a process's lifetime,
+//! so the capacity persists.
+//!
+//! This is most relevant for services that open many connections concurrently
+//! (e.g. HTTP servers, connection pools). The pre-warm target should be at least
+//! your expected peak FD count, and must not exceed `RLIMIT_NOFILE`.
+//!
+//! See: <https://github.com/tokio-rs/tokio/issues/7970>
+//!
+//! Usage:
+//!
+//!     cargo run --example prewarm-fd-table
+
+#![warn(rust_2018_idioms)]
+
+/// Pre-warms the FD table using `fcntl(F_DUPFD_CLOEXEC)` to duplicate an FD
+/// into a high slot, expanding the table in a single syscall. `F_DUPFD_CLOEXEC`
+/// allocates the lowest available FD >= `target`, so it never clobbers an
+/// existing FD.
+#[cfg(target_os = "linux")]
+fn prewarm_fd_table(target: i32) -> std::io::Result<()> {
+    use std::os::unix::io::{FromRawFd, OwnedFd};
+
+    let dev_null = std::fs::File::open("/dev/null")?;
+    let raw = unsafe {
+        libc::fcntl(
+            std::os::unix::io::AsRawFd::as_raw_fd(&dev_null),
+            libc::F_DUPFD_CLOEXEC,
+            target,
+        )
+    };
+    if raw < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+
+    // Close both FDs. The table capacity persists.
+    let _owned = unsafe { OwnedFd::from_raw_fd(raw) };
+    drop(dev_null);
+
+    Ok(())
+}
+
+/// Fully safe alternative using only stdlib. Requires O(n) syscalls instead of
+/// one, but avoids `unsafe` entirely.
+#[cfg(target_os = "linux")]
+#[allow(dead_code)]
+fn prewarm_fd_table_safe(target: i32) -> std::io::Result<()> {
+    let f = std::fs::File::open("/dev/null")?;
+    let _fds: Vec<_> = (0..target)
+        .map(|_| f.try_clone())
+        .collect::<Result<_, _>>()?;
+    Ok(())
+}
+
+fn main() {
+    #[cfg(target_os = "linux")]
+    {
+        const FD_TARGET: i32 = 10_000;
+
+        println!("Pre-warming FD table to {FD_TARGET} entries...");
+        if let Err(e) = prewarm_fd_table(FD_TARGET) {
+            eprintln!("Warning: failed to pre-warm FD table: {e}");
+        } else {
+            println!("FD table pre-warmed successfully.");
+        }
+    }
+
+    // Build the runtime *after* pre-warming.
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+
+    rt.block_on(async {});
+}

spellcheck.dic

@@ -1,4 +1,4 @@
-308
+314
 &
 +
 <
@@ -64,6 +64,7 @@ codec
 codecs
 combinator
 combinators
+condvar
 config
 Config
 connectionless
@@ -163,6 +164,7 @@ Lauck
 libc
 lifecycle
 lifo
+LLVM
 lookups
 macOS
 MacOS
@@ -193,6 +195,7 @@ ntasks
 NUMA
 ok
 oneshot
+opcode
 ORed
 os
 parker
@@ -206,6 +209,7 @@ POSIX
 proxied
 qos
 RAII
+RCU
 reallocations
 recv's
 refactors
@@ -306,4 +310,6 @@ wakers
 Wakers
 wakeup
 wakeups
+WASI
 workstealing
+ZST

tests-build/tests/fail/macros_invalid_input.rs

@@ -68,4 +68,7 @@ async fn test_has_second_test_attr_rust_2021() {}
 #[tokio::test]
 async fn test_has_generated_second_test_attr() {}
 
+#[tokio::test(name = 123)]
+async fn test_name_not_string() {}
+
 fn main() {}

tests-build/tests/fail/macros_invalid_input.stderr

@@ -4,7 +4,7 @@ error: the `async` keyword is missing from the function declaration
 6 | fn main_is_not_async() {}
   | ^^
 
-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`, `name`.
  --> tests/fail/macros_invalid_input.rs:8:15
   |
 8 | #[tokio::main(foo)]
@@ -22,13 +22,13 @@ error: the `async` keyword is missing from the function declaration
 15 | fn test_is_not_async() {}
    | ^^
 
-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`, `name`.
   --> tests/fail/macros_invalid_input.rs:17:15
    |
 17 | #[tokio::test(foo)]
    |               ^^^
 
-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`, `name`.
   --> tests/fail/macros_invalid_input.rs:20:15
    |
 20 | #[tokio::test(foo = 123)]
@@ -119,3 +119,9 @@ error: second test attribute is supplied, consider removing or changing the orde
    | ^^^^^^^^^^^^^^
    |
    = note: this error originates in the attribute macro `tokio::test` (in Nightly builds, run with -Z macro-backtrace for more info)
+
+error: Failed to parse value of `name` as string.
+  --> tests/fail/macros_invalid_input.rs:71:22
+   |
+71 | #[tokio::test(name = 123)]
+   |                      ^^^

tests-build/tests/fail/macros_join.rs

@@ -0,0 +1,42 @@
+use tests_build::tokio;
+
+#[tokio::main]
+async fn main() {
+    // do not leak `RotatorSelect`
+    let _ = tokio::join!(async {
+        fn foo(_: impl RotatorSelect) {}
+    });
+
+    // do not leak `std::task::Poll::Pending`
+    let _ = tokio::join!(async { Pending });
+
+    // do not leak `std::task::Poll::Ready`
+    let _ = tokio::join!(async { Ready(0) });
+
+    // do not leak `std::future::Future`
+    let _ = tokio::join!(async {
+        struct MyFuture;
+
+        impl Future for MyFuture {
+            type Output = ();
+
+            fn poll(
+                self: std::pin::Pin<&mut Self>,
+                _cx: &mut std::task::Context<'_>,
+            ) -> std::task::Poll<Self::Output> {
+                todo!()
+            }
+        }
+    });
+
+    // do not leak `std::pin::Pin`
+    let _ = tokio::join!(async {
+        let mut x = 5;
+        let _ = Pin::new(&mut x);
+    });
+
+    // do not leak `std::future::poll_fn`
+    let _ = tokio::join!(async {
+        let _ = poll_fn(|_cx| todo!());
+    });
+}

tests-build/tests/fail/macros_join.stderr

@@ -0,0 +1,60 @@
+error[E0405]: cannot find trait `RotatorSelect` in this scope
+ --> tests/fail/macros_join.rs:7:24
+  |
+7 |         fn foo(_: impl RotatorSelect) {}
+  |                        ^^^^^^^^^^^^^ not found in this scope
+
+error[E0425]: cannot find value `Pending` in this scope
+  --> tests/fail/macros_join.rs:11:34
+   |
+11 |     let _ = tokio::join!(async { Pending });
+   |                                  ^^^^^^^ not found in this scope
+   |
+help: consider importing this unit variant
+   |
+ 1 + use std::task::Poll::Pending;
+   |
+
+error[E0425]: cannot find function, tuple struct or tuple variant `Ready` in this scope
+  --> tests/fail/macros_join.rs:14:34
+   |
+14 |     let _ = tokio::join!(async { Ready(0) });
+   |                                  ^^^^^ not found in this scope
+   |
+help: consider importing this tuple variant
+   |
+ 1 + use std::task::Poll::Ready;
+   |
+
+error[E0405]: cannot find trait `Future` in this scope
+  --> tests/fail/macros_join.rs:20:14
+   |
+20 |         impl Future for MyFuture {
+   |              ^^^^^^ not found in this scope
+   |
+help: consider importing this trait
+   |
+ 1 + use std::future::Future;
+   |
+
+error[E0433]: failed to resolve: use of undeclared type `Pin`
+  --> tests/fail/macros_join.rs:35:17
+   |
+35 |         let _ = Pin::new(&mut x);
+   |                 ^^^ use of undeclared type `Pin`
+   |
+help: consider importing this struct
+   |
+ 1 + use std::pin::Pin;
+   |
+
+error[E0425]: cannot find function `poll_fn` in this scope
+  --> tests/fail/macros_join.rs:40:17
+   |
+40 |         let _ = poll_fn(|_cx| todo!());
+   |                 ^^^^^^^ not found in this scope
+   |
+help: consider importing this function
+   |
+ 1 + use std::future::poll_fn;
+   |

tests-build/tests/fail/macros_try_join.rs

@@ -0,0 +1,42 @@
+use tests_build::tokio;
+
+#[tokio::main]
+async fn main() {
+    // do not leak `RotatorSelect`
+    let _ = tokio::try_join!(async {
+        fn foo(_: impl RotatorSelect) {}
+    });
+
+    // do not leak `std::task::Poll::Pending`
+    let _ = tokio::try_join!(async { Pending });
+
+    // do not leak `std::task::Poll::Ready`
+    let _ = tokio::try_join!(async { Ready(0) });
+
+    // do not leak `std::future::Future`
+    let _ = tokio::try_join!(async {
+        struct MyFuture;
+
+        impl Future for MyFuture {
+            type Output = ();
+
+            fn poll(
+                self: std::pin::Pin<&mut Self>,
+                _cx: &mut std::task::Context<'_>,
+            ) -> std::task::Poll<Self::Output> {
+                todo!()
+            }
+        }
+    });
+
+    // do not leak `std::pin::Pin`
+    let _ = tokio::try_join!(async {
+        let mut x = 5;
+        let _ = Pin::new(&mut x);
+    });
+
+    // do not leak `std::future::poll_fn`
+    let _ = tokio::try_join!(async {
+        let _ = poll_fn(|_cx| todo!());
+    });
+}

tests-build/tests/fail/macros_try_join.stderr

@@ -0,0 +1,60 @@
+error[E0405]: cannot find trait `RotatorSelect` in this scope
+ --> tests/fail/macros_try_join.rs:7:24
+  |
+7 |         fn foo(_: impl RotatorSelect) {}
+  |                        ^^^^^^^^^^^^^ not found in this scope
+
+error[E0425]: cannot find value `Pending` in this scope
+  --> tests/fail/macros_try_join.rs:11:38
+   |
+11 |     let _ = tokio::try_join!(async { Pending });
+   |                                      ^^^^^^^ not found in this scope
+   |
+help: consider importing this unit variant
+   |
+ 1 + use std::task::Poll::Pending;
+   |
+
+error[E0425]: cannot find function, tuple struct or tuple variant `Ready` in this scope
+  --> tests/fail/macros_try_join.rs:14:38
+   |
+14 |     let _ = tokio::try_join!(async { Ready(0) });
+   |                                      ^^^^^ not found in this scope
+   |
+help: consider importing this tuple variant
+   |
+ 1 + use std::task::Poll::Ready;
+   |
+
+error[E0405]: cannot find trait `Future` in this scope
+  --> tests/fail/macros_try_join.rs:20:14
+   |
+20 |         impl Future for MyFuture {
+   |              ^^^^^^ not found in this scope
+   |
+help: consider importing this trait
+   |
+ 1 + use std::future::Future;
+   |
+
+error[E0433]: failed to resolve: use of undeclared type `Pin`
+  --> tests/fail/macros_try_join.rs:35:17
+   |
+35 |         let _ = Pin::new(&mut x);
+   |                 ^^^ use of undeclared type `Pin`
+   |
+help: consider importing this struct
+   |
+ 1 + use std::pin::Pin;
+   |
+
+error[E0425]: cannot find function `poll_fn` in this scope
+  --> tests/fail/macros_try_join.rs:40:17
+   |
+40 |         let _ = poll_fn(|_cx| todo!());
+   |                 ^^^^^^^ not found in this scope
+   |
+help: consider importing this function
+   |
+ 1 + use std::future::poll_fn;
+   |

tests-build/tests/fail/macros_type_mismatch.stderr

@@ -1,3 +1,14 @@
+error[E0271]: expected `{async block@$DIR/tests/fail/macros_type_mismatch.rs:3:1: 3:15}` to be a future that resolves to `()`, but it resolves to `Result<(), _>`
+ --> tests/fail/macros_type_mismatch.rs:3:1
+  |
+3 | #[tokio::main]
+  | ^^^^^^^^^^^^^^ expected `()`, found `Result<(), _>`
+  |
+  = note: expected unit type `()`
+                  found enum `Result<(), _>`
+  = note: required for the cast from `&{async block@$DIR/tests/fail/macros_type_mismatch.rs:3:1: 3:15}` to `&dyn Future<Output = ()>`
+  = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+
 error[E0308]: mismatched types
  --> tests/fail/macros_type_mismatch.rs:5:5
   |
@@ -15,6 +26,17 @@ help: consider using `Result::expect` to unwrap the `Result<(), _>` value, panic
 5 |     Ok(()).expect("REASON")
   |           +++++++++++++++++
 
+error[E0271]: expected `{async block@$DIR/tests/fail/macros_type_mismatch.rs:8:1: 8:15}` to be a future that resolves to `()`, but it resolves to `Result<(), _>`
+ --> tests/fail/macros_type_mismatch.rs:8:1
+  |
+8 | #[tokio::main]
+  | ^^^^^^^^^^^^^^ expected `()`, found `Result<(), _>`
+  |
+  = note: expected unit type `()`
+                  found enum `Result<(), _>`
+  = note: required for the cast from `&{async block@$DIR/tests/fail/macros_type_mismatch.rs:8:1: 8:15}` to `&dyn Future<Output = ()>`
+  = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+
 error[E0308]: mismatched types
   --> tests/fail/macros_type_mismatch.rs:10:5
    |
@@ -32,6 +54,17 @@ help: consider using `Result::expect` to unwrap the `Result<(), _>` value, panic
 10 |     return Ok(());.expect("REASON")
    |                   +++++++++++++++++
 
+error[E0271]: expected `{async block@$DIR/tests/fail/macros_type_mismatch.rs:13:1: 13:15}` to be a future that resolves to `Result<(), ()>`, but it resolves to `()`
+  --> tests/fail/macros_type_mismatch.rs:13:1
+   |
+13 | #[tokio::main]
+   | ^^^^^^^^^^^^^^ expected `Result<(), ()>`, found `()`
+   |
+   = note:   expected enum `Result<(), ()>`
+           found unit type `()`
+   = note: required for the cast from `&{async block@$DIR/tests/fail/macros_type_mismatch.rs:13:1: 13:15}` to `&dyn Future<Output = Result<(), ()>>`
+   = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+
 error[E0308]: mismatched types
   --> tests/fail/macros_type_mismatch.rs:23:5
    |
@@ -58,6 +91,17 @@ error[E0277]: the `?` operator can only be used in an async block that returns `
 40 |     None?;
    |         ^ cannot use the `?` operator in an async block that returns `()`
 
+error[E0271]: expected `{async block@$DIR/tests/fail/macros_type_mismatch.rs:38:1: 38:15}` to be a future that resolves to `Option<()>`, but it resolves to `()`
+  --> tests/fail/macros_type_mismatch.rs:38:1
+   |
+38 | #[tokio::main]
+   | ^^^^^^^^^^^^^^ expected `Option<()>`, found `()`
+   |
+   = note:   expected enum `Option<()>`
+           found unit type `()`
+   = note: required for the cast from `&{async block@$DIR/tests/fail/macros_type_mismatch.rs:38:1: 38:15}` to `&dyn Future<Output = Option<()>>`
+   = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+
 error[E0308]: mismatched types
   --> tests/fail/macros_type_mismatch.rs:40:5
    |
@@ -86,6 +130,17 @@ error[E0277]: the `?` operator can only be used in an async block that returns `
 57 |     Ok(())?;
    |           ^ cannot use the `?` operator in an async block that returns `()`
 
+error[E0271]: expected `{async block@$DIR/tests/fail/macros_type_mismatch.rs:55:1: 55:15}` to be a future that resolves to `Result<(), ()>`, but it resolves to `()`
+  --> tests/fail/macros_type_mismatch.rs:55:1
+   |
+55 | #[tokio::main]
+   | ^^^^^^^^^^^^^^ expected `Result<(), ()>`, found `()`
+   |
+   = note:   expected enum `Result<(), ()>`
+           found unit type `()`
+   = note: required for the cast from `&{async block@$DIR/tests/fail/macros_type_mismatch.rs:55:1: 55:15}` to `&dyn Future<Output = Result<(), ()>>`
+   = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+
 error[E0308]: mismatched types
   --> tests/fail/macros_type_mismatch.rs:57:5
    |
@@ -102,6 +157,15 @@ help: try adding an expression at the end of the block
 58 +     Ok(())
    |
 
+error[E0271]: expected `{async block@$DIR/tests/fail/macros_type_mismatch.rs:63:1: 63:15}` to be a future that resolves to `()`, but it resolves to `{integer}`
+  --> tests/fail/macros_type_mismatch.rs:63:1
+   |
+63 | #[tokio::main]
+   | ^^^^^^^^^^^^^^ expected `()`, found integer
+   |
+   = note: required for the cast from `&{async block@$DIR/tests/fail/macros_type_mismatch.rs:63:1: 63:15}` to `&dyn Future<Output = ()>`
+   = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+
 error[E0308]: mismatched types
   --> tests/fail/macros_type_mismatch.rs:66:5
    |

tests-build/tests/macros.rs

@@ -12,12 +12,24 @@ fn compile_fail_full() {
     #[cfg(feature = "full")]
     t.pass("tests/pass/macros_main_loop.rs");
 
+    #[cfg(feature = "full")]
+    t.pass("tests/pass/impl_trait.rs");
+
+    #[cfg(feature = "full")]
+    t.pass("tests/pass/use_builder_outer.rs");
+
     #[cfg(feature = "full")]
     t.compile_fail("tests/fail/macros_invalid_input.rs");
 
     #[cfg(feature = "full")]
     t.compile_fail("tests/fail/macros_dead_code.rs");
 
+    #[cfg(feature = "full")]
+    t.compile_fail("tests/fail/macros_join.rs");
+
+    #[cfg(feature = "full")]
+    t.compile_fail("tests/fail/macros_try_join.rs");
+
     #[cfg(feature = "full")]
     t.compile_fail("tests/fail/macros_type_mismatch.rs");
 

tests-build/tests/pass/impl_trait.rs

@@ -0,0 +1,23 @@
+use tests_build::tokio;
+
+#[tokio::main]
+async fn never() -> ! {
+    loop {}
+}
+
+#[tokio::main]
+async fn impl_trait() -> impl Iterator<Item = impl core::fmt::Debug> {
+    [()].into_iter()
+}
+
+#[tokio::main]
+async fn impl_trait2() -> Result<(), impl core::fmt::Debug> {
+    Err(())
+}
+
+fn main() {
+    if impl_trait().count() == 10 {
+        never();
+    }
+    let _ = impl_trait2();
+}

tests-build/tests/pass/use_builder_outer.rs

@@ -0,0 +1,9 @@
+#![deny(unused_qualifications)]
+
+use tests_build::tokio;
+pub use tokio::runtime;
+
+#[tokio::main]
+async fn main() {
+    if true {}
+}

tokio-macros/CHANGELOG.md

@@ -1,3 +1,19 @@
+# 2.7.0 (April 3rd, 2026)
+
+- macros: stabilize `LocalRuntime` ([#7557])
+- macros: add runtime name ([#7924])
+
+[#7557]: https://github.com/tokio-rs/tokio/pull/7557
+[#7924]: https://github.com/tokio-rs/tokio/pull/7924
+
+# 2.6.1 (Mar 2nd, 2026)
+
+- macros: improve error message for return type mismatch in #[tokio::main] ([#7856])
+- macros: use call_site hygiene to avoid unused qualification ([#7866])
+
+[#7856]: https://github.com/tokio-rs/tokio/pull/7856
+[#7866]: https://github.com/tokio-rs/tokio/pull/7866
+
 # 2.6.0 (Oct 14th, 2025)
 
 The MSRV is raised to 1.71.

tokio-macros/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "tokio-macros"
 # When releasing to crates.io:
-# - Remove path dependencies
+# - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-macros-x.y.z" git tag.
-version = "2.6.0"
+version = "2.7.0"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -27,7 +27,7 @@ quote = "1"
 syn = { version = "2.0", features = ["full"] }
 
 [dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
+tokio = { version = "1.0.0", path = "../tokio", features = ["full", "test-util"] }
 
 [package.metadata.docs.rs]
 all-features = true

tokio-macros/src/entry.rs

@@ -53,6 +53,7 @@ impl UnhandledPanic {
 }
 
 struct FinalConfig {
+    name: Option<String>,
     flavor: RuntimeFlavor,
     worker_threads: Option<usize>,
     start_paused: Option<bool>,
@@ -62,6 +63,7 @@ struct FinalConfig {
 
 /// Config used in case of the attribute not being able to build a valid config
 const DEFAULT_ERROR_CONFIG: FinalConfig = FinalConfig {
+    name: None,
     flavor: RuntimeFlavor::CurrentThread,
     worker_threads: None,
     start_paused: None,
@@ -70,6 +72,7 @@ const DEFAULT_ERROR_CONFIG: FinalConfig = FinalConfig {
 };
 
 struct Configuration {
+    name: Option<String>,
     rt_multi_thread_available: bool,
     default_flavor: RuntimeFlavor,
     flavor: Option<RuntimeFlavor>,
@@ -83,6 +86,7 @@ struct Configuration {
 impl Configuration {
     fn new(is_test: bool, rt_multi_thread: bool) -> Self {
         Configuration {
+            name: None,
             rt_multi_thread_available: rt_multi_thread,
             default_flavor: match is_test {
                 true => RuntimeFlavor::CurrentThread,
@@ -97,6 +101,16 @@ impl Configuration {
         }
     }
 
+    fn set_name(&mut self, name: syn::Lit, span: Span) -> Result<(), syn::Error> {
+        if self.name.is_some() {
+            return Err(syn::Error::new(span, "`name` set multiple times."));
+        }
+
+        let runtime_name = parse_string(name, span, "name")?;
+        self.name = Some(runtime_name);
+        Ok(())
+    }
+
     fn set_flavor(&mut self, runtime: syn::Lit, span: Span) -> Result<(), syn::Error> {
         if self.flavor.is_some() {
             return Err(syn::Error::new(span, "`flavor` set multiple times."));
@@ -227,6 +241,7 @@ impl Configuration {
         };
 
         Ok(FinalConfig {
+            name: self.name.clone(),
             crate_name: self.crate_name.clone(),
             flavor,
             worker_threads,
@@ -293,6 +308,35 @@ fn parse_bool(bool: syn::Lit, span: Span, field: &str) -> Result<bool, syn::Erro
     }
 }
 
+fn contains_impl_trait(ty: &syn::Type) -> bool {
+    match ty {
+        syn::Type::ImplTrait(_) => true,
+        syn::Type::Array(t) => contains_impl_trait(&t.elem),
+        syn::Type::Ptr(t) => contains_impl_trait(&t.elem),
+        syn::Type::Reference(t) => contains_impl_trait(&t.elem),
+        syn::Type::Slice(t) => contains_impl_trait(&t.elem),
+        syn::Type::Tuple(t) => t.elems.iter().any(contains_impl_trait),
+        syn::Type::Paren(t) => contains_impl_trait(&t.elem),
+        syn::Type::Group(t) => contains_impl_trait(&t.elem),
+        syn::Type::Path(t) => match t.path.segments.last() {
+            Some(segment) => match &segment.arguments {
+                syn::PathArguments::AngleBracketed(args) => args.args.iter().any(|arg| match arg {
+                    syn::GenericArgument::Type(t) => contains_impl_trait(t),
+                    syn::GenericArgument::AssocType(t) => contains_impl_trait(&t.ty),
+                    _ => false,
+                }),
+                syn::PathArguments::Parenthesized(args) => {
+                    args.inputs.iter().any(contains_impl_trait)
+                        || matches!(&args.output, syn::ReturnType::Type(_, t) if contains_impl_trait(t))
+                }
+                syn::PathArguments::None => false,
+            },
+            None => false,
+        },
+        _ => false,
+    }
+}
+
 fn build_config(
     input: &ItemFn,
     args: AttributeArgs,
@@ -343,9 +387,12 @@ fn build_config(
                         config
                             .set_unhandled_panic(lit.clone(), syn::spanned::Spanned::span(lit))?;
                     }
+                    "name" => {
+                        config.set_name(lit.clone(), syn::spanned::Spanned::span(lit))?;
+                    }
                     name => {
                         let msg = format!(
-                            "Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`",
+                            "Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`, `name`.",
                         );
                         return Err(syn::Error::new_spanned(namevalue, msg));
                     }
@@ -368,11 +415,12 @@ fn build_config(
                             "Set the runtime flavor with #[{macro_name}(flavor = \"current_thread\")]."
                         )
                     }
-                    "flavor" | "worker_threads" | "start_paused" | "crate" | "unhandled_panic" => {
+                    "flavor" | "worker_threads" | "start_paused" | "crate" | "unhandled_panic"
+                    | "name" => {
                         format!("The `{name}` attribute requires an argument.")
                     }
                     name => {
-                        format!("Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.")
+                        format!("Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`, `name`.")
                     }
                 };
                 return Err(syn::Error::new_spanned(path, msg));
@@ -408,25 +456,27 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
     let crate_path = config
         .crate_name
         .map(ToTokens::into_token_stream)
-        .unwrap_or_else(|| Ident::new("tokio", last_stmt_start_span).into_token_stream());
+        .unwrap_or_else(|| {
+            Ident::new("tokio", Span::call_site().located_at(last_stmt_start_span))
+                .into_token_stream()
+        });
+
+    let use_builder = quote_spanned! {Span::call_site().located_at(last_stmt_start_span)=>
+        use #crate_path::runtime::Builder;
+    };
 
     let mut rt = match config.flavor {
         RuntimeFlavor::CurrentThread | RuntimeFlavor::Local => {
             quote_spanned! {last_stmt_start_span=>
-                #crate_path::runtime::Builder::new_current_thread()
+                Builder::new_current_thread()
             }
         }
         RuntimeFlavor::Threaded => quote_spanned! {last_stmt_start_span=>
-            #crate_path::runtime::Builder::new_multi_thread()
+            Builder::new_multi_thread()
         },
     };
 
-    let mut checks = vec![];
-    let mut errors = vec![];
-
     let build = if let RuntimeFlavor::Local = config.flavor {
-        checks.push(quote! { tokio_unstable });
-        errors.push("The local runtime flavor is only available when `tokio_unstable` is set.");
         quote_spanned! {last_stmt_start_span=> build_local(Default::default())}
     } else {
         quote_spanned! {last_stmt_start_span=> build()}
@@ -442,6 +492,9 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
         let unhandled_panic = v.into_tokens(&crate_path);
         rt = quote_spanned! {last_stmt_start_span=> #rt.unhandled_panic(#unhandled_panic) };
     }
+    if let Some(v) = config.name {
+        rt = quote_spanned! {last_stmt_start_span=> #rt.name(#v) };
+    }
 
     let generated_attrs = if is_test {
         quote! {
@@ -451,25 +504,14 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
         quote! {}
     };
 
-    let do_checks: TokenStream = checks
-        .iter()
-        .zip(&errors)
-        .map(|(check, error)| {
-            quote! {
-                #[cfg(not(#check))]
-                compile_error!(#error);
-            }
-        })
-        .collect();
-
     let body_ident = quote! { body };
     // This explicit `return` is intentional. See tokio-rs/tokio#4636
     let last_block = quote_spanned! {last_stmt_end_span=>
-        #do_checks
 
-        #[cfg(all(#(#checks),*))]
         #[allow(clippy::expect_used, clippy::diverging_sub_expression, clippy::needless_return, clippy::unwrap_in_result)]
         {
+            #use_builder
+
             return #rt
                 .enable_all()
                 .#build
@@ -477,10 +519,6 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
                 .block_on(#body_ident);
         }
 
-        #[cfg(not(all(#(#checks),*)))]
-        {
-            panic!("fell through checks")
-        }
     };
 
     let body = input.body();
@@ -494,22 +532,42 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
     //
     // We don't do this for the main function as it should only be used once so
     // there will be no benefit.
+    let output_type = match &input.sig.output {
+        // For functions with no return value syn doesn't print anything,
+        // but that doesn't work as `Output` for our boxed `Future`, so
+        // default to `()` (the same type as the function output).
+        syn::ReturnType::Default => quote! { () },
+        syn::ReturnType::Type(_, ret_type) => quote! { #ret_type },
+    };
+
     let body = if is_test {
-        let output_type = match &input.sig.output {
-            // For functions with no return value syn doesn't print anything,
-            // but that doesn't work as `Output` for our boxed `Future`, so
-            // default to `()` (the same type as the function output).
-            syn::ReturnType::Default => quote! { () },
-            syn::ReturnType::Type(_, ret_type) => quote! { #ret_type },
-        };
         quote! {
             let body = async #body;
             #crate_path::pin!(body);
             let body: ::core::pin::Pin<&mut dyn ::core::future::Future<Output = #output_type>> = body;
         }
     } else {
+        // force typecheck without runtime overhead
+        let check_block = match &input.sig.output {
+            syn::ReturnType::Type(_, t)
+                if matches!(**t, syn::Type::Never(_)) || contains_impl_trait(t) =>
+            {
+                quote! {}
+            }
+            _ => quote! {
+                if false {
+                    let _: &dyn ::core::future::Future<Output = #output_type> = &body;
+                }
+            },
+        };
+
         quote! {
             let body = async #body;
+            // Compile-time assertion that the future's output matches the return type.
+            let body = {
+                #check_block
+                body
+            };
         }
     };
 

tokio-macros/src/lib.rs

@@ -12,11 +12,6 @@
 
 //! Macros for use with Tokio
 
-// This `extern` is required for older `rustc` versions but newer `rustc`
-// versions warn about the unused `extern crate`.
-#[allow(unused_extern_crates)]
-extern crate proc_macro;
-
 mod entry;
 mod select;
 
@@ -78,16 +73,11 @@ use proc_macro::TokenStream;
 ///
 /// ## Local
 ///
-/// [Unstable API][unstable] only.
-///
 /// To use the [local runtime], the macro can be configured using
 ///
 /// ```rust
-/// # #[cfg(tokio_unstable)]
 /// #[tokio::main(flavor = "local")]
 /// # async fn main() {}
-/// # #[cfg(not(tokio_unstable))]
-/// # fn main() {}
 /// ```
 ///
 /// # Function arguments
@@ -96,6 +86,30 @@ use proc_macro::TokenStream;
 ///
 /// # Usage
 ///
+/// ## Set the name of the runtime
+///
+/// ```rust
+/// #[tokio::main(name = "my-runtime")]
+/// async fn main() {
+///     println!("Hello world");
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::main]`
+///
+/// ```rust
+/// fn main() {
+///     tokio::runtime::Builder::new_multi_thread()
+///         .enable_all()
+///         .name("my-runtime")
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             println!("Hello world");
+///         })
+/// }
+/// ```
+///
 /// ## Using the multi-threaded runtime
 ///
 /// ```rust
@@ -146,25 +160,19 @@ use proc_macro::TokenStream;
 ///
 /// ## Using the local runtime
 ///
-/// Available in the [unstable API][unstable] only.
-///
 /// The [local runtime] is similar to the current-thread runtime but
 /// supports [`task::spawn_local`](../tokio/task/fn.spawn_local.html).
 ///
 /// ```rust
-/// # #[cfg(tokio_unstable)]
 /// #[tokio::main(flavor = "local")]
 /// async fn main() {
 ///     println!("Hello world");
 /// }
-/// # #[cfg(not(tokio_unstable))]
-/// # fn main() {}
 /// ```
 ///
 /// Equivalent code not using `#[tokio::main]`
 ///
 /// ```rust
-/// # #[cfg(tokio_unstable)]
 /// fn main() {
 ///     tokio::runtime::Builder::new_current_thread()
 ///         .enable_all()
@@ -174,8 +182,6 @@ use proc_macro::TokenStream;
 ///             println!("Hello world");
 ///         })
 /// }
-/// # #[cfg(not(tokio_unstable))]
-/// # fn main() {}
 /// ```
 ///
 ///
@@ -413,6 +419,31 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 ///
 /// ## Usage
 ///
+/// ### Set the name of the runtime
+///
+/// ```no_run
+/// #[tokio::test(name = "my-test-runtime")]
+/// async fn my_test() {
+///     assert!(true);
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::test]`
+///
+/// ```no_run
+/// #[test]
+/// fn my_test() {
+///     tokio::runtime::Builder::new_current_thread()
+///         .enable_all()
+///         .name("my-test-runtime")
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             assert!(true);
+///         })
+/// }
+/// ```
+///
 /// ### Using the multi-thread runtime
 ///
 /// ```no_run

tokio-stream/CHANGELOG.md

@@ -1,3 +1,25 @@
+# 0.1.18 (January 4th, 2026)
+
+### Added
+
+- stream: add `ChunksTimeout::into_remainder` ([#7715])
+- stream: add examples to wrapper types ([#7024])
+- sync: implement `Stream::size_hint` for `ReceiverStream` and `UnboundedReceiverStream` ([#7492])
+
+### Fixed
+
+- stream: work around the rustc bug in `StreamExt::collect` ([#7754])
+
+### Documented
+
+- stream: improve the the docs of `TcpListenerStream` ([#7578])
+
+[#7024]: https://github.com/tokio-rs/tokio/pull/7024
+[#7492]: https://github.com/tokio-rs/tokio/pull/7492
+[#7578]: https://github.com/tokio-rs/tokio/pull/7578
+[#7715]: https://github.com/tokio-rs/tokio/pull/7715
+[#7754]: https://github.com/tokio-rs/tokio/pull/7754
+
 # 0.1.17 (December 6th, 2024)
 
 - deps: fix dev-dependency on tokio-test ([#6931], [#7019])

tokio-stream/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "tokio-stream"
 # When releasing to crates.io:
-# - Remove path dependencies
+# - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.17"
+version = "0.1.18"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -38,14 +38,14 @@ signal = ["tokio/signal"]
 [dependencies]
 futures-core = { version = "0.3.0" }
 pin-project-lite = "0.2.11"
-tokio = { version = "1.15.0", path = "../tokio", features = ["sync"] }
-tokio-util = { version = "0.7.0", path = "../tokio-util", optional = true }
+tokio = { version = "1.38.0", features = ["sync"] }
+tokio-util = { version = "0.7.0", optional = true }
 
 [dev-dependencies]
-tokio = { version = "1.2.0", path = "../tokio", features = ["full", "test-util"] }
+tokio = { version = "1.38.0", features = ["full", "test-util"] }
 async-stream = "0.3"
 parking_lot = "0.12.0"
-tokio-test = { version = "0.4", path = "../tokio-test" }
+tokio-test = "0.4"
 futures = { version = "0.3", default-features = false }
 
 [package.metadata.docs.rs]

tokio-stream/src/stream_close.rs

@@ -83,7 +83,7 @@ where
     #[inline]
     fn size_hint(&self) -> (usize, Option<usize>) {
         if let Some(inner) = &self.inner {
-            // We always return +1 because when there's stream there's atleast one more item.
+            // We always return +1 because when there's a stream there's at least one more item.
             let (l, u) = inner.size_hint();
             (l.saturating_add(1), u.and_then(|u| u.checked_add(1)))
         } else {

tokio-stream/src/stream_ext.rs

@@ -916,7 +916,7 @@ pub trait StreamExt: Stream {
     /// assert_eq!(Err("no"), values);
     /// # }
     /// ```
-    fn collect<T>(self) -> Collect<Self, T>
+    fn collect<T>(self) -> Collect<Self, T, T::InternalCollection>
     where
         T: FromStream<Self::Item>,
         Self: Sized,

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -33,6 +33,12 @@ impl<S: Stream> ChunksTimeout<S> {
             cap: max_size,
         }
     }
+
+    /// Consumes the [`ChunksTimeout`] and then returns all buffered items.
+    pub fn into_remainder(mut self: Pin<&mut Self>) -> Vec<S::Item> {
+        let me = self.as_mut().project();
+        std::mem::take(me.items)
+    }
 }
 
 impl<S: Stream> Stream for ChunksTimeout<S> {

tokio-stream/src/stream_ext/collect.rs

@@ -1,25 +1,25 @@
 use crate::Stream;
 
 use core::future::Future;
-use core::marker::PhantomPinned;
+use core::marker::{PhantomData, PhantomPinned};
 use core::mem;
 use core::pin::Pin;
 use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
+use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet, LinkedList, VecDeque};
+use std::hash::Hash;
 
 // Do not export this struct until `FromStream` can be unsealed.
 pin_project! {
     /// Future returned by the [`collect`](super::StreamExt::collect) method.
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     #[derive(Debug)]
-    pub struct Collect<T, U>
-    where
-        T: Stream,
-        U: FromStream<T::Item>,
+    pub struct Collect<T, U, C>
     {
         #[pin]
         stream: T,
-        collection: U::InternalCollection,
+        collection: C,
+        _output: PhantomData<U>,
         // Make this future `!Unpin` for compatibility with async trait methods.
         #[pin]
         _pin: PhantomPinned,
@@ -38,24 +38,25 @@ pin_project! {
 /// enhancements to the Rust language.
 pub trait FromStream<T>: sealed::FromStreamPriv<T> {}
 
-impl<T, U> Collect<T, U>
+impl<T, U> Collect<T, U, U::InternalCollection>
 where
     T: Stream,
     U: FromStream<T::Item>,
 {
-    pub(super) fn new(stream: T) -> Collect<T, U> {
+    pub(super) fn new(stream: T) -> Collect<T, U, U::InternalCollection> {
         let (lower, upper) = stream.size_hint();
         let collection = U::initialize(sealed::Internal, lower, upper);
 
         Collect {
             stream,
             collection,
+            _output: PhantomData,
             _pin: PhantomPinned,
         }
     }
 }
 
-impl<T, U> Future for Collect<T, U>
+impl<T, U> Future for Collect<T, U, U::InternalCollection>
 where
     T: Stream,
     U: FromStream<T::Item>,
@@ -136,6 +137,139 @@ impl<T> sealed::FromStreamPriv<T> for Vec<T> {
     }
 }
 
+impl<T> FromStream<T> for VecDeque<T> {}
+
+impl<T> sealed::FromStreamPriv<T> for VecDeque<T> {
+    type InternalCollection = VecDeque<T>;
+
+    fn initialize(_: sealed::Internal, lower: usize, _upper: Option<usize>) -> VecDeque<T> {
+        VecDeque::with_capacity(lower)
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut VecDeque<T>, item: T) -> bool {
+        collection.push_back(item);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut VecDeque<T>) -> VecDeque<T> {
+        mem::take(collection)
+    }
+}
+
+impl<T> FromStream<T> for LinkedList<T> {}
+
+impl<T> sealed::FromStreamPriv<T> for LinkedList<T> {
+    type InternalCollection = LinkedList<T>;
+
+    fn initialize(_: sealed::Internal, _lower: usize, _upper: Option<usize>) -> LinkedList<T> {
+        LinkedList::new()
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut LinkedList<T>, item: T) -> bool {
+        collection.push_back(item);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut LinkedList<T>) -> LinkedList<T> {
+        mem::take(collection)
+    }
+}
+
+impl<T: Ord> FromStream<T> for BTreeSet<T> {}
+
+impl<T: Ord> sealed::FromStreamPriv<T> for BTreeSet<T> {
+    type InternalCollection = BTreeSet<T>;
+
+    fn initialize(_: sealed::Internal, _lower: usize, _upper: Option<usize>) -> BTreeSet<T> {
+        BTreeSet::new()
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut BTreeSet<T>, item: T) -> bool {
+        collection.insert(item);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut BTreeSet<T>) -> BTreeSet<T> {
+        mem::take(collection)
+    }
+}
+
+impl<K: Ord, V> FromStream<(K, V)> for BTreeMap<K, V> {}
+
+impl<K: Ord, V> sealed::FromStreamPriv<(K, V)> for BTreeMap<K, V> {
+    type InternalCollection = BTreeMap<K, V>;
+
+    fn initialize(_: sealed::Internal, _lower: usize, _upper: Option<usize>) -> BTreeMap<K, V> {
+        BTreeMap::new()
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut BTreeMap<K, V>, (key, value): (K, V)) -> bool {
+        collection.insert(key, value);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut BTreeMap<K, V>) -> BTreeMap<K, V> {
+        mem::take(collection)
+    }
+}
+
+impl<T: Eq + Hash> FromStream<T> for HashSet<T> {}
+
+impl<T: Eq + Hash> sealed::FromStreamPriv<T> for HashSet<T> {
+    type InternalCollection = HashSet<T>;
+
+    fn initialize(_: sealed::Internal, lower: usize, _upper: Option<usize>) -> HashSet<T> {
+        HashSet::with_capacity(lower)
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut HashSet<T>, item: T) -> bool {
+        collection.insert(item);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut HashSet<T>) -> HashSet<T> {
+        mem::take(collection)
+    }
+}
+
+impl<K: Eq + Hash, V> FromStream<(K, V)> for HashMap<K, V> {}
+
+impl<K: Eq + Hash, V> sealed::FromStreamPriv<(K, V)> for HashMap<K, V> {
+    type InternalCollection = HashMap<K, V>;
+
+    fn initialize(_: sealed::Internal, lower: usize, _upper: Option<usize>) -> HashMap<K, V> {
+        HashMap::with_capacity(lower)
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut HashMap<K, V>, (key, value): (K, V)) -> bool {
+        collection.insert(key, value);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut HashMap<K, V>) -> HashMap<K, V> {
+        mem::take(collection)
+    }
+}
+
+impl<T: Ord> FromStream<T> for BinaryHeap<T> {}
+
+impl<T: Ord> sealed::FromStreamPriv<T> for BinaryHeap<T> {
+    type InternalCollection = BinaryHeap<T>;
+
+    fn initialize(_: sealed::Internal, lower: usize, _upper: Option<usize>) -> BinaryHeap<T> {
+        BinaryHeap::with_capacity(lower)
+    }
+
+    fn extend(_: sealed::Internal, collection: &mut BinaryHeap<T>, item: T) -> bool {
+        collection.push(item);
+        true
+    }
+
+    fn finalize(_: sealed::Internal, collection: &mut BinaryHeap<T>) -> BinaryHeap<T> {
+        mem::take(collection)
+    }
+}
+
 impl<T> FromStream<T> for Box<[T]> {}
 
 impl<T> sealed::FromStreamPriv<T> for Box<[T]> {

tokio-stream/src/stream_map.rs

@@ -579,9 +579,11 @@ where
     /// # Cancel safety
     ///
     /// This method is cancel safe. If `next_many` is used as the event in a
-    /// [`tokio::select!`](tokio::select) statement and some other branch
-    /// completes first, it is guaranteed that no items were received on any of
-    /// the underlying streams.
+    /// [`tokio::select!`] statement and some other branch completes first,
+    /// it is guaranteed that no items were received on any of the underlying
+    /// streams.
+    ///
+    /// [`tokio::select!`]: https://docs.rs/tokio/latest/tokio/macro.select.html
     pub async fn next_many(&mut self, buffer: &mut Vec<(K, V::Item)>, limit: usize) -> usize {
         poll_fn(|cx| self.poll_next_many(cx, buffer, limit)).await
     }

tokio-stream/src/wrappers.rs

@@ -22,9 +22,9 @@ cfg_sync! {
 }
 
 cfg_signal! {
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     mod signal_unix;
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     pub use signal_unix::SignalStream;
 
     #[cfg(any(windows, docsrs))]
@@ -39,12 +39,14 @@ cfg_time! {
 }
 
 cfg_net! {
+    #[cfg(not(loom))]
     mod tcp_listener;
+    #[cfg(not(loom))]
     pub use tcp_listener::TcpListenerStream;
 
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     mod unix_listener;
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     pub use unix_listener::UnixListenerStream;
 }
 
@@ -57,6 +59,8 @@ cfg_io_util! {
 }
 
 cfg_fs! {
+    #[cfg(not(loom))]
     mod read_dir;
+    #[cfg(not(loom))]
     pub use read_dir::ReadDirStream;
 }

tokio-stream/src/wrappers/interval.rs

@@ -1,4 +1,5 @@
 use crate::Stream;
+use futures_core::stream::FusedStream;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::time::{Instant, Interval};
@@ -57,6 +58,12 @@ impl Stream for IntervalStream {
     }
 }
 
+impl FusedStream for IntervalStream {
+    fn is_terminated(&self) -> bool {
+        false
+    }
+}
+
 impl AsRef<Interval> for IntervalStream {
     fn as_ref(&self) -> &Interval {
         &self.inner

tokio-stream/tests/stream_chunks_timeout.rs

@@ -0,0 +1,30 @@
+#![warn(rust_2018_idioms)]
+
+use futures::FutureExt;
+use std::error::Error;
+use tokio::time;
+use tokio::time::Duration;
+use tokio_stream::{self as stream, StreamExt};
+use tokio_test::assert_pending;
+use tokio_test::task;
+
+#[tokio::test(start_paused = true)]
+async fn stream_chunks_remainder() -> Result<(), Box<dyn Error>> {
+    let stream1 =
+        stream::iter([5]).then(move |n| time::sleep(Duration::from_secs(1)).map(move |_| n));
+
+    let inner = stream::iter([1, 2, 3, 4]).chain(stream1);
+    tokio::pin!(inner);
+
+    let chunked = (&mut inner).chunks_timeout(10, Duration::from_millis(20));
+
+    let mut chunked = task::spawn(chunked);
+    assert_pending!(chunked.poll_next());
+
+    let remainder = chunked.enter(|_, stream| stream.into_remainder());
+
+    assert_eq!(remainder, vec![1, 2, 3, 4]);
+    time::advance(Duration::from_secs(2)).await;
+    assert_eq!(inner.next().await, Some(5));
+    Ok(())
+}

tokio-stream/tests/stream_collect.rs

@@ -1,3 +1,5 @@
+use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet, LinkedList, VecDeque};
+
 use tokio_stream::{self as stream, StreamExt};
 use tokio_test::{assert_pending, assert_ready, assert_ready_err, assert_ready_ok, task};
 
@@ -61,6 +63,155 @@ async fn collect_vec_items() {
     assert_eq!(vec![1, 2], coll);
 }
 
+#[tokio::test]
+async fn collect_vecdeque_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<VecDeque<i32>>());
+
+    assert_pending!(fut.poll());
+
+    let xs = [1, 2, 42, 3];
+    for x in xs {
+        tx.send(x).unwrap();
+        assert!(fut.is_woken());
+        assert_pending!(fut.poll());
+    }
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+
+    assert_eq!(coll, VecDeque::from(xs));
+    assert_eq!(coll.into_iter().collect::<Vec<_>>(), xs);
+}
+
+#[tokio::test]
+async fn collect_linkedlist_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<LinkedList<i32>>());
+
+    assert_pending!(fut.poll());
+
+    let xs = [1, 2, 42, 3];
+    for x in xs {
+        tx.send(x).unwrap();
+        assert!(fut.is_woken());
+        assert_pending!(fut.poll());
+    }
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+
+    assert_eq!(coll, LinkedList::from(xs));
+    assert_eq!(coll.into_iter().collect::<Vec<_>>(), xs);
+}
+
+#[tokio::test]
+async fn collect_btreeset_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<BTreeSet<i32>>());
+
+    assert_pending!(fut.poll());
+
+    tx.send(2).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    tx.send(1).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+    assert_eq!(BTreeSet::from([1, 2]), coll);
+}
+
+#[tokio::test]
+async fn collect_btreemap_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<BTreeMap<i32, i32>>());
+
+    assert_pending!(fut.poll());
+
+    tx.send((3, 4)).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    tx.send((1, 2)).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+    assert_eq!(BTreeMap::from([(1, 2), (3, 4)]), coll);
+}
+
+#[tokio::test]
+async fn collect_hashset_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<HashSet<i32>>());
+
+    assert_pending!(fut.poll());
+
+    tx.send(1).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    tx.send(2).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+    assert_eq!(HashSet::from([1, 2]), coll);
+}
+
+#[tokio::test]
+async fn collect_hashmap_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<HashMap<i32, i32>>());
+
+    assert_pending!(fut.poll());
+
+    tx.send((1, 2)).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    tx.send((3, 4)).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+    assert_eq!(HashMap::from([(1, 2), (3, 4)]), coll);
+}
+
+#[tokio::test]
+async fn collect_binaryheap_items() {
+    let (tx, rx) = mpsc::unbounded_channel_stream();
+    let mut fut = task::spawn(rx.collect::<BinaryHeap<i32>>());
+
+    assert_pending!(fut.poll());
+
+    tx.send(2).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    tx.send(1).unwrap();
+    assert!(fut.is_woken());
+    assert_pending!(fut.poll());
+
+    drop(tx);
+    assert!(fut.is_woken());
+    let coll = assert_ready!(fut.poll());
+    assert_eq!(vec![1, 2], coll.into_sorted_vec());
+}
+
 #[tokio::test]
 async fn collect_string_items() {
     let (tx, rx) = mpsc::unbounded_channel_stream();

tokio-stream/tests/stream_fuse.rs

@@ -48,3 +48,15 @@ async fn basic_usage() {
     assert_eq!(stream.next().await, None);
     assert_eq!(stream.size_hint(), (0, Some(0)));
 }
+
+#[tokio::test]
+#[cfg(feature = "time")]
+async fn interval_stream_is_never_terminated() {
+    use futures_core::stream::FusedStream;
+    use tokio_stream::wrappers::IntervalStream;
+
+    let interval = tokio::time::interval(std::time::Duration::from_millis(1));
+    let stream = IntervalStream::new(interval);
+
+    assert!(!stream.is_terminated());
+}

tokio-test/CHANGELOG.md

@@ -1,3 +1,21 @@
+# 0.4.5 (January 4th, 2026)
+
+### Added
+
+- test: add `io::Builder::name` for better panic messages ([#7212])
+
+### Fixed
+
+- test: make `Spawn` forward `size_hint` ([#6607])
+
+### Changed
+
+- test: remove unused `async-stream` and `bytes` dependencies ([#7214])
+
+[#6607]: https://github.com/tokio-rs/tokio/pull/6607
+[#7212]: https://github.com/tokio-rs/tokio/pull/7212
+[#7214]: https://github.com/tokio-rs/tokio/pull/7214
+
 # 0.4.4 (March 14, 2024)
 
 - task: mark `Spawn` as `#[must_use]` ([#6371])

tokio-test/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "tokio-test"
 # When releasing to crates.io:
-# - Remove path dependencies
+# - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-test-0.4.x" git tag.
-version = "0.4.4"
+version = "0.4.5"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -17,12 +17,12 @@ Testing utilities for Tokio- and futures-based code
 categories = ["asynchronous", "development-tools::testing"]
 
 [dependencies]
-tokio = { version = "1.2.0", path = "../tokio", features = ["rt", "sync", "time", "test-util"] }
-tokio-stream = { version = "0.1.1", path = "../tokio-stream" }
+tokio = { version = "1.2.0", features = ["rt", "sync", "time", "test-util"] }
+tokio-stream = "0.1.1"
 futures-core = "0.3.0"
 
 [dev-dependencies]
-tokio = { version = "1.2.0", path = "../tokio", features = ["full"] }
+tokio = { version = "1.2.0", features = ["full"] }
 futures-util = "0.3.0"
 
 [package.metadata.docs.rs]

tokio-test/src/io.rs

@@ -438,7 +438,30 @@ impl AsyncWrite for Mock {
                         let until = Instant::now() + rem;
                         self.inner.sleep = Some(Box::pin(time::sleep_until(until)));
                     } else {
-                        panic!("unexpected WouldBlock {}", self.pmsg());
+                        // A race condition (TOCTOU) can occur if the
+                        // timer expires between the `write()` call
+                        // and `remaining_wait()` due to preemption or other
+                        // delays. In this case, the `Wait` action is already popped by
+                        // `action()`, so we continue to the next one.
+                        //
+                        // Consider the following sequence:
+                        //
+                        // poll_write      Inner          action()
+                        //    |--write()--->|               |
+                        //    |             |--action()---->| (returns Wait)
+                        //    |<-WouldBlk---|               |
+                        //    |             |               |
+                        //    |      <--- TIMEOUT! --->     |
+                        //    |  (due to preemption, etc.)  |
+                        //    |             |               |
+                        //    |-rem_wait()->|               |
+                        //    |             |--action()---->| (time's up, pop Wait)
+                        //    |<--None------|               |
+                        //    |             |               |
+                        //    |---continue->| (process next action)
+                        //
+                        // See <https://github.com/tokio-rs/tokio/issues/7881>.
+                        continue;
                     }
                 }
                 Ok(0) => {

tokio-util/CHANGELOG.md

@@ -1,3 +1,58 @@
+# 0.7.18 (January 4th, 2026)
+
+### Added
+
+- io: add `tokio_util::io::simplex` ([#7565])
+
+### Changed
+
+- task: remove unnecessary trait bounds on the `Debug` implementation for `JoinQueue` and `AbortOnDropHandle` ([#7720])
+
+### Fixed
+
+- deps: bump `tokio` to `1.44.0` ([#7733])
+
+### Documented
+
+- io: document the default capacity of the `ReaderStream` ([#7147])
+- sync: fix a typo in the docs of `PollSender::is_closed` ([#7737])
+
+[#7147]: https://github.com/tokio-rs/tokio/pull/7147
+[#7565]: https://github.com/tokio-rs/tokio/pull/7565
+[#7720]: https://github.com/tokio-rs/tokio/pull/7720
+[#7733]: https://github.com/tokio-rs/tokio/pull/7733
+[#7737]: https://github.com/tokio-rs/tokio/pull/7737
+
+# 0.7.17 (November 2nd, 2025)
+
+The MSRV is increased to 1.71.
+
+### Added
+
+- codec: add `{FramedRead,FramedWrite}::into_parts()` ([#7566])
+- time: add `#[track_caller]` to `FutureExt::timeout` ([#7588])
+- task: add `tokio_util::task::JoinQueue` ([#7590])
+
+### Changed
+
+- codec: remove unnecessary trait bounds on all Framed constructors ([#7716])
+
+### Documented
+
+- time: clarify the cancellation safety of the `DelayQueue` ([#7564])
+- docs: fix some docs links ([#7654])
+- task: simplify the example of `TaskTracker` ([#7657])
+- task: clarify the behavior of several `spawn_local` methods ([#7669])
+
+[#7564]: https://github.com/tokio-rs/tokio/pull/7564
+[#7566]: https://github.com/tokio-rs/tokio/pull/7566
+[#7588]: https://github.com/tokio-rs/tokio/pull/7588
+[#7590]: https://github.com/tokio-rs/tokio/pull/7590
+[#7654]: https://github.com/tokio-rs/tokio/pull/7654
+[#7657]: https://github.com/tokio-rs/tokio/pull/7657
+[#7669]: https://github.com/tokio-rs/tokio/pull/7669
+[#7716]: https://github.com/tokio-rs/tokio/pull/7716
+
 # 0.7.16 (August 3rd, 2025)
 
 ### Added

tokio-util/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "tokio-util"
 # When releasing to crates.io:
-# - Remove path dependencies
+# - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.16"
+version = "0.7.18"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -35,7 +35,7 @@ join-map = ["rt", "hashbrown"]
 __docs_rs = ["futures-util"]
 
 [dependencies]
-tokio = { version = "1.28.0", path = "../tokio", features = ["sync"] }
+tokio = { version = "1.47.0", features = ["sync"] }
 bytes = "1.5.0"
 futures-core = "0.3.0"
 futures-sink = "0.3.0"
@@ -47,9 +47,9 @@ tracing = { version = "0.1.29", default-features = false, features = ["std"], op
 hashbrown = { version = "0.15.0", default-features = false, optional = true }
 
 [dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
-tokio-test = { version = "0.4.0", path = "../tokio-test" }
-tokio-stream = { version = "0.1", path = "../tokio-stream" }
+tokio = { version = "1.0.0", features = ["full"] }
+tokio-test = "0.4.0"
+tokio-stream = "0.1"
 
 async-stream = "0.3.0"
 futures = "0.3.0"
@@ -57,6 +57,9 @@ futures-test = "0.3.5"
 parking_lot = "0.12.0"
 tempfile = "3.1.0"
 
+[target.'cfg(loom)'.dev-dependencies]
+loom = { version = "0.7", features = ["futures", "checkpoint"] }
+
 [package.metadata.docs.rs]
 all-features = true
 # enable unstable features in the documentation

tokio-util/src/cfg.rs

@@ -60,6 +60,15 @@ macro_rules! cfg_rt {
     }
 }
 
+macro_rules! cfg_not_rt {
+    ($($item:item)*) => {
+        $(
+            #[cfg(not(feature = "rt"))]
+            $item
+        )*
+    }
+}
+
 macro_rules! cfg_time {
     ($($item:item)*) => {
         $(

tokio-util/src/codec/framed.rs

@@ -41,10 +41,7 @@ pin_project! {
     }
 }
 
-impl<T, U> Framed<T, U>
-where
-    T: AsyncRead + AsyncWrite,
-{
+impl<T, U> Framed<T, U> {
     /// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
     /// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data.
     ///
@@ -127,9 +124,7 @@ where
             },
         }
     }
-}
 
-impl<T, U> Framed<T, U> {
     /// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
     /// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data.
     ///

tokio-util/src/codec/framed_impl.rs

@@ -65,6 +65,7 @@ impl Default for WriteFrame {
 
 impl From<BytesMut> for ReadFrame {
     fn from(mut buffer: BytesMut) -> Self {
+        let is_readable = !buffer.is_empty();
         let size = buffer.capacity();
         if size < INITIAL_CAPACITY {
             buffer.reserve(INITIAL_CAPACITY - size);
@@ -72,7 +73,7 @@ impl From<BytesMut> for ReadFrame {
 
         Self {
             buffer,
-            is_readable: size > 0,
+            is_readable,
             eof: false,
             has_errored: false,
         }

tokio-util/src/codec/framed_read.rs

@@ -36,11 +36,7 @@ pin_project! {
 
 // ===== impl FramedRead =====
 
-impl<T, D> FramedRead<T, D>
-where
-    T: AsyncRead,
-    D: Decoder,
-{
+impl<T, D> FramedRead<T, D> {
     /// Creates a new `FramedRead` with the given `decoder`.
     pub fn new(inner: T, decoder: D) -> FramedRead<T, D> {
         FramedRead {
@@ -68,9 +64,7 @@ where
             },
         }
     }
-}
 
-impl<T, D> FramedRead<T, D> {
     /// Returns a reference to the underlying I/O stream wrapped by
     /// `FramedRead`.
     ///

tokio-util/src/codec/framed_write.rs

@@ -35,10 +35,7 @@ pin_project! {
     }
 }
 
-impl<T, E> FramedWrite<T, E>
-where
-    T: AsyncWrite,
-{
+impl<T, E> FramedWrite<T, E> {
     /// Creates a new `FramedWrite` with the given `encoder`.
     pub fn new(inner: T, encoder: E) -> FramedWrite<T, E> {
         FramedWrite {
@@ -64,9 +61,7 @@ where
             },
         }
     }
-}
 
-impl<T, E> FramedWrite<T, E> {
     /// Returns a reference to the underlying I/O stream wrapped by
     /// `FramedWrite`.
     ///

tokio-util/src/either.rs

@@ -212,7 +212,7 @@ where
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, not(loom)))]
 mod tests {
     use super::*;
     use tokio::io::{repeat, AsyncReadExt, Repeat};

tokio-util/src/io/mod.rs

@@ -14,8 +14,10 @@ mod copy_to_bytes;
 mod inspect;
 mod read_buf;
 mod reader_stream;
+pub mod simplex;
 mod sink_writer;
 mod stream_reader;
+mod write_all_vectored;
 
 cfg_io_util! {
     mod read_arc;
@@ -31,4 +33,5 @@ pub use self::read_buf::read_buf;
 pub use self::reader_stream::ReaderStream;
 pub use self::sink_writer::SinkWriter;
 pub use self::stream_reader::StreamReader;
+pub use self::write_all_vectored::{write_all_vectored, WriteAllVectored};
 pub use crate::util::{poll_read_buf, poll_write_buf};

tokio-util/src/io/read_buf.rs

@@ -1,8 +1,7 @@
 use bytes::BufMut;
-use std::future::Future;
+use std::future::poll_fn;
 use std::io;
 use std::pin::Pin;
-use std::task::{Context, Poll};
 use tokio::io::AsyncRead;
 
 /// Read data from an `AsyncRead` into an implementer of the [`BufMut`] trait.
@@ -46,20 +45,5 @@ where
     R: AsyncRead + Unpin,
     B: BufMut,
 {
-    return ReadBufFn(read, buf).await;
-
-    struct ReadBufFn<'a, R, B>(&'a mut R, &'a mut B);
-
-    impl<'a, R, B> Future for ReadBufFn<'a, R, B>
-    where
-        R: AsyncRead + Unpin,
-        B: BufMut,
-    {
-        type Output = io::Result<usize>;
-
-        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-            let this = &mut *self;
-            crate::util::poll_read_buf(Pin::new(this.0), cx, this.1)
-        }
-    }
+    poll_fn(|cx| crate::util::poll_read_buf(Pin::new(read), cx, buf)).await
 }

tokio-util/src/io/reader_stream.rs

@@ -58,6 +58,11 @@ impl<R: AsyncRead> ReaderStream<R> {
     /// Convert an [`AsyncRead`] into a [`Stream`] with item type
     /// `Result<Bytes, std::io::Error>`.
     ///
+    /// Currently, the default capacity 4096 bytes (4 KiB).
+    /// This capacity is not part of the semver contract
+    /// and may be tweaked in future releases without
+    /// requiring a major version bump.
+    ///
     /// [`AsyncRead`]: tokio::io::AsyncRead
     /// [`Stream`]: futures_core::Stream
     pub fn new(reader: R) -> Self {

tokio-util/src/io/simplex.rs

@@ -0,0 +1,360 @@
+//! Unidirectional byte-oriented channel.
+
+use crate::util::poll_proceed;
+
+use bytes::Buf;
+use bytes::BytesMut;
+use futures_core::ready;
+use std::io::Error as IoError;
+use std::io::ErrorKind as IoErrorKind;
+use std::io::IoSlice;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll, Waker};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+type IoResult<T> = Result<T, IoError>;
+
+const CLOSED_ERROR_MSG: &str = "simplex has been closed";
+
+#[derive(Debug)]
+struct Inner {
+    /// `poll_write` will return [`Poll::Pending`] if the backpressure boundary is reached
+    backpressure_boundary: usize,
+
+    /// either [`Sender`] or [`Receiver`] is closed
+    is_closed: bool,
+
+    /// Waker used to wake the [`Receiver`]
+    receiver_waker: Option<Waker>,
+
+    /// Waker used to wake the [`Sender`]
+    sender_waker: Option<Waker>,
+
+    /// Buffer used to read and write data
+    buf: BytesMut,
+}
+
+impl Inner {
+    fn with_capacity(capacity: usize) -> Self {
+        Self {
+            backpressure_boundary: capacity,
+            is_closed: false,
+            receiver_waker: None,
+            sender_waker: None,
+            buf: BytesMut::with_capacity(capacity),
+        }
+    }
+
+    fn register_receiver_waker(&mut self, waker: &Waker) -> Option<Waker> {
+        match self.receiver_waker.as_mut() {
+            Some(old) if old.will_wake(waker) => None,
+            _ => self.receiver_waker.replace(waker.clone()),
+        }
+    }
+
+    fn register_sender_waker(&mut self, waker: &Waker) -> Option<Waker> {
+        match self.sender_waker.as_mut() {
+            Some(old) if old.will_wake(waker) => None,
+            _ => self.sender_waker.replace(waker.clone()),
+        }
+    }
+
+    fn take_receiver_waker(&mut self) -> Option<Waker> {
+        self.receiver_waker.take()
+    }
+
+    fn take_sender_waker(&mut self) -> Option<Waker> {
+        self.sender_waker.take()
+    }
+
+    fn is_closed(&self) -> bool {
+        self.is_closed
+    }
+
+    fn close_receiver(&mut self) -> Option<Waker> {
+        self.is_closed = true;
+        self.take_sender_waker()
+    }
+
+    fn close_sender(&mut self) -> Option<Waker> {
+        self.is_closed = true;
+        self.take_receiver_waker()
+    }
+}
+
+/// Receiver of the simplex channel.
+///
+/// # Cancellation safety
+///
+/// The `Receiver` is cancel safe. If it is used as the event in a
+/// [`tokio::select!`] statement and some other branch completes
+/// first, it is guaranteed that no bytes were received on this
+/// channel.
+///
+/// You can still read the remaining data from the buffer
+/// even if the write half has been dropped.
+/// See [`Sender::poll_shutdown`] and [`Sender::drop`] for more details.
+///
+/// [`tokio::select!`]: https://docs.rs/tokio/latest/tokio/macro.select.html
+#[derive(Debug)]
+pub struct Receiver {
+    inner: Arc<Mutex<Inner>>,
+}
+
+impl Drop for Receiver {
+    /// This also wakes up the [`Sender`].
+    fn drop(&mut self) {
+        let maybe_waker = {
+            let mut inner = self.inner.lock().unwrap();
+            inner.close_receiver()
+        };
+
+        if let Some(waker) = maybe_waker {
+            waker.wake();
+        }
+    }
+}
+
+impl AsyncRead for Receiver {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<IoResult<()>> {
+        let coop = ready!(poll_proceed(cx));
+
+        let mut inner = self.inner.lock().unwrap();
+
+        let to_read = buf.remaining().min(inner.buf.remaining());
+        if to_read == 0 {
+            if inner.is_closed() || buf.remaining() == 0 {
+                return Poll::Ready(Ok(()));
+            }
+
+            let old_waker = inner.register_receiver_waker(cx.waker());
+            let maybe_waker = inner.take_sender_waker();
+
+            // unlock before waking up and dropping old waker
+            drop(inner);
+            drop(old_waker);
+            if let Some(waker) = maybe_waker {
+                waker.wake();
+            }
+            return Poll::Pending;
+        }
+
+        // this is to avoid starving other tasks
+        coop.made_progress();
+
+        buf.put_slice(&inner.buf[..to_read]);
+        inner.buf.advance(to_read);
+
+        let waker = inner.take_sender_waker();
+        drop(inner); // unlock before waking up
+        if let Some(waker) = waker {
+            waker.wake();
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
+
+/// Sender of the simplex channel.
+///
+/// # Cancellation safety
+///
+/// The `Sender` is cancel safe. If it is used as the event in a
+/// [`tokio::select!`] statement and some other branch completes
+/// first, it is guaranteed that no bytes were sent on this channel.
+///
+/// # Shutdown
+///
+/// See [`Sender::poll_shutdown`].
+///
+/// [`tokio::select!`]: https://docs.rs/tokio/latest/tokio/macro.select.html
+#[derive(Debug)]
+pub struct Sender {
+    inner: Arc<Mutex<Inner>>,
+}
+
+impl Drop for Sender {
+    /// This also wakes up the [`Receiver`].
+    fn drop(&mut self) {
+        let maybe_waker = {
+            let mut inner = self.inner.lock().unwrap();
+            inner.close_sender()
+        };
+
+        if let Some(waker) = maybe_waker {
+            waker.wake();
+        }
+    }
+}
+
+impl AsyncWrite for Sender {
+    /// # Errors
+    ///
+    /// This method will return [`IoErrorKind::BrokenPipe`]
+    /// if the channel has been closed.
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<IoResult<usize>> {
+        let coop = ready!(poll_proceed(cx));
+
+        let mut inner = self.inner.lock().unwrap();
+
+        if inner.is_closed() {
+            return Poll::Ready(Err(IoError::new(IoErrorKind::BrokenPipe, CLOSED_ERROR_MSG)));
+        }
+
+        let free = inner
+            .backpressure_boundary
+            .checked_sub(inner.buf.len())
+            .expect("backpressure boundary overflow");
+        let to_write = buf.len().min(free);
+        if to_write == 0 {
+            if buf.is_empty() {
+                return Poll::Ready(Ok(0));
+            }
+
+            let old_waker = inner.register_sender_waker(cx.waker());
+            let waker = inner.take_receiver_waker();
+
+            // unlock before waking up and dropping old waker
+            drop(inner);
+            drop(old_waker);
+            if let Some(waker) = waker {
+                waker.wake();
+            }
+
+            return Poll::Pending;
+        }
+
+        // this is to avoid starving other tasks
+        coop.made_progress();
+
+        inner.buf.extend_from_slice(&buf[..to_write]);
+
+        let waker = inner.take_receiver_waker();
+        drop(inner); // unlock before waking up
+        if let Some(waker) = waker {
+            waker.wake();
+        }
+
+        Poll::Ready(Ok(to_write))
+    }
+
+    /// # Errors
+    ///
+    /// This method will return [`IoErrorKind::BrokenPipe`]
+    /// if the channel has been closed.
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<IoResult<()>> {
+        let inner = self.inner.lock().unwrap();
+        if inner.is_closed() {
+            Poll::Ready(Err(IoError::new(IoErrorKind::BrokenPipe, CLOSED_ERROR_MSG)))
+        } else {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    /// After returns [`Poll::Ready`], all the following call to
+    /// [`Sender::poll_write`] and [`Sender::poll_flush`]
+    /// will return error.
+    ///
+    /// The [`Receiver`] can still be used to read remaining data
+    /// until all bytes have been consumed.
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<IoResult<()>> {
+        let maybe_waker = {
+            let mut inner = self.inner.lock().unwrap();
+            inner.close_sender()
+        };
+
+        if let Some(waker) = maybe_waker {
+            waker.wake();
+        }
+
+        Poll::Ready(Ok(()))
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        true
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<Result<usize, IoError>> {
+        let coop = ready!(poll_proceed(cx));
+
+        let mut inner = self.inner.lock().unwrap();
+        if inner.is_closed() {
+            return Poll::Ready(Err(IoError::new(IoErrorKind::BrokenPipe, CLOSED_ERROR_MSG)));
+        }
+
+        let free = inner
+            .backpressure_boundary
+            .checked_sub(inner.buf.len())
+            .expect("backpressure boundary overflow");
+        if free == 0 {
+            let old_waker = inner.register_sender_waker(cx.waker());
+            let maybe_waker = inner.take_receiver_waker();
+
+            // unlock before waking up and dropping old waker
+            drop(inner);
+            drop(old_waker);
+            if let Some(waker) = maybe_waker {
+                waker.wake();
+            }
+
+            return Poll::Pending;
+        }
+
+        // this is to avoid starving other tasks
+        coop.made_progress();
+
+        let mut rem = free;
+        for buf in bufs {
+            if rem == 0 {
+                break;
+            }
+
+            let to_write = buf.len().min(rem);
+            if to_write == 0 {
+                assert_ne!(rem, 0);
+                assert_eq!(buf.len(), 0);
+                continue;
+            }
+
+            inner.buf.extend_from_slice(&buf[..to_write]);
+            rem -= to_write;
+        }
+
+        let waker = inner.take_receiver_waker();
+        drop(inner); // unlock before waking up
+        if let Some(waker) = waker {
+            waker.wake();
+        }
+
+        Poll::Ready(Ok(free - rem))
+    }
+}
+
+/// Create a simplex channel.
+///
+/// The `capacity` parameter specifies the maximum number of bytes that can be
+/// stored in the channel without making the [`Sender::poll_write`]
+/// return [`Poll::Pending`].
+///
+/// # Panics
+///
+/// This function will panic if `capacity` is zero.
+pub fn new(capacity: usize) -> (Sender, Receiver) {
+    assert_ne!(capacity, 0, "capacity must be greater than zero");
+
+    let inner = Arc::new(Mutex::new(Inner::with_capacity(capacity)));
+    let tx = Sender {
+        inner: Arc::clone(&inner),
+    };
+    let rx = Receiver { inner };
+    (tx, rx)
+}

tokio-util/src/io/write_all_vectored.rs

@@ -0,0 +1,165 @@
+use tokio::io::AsyncWrite;
+
+use pin_project_lite::pin_project;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{ready, Context, Poll};
+use std::{future::Future, io::IoSlice};
+use std::{io, mem};
+
+pin_project! {
+    /// A future that writes all data from multiple buffers to a writer.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct WriteAllVectored<'a, 'b, W: ?Sized> {
+        writer: &'a mut W,
+        bufs: &'a mut [IoSlice<'b>],
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+/// Like [`write_all`] but writes all data from multiple buffers into this writer.
+///
+/// This function writes multiple (possibly non-contiguous) buffers into the writer,
+/// using the `writev` syscall to potentially write in a single system call.
+///
+/// Equivalent to:
+///
+/// ```ignore
+/// async fn write_all_vectored<W: AsyncWrite + Unpin + ?Sized>(
+///     writer: &mut W,
+///     mut bufs: &mut [IoSlice<'_>]
+/// ) -> io::Result<()> {
+///     while !bufs.is_empty() {
+///         let n = write_vectored(writer, bufs).await?;
+///         if n == 0 {
+///             return Err(io::ErrorKind::WriteZero.into());
+///         }
+///         IoSlice::advance_slices(&mut bufs, n);
+///     }
+///     Ok(())
+/// }
+/// ```
+///
+/// # Cancel safety
+///
+/// This method is not cancellation safe. If it is used as the event
+/// in a `tokio::select!` statement and some other
+/// branch completes first, then the provided buffer may have been
+/// partially written, but future calls to `write_all_vectored` will
+/// have lost its place in the buffer.
+///
+/// # Examples
+///
+/// ```rust
+/// use tokio_util::io::write_all_vectored;
+/// use std::io::IoSlice;
+///
+/// #[tokio::main(flavor = "current_thread")]
+/// async fn main() -> std::io::Result<()> {
+///
+///     let mut writer = Vec::new();
+///     let bufs = &mut [
+///         IoSlice::new(&[1]),
+///         IoSlice::new(&[2, 3]),
+///         IoSlice::new(&[4, 5, 6]),
+///     ];
+///
+///     write_all_vectored(&mut writer, bufs).await?;
+///
+///     // Note: `bufs` has been modified by `IoSlice::advance_slices` and should not be reused.
+///     assert_eq!(writer, &[1, 2, 3, 4, 5, 6]);
+///     Ok(())
+/// }
+/// ```
+///
+/// # Notes
+///
+/// See the documentation for [`Write::write_all_vectored`] from std.
+/// After calling this function, the buffer slices may have
+/// been advanced and should not be reused.
+///
+/// [`Write::write_all_vectored`]: std::io::Write::write_all_vectored
+/// [`write_all`]: tokio::io::AsyncWriteExt::write_all
+/// [`writev`]: https://man7.org/linux/man-pages/man3/writev.3p.html
+pub fn write_all_vectored<'a, 'b, W>(
+    writer: &'a mut W,
+    bufs: &'a mut [IoSlice<'b>],
+) -> WriteAllVectored<'a, 'b, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    WriteAllVectored {
+        writer,
+        bufs,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<W> Future for WriteAllVectored<'_, '_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let me = self.project();
+        while !me.bufs.is_empty() {
+            // advance to first non-empty buffer
+            let non_empty = match me.bufs.iter().position(|b| !b.is_empty()) {
+                Some(pos) => pos,
+                None => return Poll::Ready(Ok(())),
+            };
+
+            // drop empty buffers at the start
+            *me.bufs = &mut mem::take(me.bufs)[non_empty..];
+
+            let n = ready!(Pin::new(&mut *me.writer).poll_write_vectored(cx, me.bufs))?;
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+            self::advance_slices(me.bufs, n);
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
+
+// copied from `std::IoSlice::advance_slices`
+// replace with method when MSRV is 1.81.0
+fn advance_slices<'a>(bufs: &mut &mut [IoSlice<'a>], n: usize) {
+    // Number of buffers to remove.
+    let mut remove = 0;
+    // Remaining length before reaching n. This prevents overflow
+    // that could happen if the length of slices in `bufs` were instead
+    // accumulated. Those slice may be aliased and, if they are large
+    // enough, their added length may overflow a `usize`.
+    let mut left = n;
+    for buf in bufs.iter() {
+        if let Some(remainder) = left.checked_sub(buf.len()) {
+            left = remainder;
+            remove += 1;
+        } else {
+            break;
+        }
+    }
+
+    *bufs = &mut std::mem::take(bufs)[remove..];
+    if let Some(first) = bufs.first_mut() {
+        let buf = &first[..left];
+        // necessary due to limitating in the borrow checker,
+        // when tokio MSRV reaches 1.81.0 this entire function
+        // can be replaced with `IoSlice::advance_slices`
+        //
+        // SAFETY: transmute a sub-slice of an IoSlice<'a> back to
+        // the lifetime `'a`. This is safe because the underlying memory
+        // is guaranteed to live for 'a, we have shared access, and no
+        // underlying data is reinterpreted to a different type.
+        unsafe {
+            *first = IoSlice::new(std::mem::transmute::<&[u8], &'a [u8]>(buf));
+        }
+    } else {
+        assert!(left == 0, "advancing io slices beyond their length");
+    }
+}

tokio-util/src/loom.rs

@@ -1 +1,9 @@
-pub(crate) use std::sync;
+//! This module abstracts over `loom` and `std::sync` types depending on whether we
+//! are running loom tests or not.
+
+pub(crate) mod sync {
+    #[cfg(all(test, loom))]
+    pub(crate) use loom::sync::{Arc, Mutex, MutexGuard};
+    #[cfg(not(all(test, loom)))]
+    pub(crate) use std::sync::{Arc, Mutex, MutexGuard};
+}

tokio-util/src/net/mod.rs

@@ -1,3 +1,5 @@
+#![cfg(not(loom))]
+
 //! TCP/UDP/Unix helpers for tokio.
 
 use crate::either::Either;

tokio-util/src/sync/mod.rs

@@ -15,3 +15,6 @@ pub use poll_semaphore::PollSemaphore;
 
 mod reusable_box;
 pub use reusable_box::ReusableBoxFuture;
+
+#[cfg(test)]
+mod tests;

tokio-util/src/sync/mpsc.rs

@@ -200,7 +200,7 @@ impl<T: Send> PollSender<T> {
         result
     }
 
-    /// Checks whether this sender is been closed.
+    /// Checks whether this sender is closed.
     ///
     /// The underlying channel that this sender was wrapping may still be open.
     pub fn is_closed(&self) -> bool {

tokio-util/src/sync/tests/loom_cancellation_token.rs

@@ -100,6 +100,9 @@ fn drop_token_no_child() {
     });
 }
 
+// Temporarily disabled due to a false positive in loom -
+// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
+#[ignore]
 #[test]
 fn drop_token_with_children() {
     loom::model(|| {
@@ -125,6 +128,9 @@ fn drop_token_with_children() {
     });
 }
 
+// Temporarily disabled due to a false positive in loom -
+// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
+#[ignore]
 #[test]
 fn drop_and_cancel_token() {
     loom::model(|| {
@@ -150,6 +156,9 @@ fn drop_and_cancel_token() {
     });
 }
 
+// Temporarily disabled due to a false positive in loom -
+// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
+#[ignore]
 #[test]
 fn cancel_parent_and_child() {
     loom::model(|| {

tokio-util/src/sync/tests/mod.rs

@@ -1 +1,2 @@
-
+#[cfg(loom)]
+mod loom_cancellation_token;

tokio-util/src/task/abort_on_drop.rs

@@ -1,5 +1,8 @@
 //! An [`AbortOnDropHandle`] is like a [`JoinHandle`], except that it
 //! will abort the task as soon as it is dropped.
+//!
+//! Correspondingly, an [`AbortOnDrop`] is like a [`AbortHandle`] that will abort
+//! the task as soon as it is dropped.
 
 use tokio::task::{AbortHandle, JoinError, JoinHandle};
 
@@ -15,12 +18,11 @@ use std::{
 ///
 /// [aborts]: tokio::task::JoinHandle::abort
 #[must_use = "Dropping the handle aborts the task immediately"]
-#[derive(Debug)]
 pub struct AbortOnDropHandle<T>(JoinHandle<T>);
 
 impl<T> Drop for AbortOnDropHandle<T> {
     fn drop(&mut self) {
-        self.0.abort()
+        self.abort()
     }
 }
 
@@ -32,12 +34,14 @@ impl<T> AbortOnDropHandle<T> {
 
     /// Abort the task associated with this handle,
     /// equivalent to [`JoinHandle::abort`].
+    #[inline]
     pub fn abort(&self) {
         self.0.abort()
     }
 
     /// Checks if the task associated with this handle is finished,
     /// equivalent to [`JoinHandle::is_finished`].
+    #[inline]
     pub fn is_finished(&self) -> bool {
         self.0.is_finished()
     }
@@ -58,6 +62,14 @@ impl<T> AbortOnDropHandle<T> {
     }
 }
 
+impl<T> std::fmt::Debug for AbortOnDropHandle<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AbortOnDropHandle")
+            .field("id", &self.0.id())
+            .finish()
+    }
+}
+
 impl<T> Future for AbortOnDropHandle<T> {
     type Output = Result<T, JoinError>;
 
@@ -71,3 +83,75 @@ impl<T> AsRef<JoinHandle<T>> for AbortOnDropHandle<T> {
         &self.0
     }
 }
+
+/// A wrapper around a [`tokio::task::AbortHandle`],
+/// which [aborts] the task when it is dropped.
+///
+/// Unlike [`AbortOnDropHandle`], [`AbortOnDrop`] cannot be `.await`ed for a result.
+///
+/// It has no generic parameter, making it suitable when you only need to keep
+/// a task handle in a struct and do not care about the output.
+///
+/// [aborts]: tokio::task::AbortHandle::abort
+#[must_use = "Dropping the handle aborts the task immediately"]
+pub struct AbortOnDrop(AbortHandle);
+
+impl Drop for AbortOnDrop {
+    fn drop(&mut self) {
+        self.abort()
+    }
+}
+
+impl AbortOnDrop {
+    /// Create an [`AbortOnDrop`] from a [`AbortHandle`].
+    pub fn new(handle: AbortHandle) -> Self {
+        Self(handle)
+    }
+
+    /// Abort the task associated with this handle,
+    /// equivalent to [`AbortHandle::abort`].
+    #[inline]
+    pub fn abort(&self) {
+        self.0.abort()
+    }
+
+    /// Checks if the task associated with this handle is finished,
+    /// equivalent to [`AbortHandle::is_finished`].
+    #[inline]
+    pub fn is_finished(&self) -> bool {
+        self.0.is_finished()
+    }
+
+    /// Cancels aborting on drop and returns the original [`AbortHandle`].
+    pub fn detach(self) -> AbortHandle {
+        // Avoid invoking `AbortOnDrop`'s `Drop` impl
+        let this = ManuallyDrop::new(self);
+        // SAFETY: `&this.0` is a reference, so it is certainly initialized, and
+        // it won't be double-dropped because it's in a `ManuallyDrop`
+        unsafe { std::ptr::read(&this.0) }
+    }
+}
+
+impl std::fmt::Debug for AbortOnDrop {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AbortOnDrop")
+            .field("id", &self.0.id())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A simple type that does not implement [`std::fmt::Debug`].
+    struct NotDebug;
+
+    fn is_debug<T: std::fmt::Debug>() {}
+
+    #[test]
+    fn assert_debug() {
+        is_debug::<AbortOnDrop>();
+        is_debug::<AbortOnDropHandle<NotDebug>>();
+    }
+}

tokio-util/src/task/join_map.rs

@@ -451,7 +451,7 @@ where
     ///    that panicked or was aborted.
     ///  * `None` if the `JoinMap` is empty.
     ///
-    /// [`tokio::select!`]: tokio::select
+    /// [`tokio::select!`]: https://docs.rs/tokio/latest/tokio/macro.select.html
     pub async fn join_next(&mut self) -> Option<(K, Result<V, JoinError>)> {
         loop {
             let (res, id) = match self.tasks.join_next_with_id().await {

tokio-util/src/task/join_queue.rs

@@ -21,7 +21,6 @@ use tokio::{
 ///
 /// When the [`JoinQueue`] is dropped, all tasks in the [`JoinQueue`] are
 /// immediately aborted.
-#[derive(Debug)]
 pub struct JoinQueue<T>(VecDeque<AbortOnDropHandle<T>>);
 
 impl<T> JoinQueue<T> {
@@ -379,6 +378,14 @@ impl<T> JoinQueue<T> {
     }
 }
 
+impl<T> std::fmt::Debug for JoinQueue<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_list()
+            .entries(self.0.iter().map(|jh| JoinHandle::id(jh.as_ref())))
+            .finish()
+    }
+}
+
 impl<T> Default for JoinQueue<T> {
     fn default() -> Self {
         Self::new()
@@ -401,3 +408,18 @@ where
         set
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A simple type that does not implement [`std::fmt::Debug`].
+    struct NotDebug;
+
+    fn is_debug<T: std::fmt::Debug>() {}
+
+    #[test]
+    fn assert_debug() {
+        is_debug::<JoinQueue<NotDebug>>();
+    }
+}

tokio-util/src/task/mod.rs

@@ -12,7 +12,7 @@ cfg_rt! {
     pub use task_tracker::TaskTracker;
 
     mod abort_on_drop;
-    pub use abort_on_drop::AbortOnDropHandle;
+    pub use abort_on_drop::{AbortOnDrop, AbortOnDropHandle};
 
     mod join_queue;
     pub use join_queue::JoinQueue;

tokio-util/src/time/wheel/mod.rs

@@ -10,15 +10,17 @@ use std::fmt::Debug;
 
 /// Timing wheel implementation.
 ///
-/// This type provides the hashed timing wheel implementation that backs `Timer`
-/// and `DelayQueue`.
+/// This type provides the hashed timing wheel implementation that backs
+/// [`DelayQueue`].
 ///
 /// The structure is generic over `T: Stack`. This allows handling timeout data
 /// being stored on the heap or in a slab. In order to support the latter case,
 /// the slab must be passed into each function allowing the implementation to
 /// lookup timer entries.
 ///
-/// See `Timer` documentation for some implementation notes.
+/// See `Driver` documentation for some implementation notes.
+///
+/// [`DelayQueue`]: crate::time::DelayQueue
 #[derive(Debug)]
 pub(crate) struct Wheel<T> {
     /// The number of milliseconds elapsed since the wheel started.

tokio-util/src/udp/mod.rs

@@ -1,3 +1,5 @@
+#![cfg(not(loom))]
+
 //! UDP framing
 
 mod frame;

tokio-util/src/util/mod.rs

@@ -6,3 +6,26 @@ pub(crate) use maybe_dangling::MaybeDangling;
 #[cfg(any(feature = "io", feature = "codec"))]
 #[cfg_attr(not(feature = "io"), allow(unreachable_pub))]
 pub use poll_buf::{poll_read_buf, poll_write_buf};
+
+cfg_rt! {
+    #[cfg_attr(not(feature = "io"), allow(unused))]
+    pub(crate) use tokio::task::coop::poll_proceed;
+}
+
+cfg_not_rt! {
+    #[cfg_attr(not(feature = "io"), allow(unused))]
+    use std::task::{Context, Poll};
+
+    #[cfg_attr(not(feature = "io"), allow(unused))]
+    pub(crate) struct RestoreOnPending;
+
+    #[cfg_attr(not(feature = "io"), allow(unused))]
+    impl RestoreOnPending {
+        pub(crate) fn made_progress(&self) {}
+    }
+
+    #[cfg_attr(not(feature = "io"), allow(unused))]
+    pub(crate) fn poll_proceed(_cx: &mut Context<'_>) -> Poll<RestoreOnPending> {
+        Poll::Ready(RestoreOnPending)
+    }
+}

tokio-util/tests/abort_on_drop.rs

@@ -1,5 +1,5 @@
 use tokio::{sync::oneshot, task::yield_now};
-use tokio_util::task::AbortOnDropHandle;
+use tokio_util::task::{AbortOnDrop, AbortOnDropHandle};
 
 #[tokio::test]
 async fn aborts_task_on_drop() {
@@ -35,5 +35,42 @@ async fn does_not_abort_after_detach() {
     let handle = AbortOnDropHandle::new(handle);
     handle.detach(); // returns and drops the original join handle
     yield_now().await;
-    assert!(!tx.is_closed()); // task is still live
+    assert!(!tx.is_closed()); // the task is still alive
+}
+
+#[tokio::test]
+async fn handle_aborts_task_on_drop() {
+    let (mut tx, rx) = oneshot::channel::<bool>();
+    let handle = tokio::spawn(async move {
+        let _ = rx.await;
+    });
+    let handle = AbortOnDrop::new(handle.abort_handle());
+    drop(handle);
+    tx.closed().await;
+    assert!(tx.is_closed());
+}
+
+#[tokio::test]
+async fn handle_aborts_task_directly() {
+    let (mut tx, rx) = oneshot::channel::<bool>();
+    let handle = tokio::spawn(async move {
+        let _ = rx.await;
+    });
+    let handle = AbortOnDrop::new(handle.abort_handle());
+    handle.abort();
+    tx.closed().await;
+    assert!(tx.is_closed());
+    assert!(handle.is_finished());
+}
+
+#[tokio::test]
+async fn handle_does_not_abort_after_detach() {
+    let (tx, rx) = oneshot::channel::<bool>();
+    let handle = tokio::spawn(async move {
+        let _ = rx.await;
+    });
+    let handle = AbortOnDrop::new(handle.abort_handle());
+    handle.detach(); // returns and drops the original abort handle
+    yield_now().await;
+    assert!(!tx.is_closed()); // the task is still alive
 }

tokio-util/tests/framed.rs

@@ -140,6 +140,59 @@ fn external_buf_grows_to_init() {
     assert_eq!(read_buf.capacity(), INITIAL_CAPACITY);
 }
 
+// Regression test: `Framed::from_parts` with an empty read buffer (but with
+// capacity, as produced by `into_parts()` after consuming all data) should NOT
+// call `decode()` before actually reading from the underlying IO.
+//
+// Before the fix, `is_readable` was derived from `buffer.capacity() > 0`, which
+// was always true after `reserve()`. This caused a spurious `decode()` call on
+// the empty buffer before any IO read.
+#[tokio::test]
+async fn from_parts_empty_read_buf_does_not_spuriously_decode() {
+    struct TrackingDecoder {
+        decode_count: usize,
+    }
+
+    impl Decoder for TrackingDecoder {
+        type Item = u32;
+        type Error = io::Error;
+
+        fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<u32>> {
+            self.decode_count += 1;
+            if buf.len() < 4 {
+                return Ok(None);
+            }
+            let n = buf.split_to(4).get_u32();
+            Ok(Some(n))
+        }
+    }
+
+    impl Encoder<u32> for TrackingDecoder {
+        type Error = io::Error;
+        fn encode(&mut self, item: u32, dst: &mut BytesMut) -> io::Result<()> {
+            dst.reserve(4);
+            dst.put_u32(item);
+            Ok(())
+        }
+    }
+
+    // Underlying IO provides exactly one 4-byte frame.
+    let data: &[u8] = &[0, 0, 0, 42];
+    let mut parts = FramedParts::new(data, TrackingDecoder { decode_count: 0 });
+    // Simulate a buffer recycled from a previous Framed via `into_parts()`:
+    // empty but has capacity.
+    parts.read_buf = BytesMut::with_capacity(INITIAL_CAPACITY);
+
+    let mut framed = Framed::from_parts(parts);
+    let num = assert_ok!(framed.next().await.unwrap());
+    assert_eq!(num, 42);
+
+    // With the fix: decode is called once (after reading data from IO).
+    // Before the fix: decode was called twice (once on the empty buffer, then
+    // once after reading), because `is_readable` was incorrectly `true`.
+    assert_eq!(framed.codec().decode_count, 1);
+}
+
 #[test]
 fn external_buf_does_not_shrink() {
     let mut parts = FramedParts::new(DontReadIntoThis, U32Codec::default());

tokio-util/tests/io_simplex.rs

@@ -0,0 +1,356 @@
+use futures::pin_mut;
+use futures_test::task::noop_context;
+use std::io::IoSlice;
+use std::task::Poll;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
+use tokio_test::task::spawn;
+use tokio_test::{assert_pending, assert_ready};
+use tokio_util::io::simplex;
+
+/// Sanity check for single-threaded operation.
+#[tokio::test]
+async fn single_thread() {
+    const N: usize = 64;
+    const MSG: &[u8] = b"Hello, world!";
+    const CAPS: &[usize] = &[1, MSG.len() / 2, MSG.len() - 1, MSG.len(), MSG.len() + 1];
+
+    // test different buffer capacities to cover edge cases
+    for &capacity in CAPS {
+        let (mut tx, mut rx) = simplex::new(capacity);
+
+        for _ in 0..N {
+            let mut read = 0;
+            let mut write = 0;
+            let mut buf = [0; MSG.len()];
+
+            while read < MSG.len() || write < MSG.len() {
+                if write < MSG.len() {
+                    let n = tx.write(&MSG[write..]).await.unwrap();
+                    write += n;
+                }
+
+                if read < MSG.len() {
+                    let n = rx.read(&mut buf[read..]).await.unwrap();
+                    read += n;
+                }
+            }
+
+            assert_eq!(&buf[..], MSG);
+        }
+    }
+}
+
+/// Sanity check for multi-threaded operation.
+#[test]
+#[cfg(not(target_os = "wasi"))] // No thread on wasi.
+fn multi_thread() {
+    use futures::executor::block_on;
+    use std::thread;
+
+    const N: usize = 64;
+    const MSG: &[u8] = b"Hello, world!";
+    const CAPS: &[usize] = &[1, MSG.len() / 2, MSG.len() - 1, MSG.len(), MSG.len() + 1];
+
+    // test different buffer capacities to cover edge cases
+    for &capacity in CAPS {
+        let (mut tx, mut rx) = simplex::new(capacity);
+
+        let jh0 = thread::spawn(move || {
+            block_on(async {
+                let mut buf = vec![0; MSG.len()];
+                for _ in 0..N {
+                    rx.read_exact(&mut buf).await.unwrap();
+                    assert_eq!(&buf[..], MSG);
+                    buf.clear();
+                    buf.resize(MSG.len(), 0);
+                }
+            });
+        });
+
+        let jh1 = thread::spawn(move || {
+            block_on(async {
+                for _ in 0..N {
+                    tx.write_all(MSG).await.unwrap();
+                }
+            });
+        });
+
+        jh0.join().unwrap();
+        jh1.join().unwrap();
+    }
+}
+
+#[test]
+#[should_panic(expected = "capacity must be greater than zero")]
+fn zero_capacity() {
+    let _ = simplex::new(0);
+}
+
+/// The `Receiver::poll_read` should return `Poll::Ready(Ok(()))`
+/// if the `ReadBuf` has zero remaining capacity.
+#[tokio::test]
+async fn read_buf_is_full() {
+    let (_tx, rx) = simplex::new(32);
+    let mut buf = ReadBuf::new(&mut []);
+    tokio::pin!(rx);
+    assert_ready!(rx.as_mut().poll_read(&mut noop_context(), &mut buf)).unwrap();
+    assert_eq!(buf.filled().len(), 0);
+}
+
+/// The `Sender::poll_write` should return `Poll::Ready(Ok(0))`
+/// if the input buffer has zero length.
+#[tokio::test]
+async fn write_buf_is_empty() {
+    let (tx, _rx) = simplex::new(32);
+    tokio::pin!(tx);
+    let n = assert_ready!(tx.as_mut().poll_write(&mut noop_context(), &[])).unwrap();
+    assert_eq!(n, 0);
+}
+
+/// The `Sender` should returns error if the `Receiver` has been dropped.
+#[tokio::test]
+async fn drop_receiver_0() {
+    let (mut tx, rx) = simplex::new(32);
+    drop(rx);
+
+    tx.write_u8(1).await.unwrap_err();
+}
+
+/// The `Sender` should be woken up if the `Receiver` has been dropped.
+#[tokio::test]
+async fn drop_receiver_1() {
+    let (mut tx, rx) = simplex::new(1);
+    let mut write_task = spawn(tx.write_u16(1));
+    assert_pending!(write_task.poll());
+
+    assert!(!write_task.is_woken());
+    drop(rx);
+    assert!(write_task.is_woken());
+}
+
+/// The `Receiver` should return error if:
+///
+/// - The `Sender` has been dropped.
+/// - AND there is no remaining data in the buffer.
+#[tokio::test]
+async fn drop_sender_0() {
+    const MSG: &[u8] = b"Hello, world!";
+
+    let (tx, mut rx) = simplex::new(32);
+    drop(tx);
+
+    let mut buf = vec![0; MSG.len()];
+    rx.read_exact(&mut buf).await.unwrap_err();
+}
+
+/// The `Receiver` should be woken up if:
+///
+/// - The `Sender` has been dropped.
+/// - AND there is still remaining data in the buffer.
+#[tokio::test]
+async fn drop_sender_1() {
+    let (mut tx, mut rx) = simplex::new(2);
+    let mut buf = vec![];
+    let mut read_task = spawn(rx.read_to_end(&mut buf));
+    assert_pending!(read_task.poll());
+
+    tx.write_u8(1).await.unwrap();
+    assert_pending!(read_task.poll());
+
+    assert!(!read_task.is_woken());
+    drop(tx);
+    assert!(read_task.is_woken());
+
+    read_task.await.unwrap();
+    assert_eq!(buf, vec![1]);
+}
+
+/// All following calls to `Sender::poll_write` and `Sender::poll_flush`
+/// should return error after `shutdown` has been called.
+#[tokio::test]
+async fn shutdown_sender_0() {
+    const MSG: &[u8] = b"Hello, world!";
+
+    let (mut tx, _rx) = simplex::new(32);
+    tx.shutdown().await.unwrap();
+
+    tx.write_all(MSG).await.unwrap_err();
+    tx.flush().await.unwrap_err();
+}
+
+/// The `Sender::poll_shutdown` should be called multiple times
+/// without error.
+#[tokio::test]
+async fn shutdown_sender_1() {
+    let (mut tx, _rx) = simplex::new(32);
+    tx.shutdown().await.unwrap();
+    tx.shutdown().await.unwrap();
+}
+
+/// The `Sender::poll_shutdown` should wake up the `Receiver`
+#[tokio::test]
+async fn shutdown_sender_2() {
+    let (mut tx, mut rx) = simplex::new(32);
+
+    let mut buf = vec![];
+    let mut read_task = spawn(rx.read_to_end(&mut buf));
+    assert_pending!(read_task.poll());
+
+    tx.write_u8(1).await.unwrap();
+    assert_pending!(read_task.poll());
+
+    assert!(!read_task.is_woken());
+    tx.shutdown().await.unwrap();
+    assert!(read_task.is_woken());
+
+    read_task.await.unwrap();
+    assert_eq!(buf, vec![1]);
+}
+
+/// Both `Sender` and `Receiver` should yield periodically
+/// in a tight-loop.
+#[tokio::test]
+#[cfg(feature = "rt")]
+async fn cooperative_scheduling() {
+    // this magic number is copied from
+    // https://github.com/tokio-rs/tokio/blob/925c614c89d0a26777a334612e2ed6ad0e7935c3/tokio/src/task/coop/mod.rs#L116
+    const INITIAL_BUDGET: usize = 128;
+
+    let (tx, _rx) = simplex::new(INITIAL_BUDGET * 2);
+    pin_mut!(tx);
+    let mut is_pending = false;
+    for _ in 0..INITIAL_BUDGET + 1 {
+        match tx.as_mut().poll_write(&mut noop_context(), &[0u8; 1]) {
+            Poll::Pending => {
+                is_pending = true;
+                break;
+            }
+            Poll::Ready(Ok(1)) => {}
+            Poll::Ready(Ok(n)) => panic!("wrote too many bytes: {n}"),
+            Poll::Ready(Err(e)) => panic!("{e}"),
+        }
+    }
+    assert!(is_pending);
+
+    let (tx, _rx) = simplex::new(INITIAL_BUDGET * 2);
+    pin_mut!(tx);
+    let mut is_pending = false;
+    let io_slices = &[IoSlice::new(&[0u8; 1])];
+    for _ in 0..INITIAL_BUDGET + 1 {
+        match tx
+            .as_mut()
+            .poll_write_vectored(&mut noop_context(), io_slices)
+        {
+            Poll::Pending => {
+                is_pending = true;
+                break;
+            }
+            Poll::Ready(Ok(1)) => {}
+            Poll::Ready(Ok(n)) => panic!("wrote too many bytes: {n}"),
+            Poll::Ready(Err(e)) => panic!("{e}"),
+        }
+    }
+    assert!(is_pending);
+
+    let (mut tx, rx) = simplex::new(INITIAL_BUDGET * 2);
+    tx.write_all(&[0u8; INITIAL_BUDGET + 2]).await.unwrap();
+    pin_mut!(rx);
+    let mut is_pending = false;
+    for _ in 0..INITIAL_BUDGET + 1 {
+        let mut buf = [0u8; 1];
+        let mut buf = ReadBuf::new(&mut buf);
+        match rx.as_mut().poll_read(&mut noop_context(), &mut buf) {
+            Poll::Pending => {
+                is_pending = true;
+                break;
+            }
+            Poll::Ready(Ok(())) => assert_eq!(buf.filled().len(), 1),
+            Poll::Ready(Err(e)) => panic!("{e}"),
+        }
+    }
+    assert!(is_pending);
+}
+
+/// The capacity is exactly same as the total length of the vectored buffers.
+#[tokio::test]
+async fn poll_write_vectored_0() {
+    const MSG1: &[u8] = b"1";
+    const MSG2: &[u8] = b"22";
+    const MSG3: &[u8] = b"333";
+    const MSG_LEN: usize = MSG1.len() + MSG2.len() + MSG3.len();
+
+    let io_slices = &[IoSlice::new(MSG1), IoSlice::new(MSG2), IoSlice::new(MSG3)];
+
+    let (tx, mut rx) = simplex::new(MSG_LEN);
+    tokio::pin!(tx);
+    let res = tx.poll_write_vectored(&mut noop_context(), io_slices);
+    let n = assert_ready!(res).unwrap();
+    assert_eq!(n, MSG_LEN);
+    let mut buf = [0; MSG_LEN];
+    let n = rx.read_exact(&mut buf).await.unwrap();
+    assert_eq!(n, MSG_LEN);
+    assert_eq!(&buf, b"122333");
+}
+
+/// The capacity is smaller than the total length of the vectored buffers.
+#[tokio::test]
+async fn poll_write_vectored_1() {
+    const MSG1: &[u8] = b"1";
+    const MSG2: &[u8] = b"22";
+    const MSG3: &[u8] = b"333";
+    const CAPACITY: usize = MSG1.len() + MSG2.len() + 1;
+
+    let io_slices = &[IoSlice::new(MSG1), IoSlice::new(MSG2), IoSlice::new(MSG3)];
+
+    let (tx, mut rx) = simplex::new(CAPACITY);
+    tokio::pin!(tx);
+
+    // ==== The poll_write_vectored should write MSG1 and MSG2 fully, and MSG3 partially. ====
+    let res = tx.poll_write_vectored(&mut noop_context(), io_slices);
+    let n = assert_ready!(res).unwrap();
+    assert_eq!(n, CAPACITY);
+    let mut buf = [0; CAPACITY];
+    let n = rx.read_exact(&mut buf).await.unwrap();
+    assert_eq!(n, CAPACITY);
+    assert_eq!(&buf, b"1223");
+}
+
+/// There are two empty buffers in the vectored buffers.
+#[tokio::test]
+async fn poll_write_vectored_2() {
+    const MSG1: &[u8] = b"1";
+    const MSG2: &[u8] = b"";
+    const MSG3: &[u8] = b"22";
+    const MSG4: &[u8] = b"";
+    const MSG5: &[u8] = b"333";
+    const MSG_LEN: usize = MSG1.len() + MSG2.len() + MSG3.len() + MSG4.len() + MSG5.len();
+
+    let io_slices = &[
+        IoSlice::new(MSG1),
+        IoSlice::new(MSG2),
+        IoSlice::new(MSG3),
+        IoSlice::new(MSG4),
+        IoSlice::new(MSG5),
+    ];
+
+    let (tx, mut rx) = simplex::new(MSG_LEN);
+    tokio::pin!(tx);
+    let res = tx.poll_write_vectored(&mut noop_context(), io_slices);
+    let n = assert_ready!(res).unwrap();
+    assert_eq!(n, MSG_LEN);
+    let mut buf = [0; MSG_LEN];
+    let n = rx.read_exact(&mut buf).await.unwrap();
+    assert_eq!(n, MSG_LEN);
+    assert_eq!(&buf, b"122333");
+}
+
+/// The `Sender::poll_write_vectored` should return `Poll::Ready(Ok(0))`
+/// if all the input buffers have zero length.
+#[tokio::test]
+async fn poll_write_vectored_3() {
+    let io_slices = &[IoSlice::new(&[]), IoSlice::new(&[]), IoSlice::new(&[])];
+    let (tx, _rx) = simplex::new(32);
+    tokio::pin!(tx);
+    let n = assert_ready!(tx.poll_write_vectored(&mut noop_context(), io_slices)).unwrap();
+    assert_eq!(n, 0);
+}

tokio-util/tests/io_write_all_vectored.rs

@@ -0,0 +1,142 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::io::AsyncWrite;
+use tokio_util::io::write_all_vectored;
+
+use bytes::BytesMut;
+use std::io;
+use std::io::IoSlice;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[tokio::test]
+async fn test_write_all_vectored() {
+    struct Wr {
+        buf: BytesMut,
+    }
+    impl AsyncWrite for Wr {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            _buf: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            // When executing `write_all_buf` with this writer,
+            // `poll_write` is not called.
+            panic!("shouldn't be called")
+        }
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Ok(()).into()
+        }
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Ok(()).into()
+        }
+        fn poll_write_vectored(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            bufs: &[io::IoSlice<'_>],
+        ) -> Poll<Result<usize, io::Error>> {
+            for buf in bufs {
+                self.buf.extend_from_slice(buf);
+            }
+            let n = self.buf.len();
+            Ok(n).into()
+        }
+        fn is_write_vectored(&self) -> bool {
+            // Enable vectored write. (doesn't need to be enabled explicitly for `write_all_vectored`)
+            true
+        }
+    }
+
+    let mut wr = Wr {
+        buf: BytesMut::with_capacity(64),
+    };
+
+    let buf = &mut [
+        IoSlice::new(&b"hello"[..]),
+        IoSlice::new(&b" "[..]),
+        IoSlice::new(&b"world"[..]),
+    ];
+
+    write_all_vectored(&mut wr, buf).await.unwrap();
+    assert_eq!(&wr.buf[..], b"hello world");
+}
+
+#[tokio::test]
+async fn write_all_vectored_with_empty_slice() {
+    struct Wr {
+        buf: BytesMut,
+    }
+    impl AsyncWrite for Wr {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            _buf: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            panic!("shouldn't be called")
+        }
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Ok(()).into()
+        }
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Ok(()).into()
+        }
+        fn poll_write_vectored(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            bufs: &[io::IoSlice<'_>],
+        ) -> Poll<Result<usize, io::Error>> {
+            for buf in bufs {
+                self.buf.extend_from_slice(buf);
+            }
+            let n = self.buf.len();
+            Ok(n).into()
+        }
+        fn is_write_vectored(&self) -> bool {
+            // Enable vectored write.
+            true
+        }
+    }
+
+    // case 1 middle empty slice
+    let mut wr = Wr {
+        buf: BytesMut::with_capacity(64),
+    };
+
+    let buf = &mut [
+        IoSlice::new(&b"hello"[..]),
+        IoSlice::new(&[]),
+        IoSlice::new(&b"world"[..]),
+    ];
+
+    write_all_vectored(&mut wr, buf).await.unwrap();
+    assert_eq!(&wr.buf[..], b"helloworld");
+
+    // case 2 no slices
+    let mut wr = Wr {
+        buf: BytesMut::with_capacity(64),
+    };
+
+    let buf = &mut [];
+
+    write_all_vectored(&mut wr, buf).await.unwrap();
+    assert_eq!(&wr.buf[..], b"");
+
+    // case 3 just an empty slice
+    let mut wr = Wr {
+        buf: BytesMut::with_capacity(64),
+    };
+    let buf = &mut [IoSlice::new(&[])];
+
+    write_all_vectored(&mut wr, buf).await.unwrap();
+    assert_eq!(&wr.buf[..], b"");
+
+    // case 4 ending with empty slice
+    let mut wr = Wr {
+        buf: BytesMut::with_capacity(64),
+    };
+    let buf = &mut [IoSlice::new(b"hello"), IoSlice::new(&[])];
+
+    write_all_vectored(&mut wr, buf).await.unwrap();
+    assert_eq!(&wr.buf[..], b"hello");
+}

tokio-util/tests/udp.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(not(target_os = "wasi"))] // Wasi doesn't support UDP
 #![cfg(not(miri))] // No `socket` in Miri.
+#![cfg(not(loom))] // No udp / UdpFramed in loom
 
 use tokio::net::UdpSocket;
 use tokio_stream::StreamExt;

tokio/CHANGELOG.md

@@ -1,3 +1,197 @@
+# 1.51.0 (April 3rd, 2026)
+
+### Added
+
+- net: implement `get_peer_cred` on Hurd ([#7989])
+- runtime: add `tokio::runtime::worker_index()` ([#7921])
+- runtime: add runtime name ([#7924])
+- runtime: stabilize `LocalRuntime` ([#7557])
+- wasm: add wasm32-wasip2 networking support ([#7933])
+
+### Changed
+
+- runtime: steal tasks from the LIFO slot ([#7431])
+
+### Fixed
+
+- docs: do not show "Available on non-loom only." doc label ([#7977])
+- macros: improve overall macro hygiene ([#7997])
+- sync: fix `notify_waiters` priority in `Notify` ([#7996])
+- sync: fix panic in `Chan::recv_many` when called with non-empty vector on closed channel ([#7991])
+
+[#7431]: https://github.com/tokio-rs/tokio/pull/7431
+[#7557]: https://github.com/tokio-rs/tokio/pull/7557
+[#7921]: https://github.com/tokio-rs/tokio/pull/7921
+[#7924]: https://github.com/tokio-rs/tokio/pull/7924
+[#7933]: https://github.com/tokio-rs/tokio/pull/7933
+[#7977]: https://github.com/tokio-rs/tokio/pull/7977
+[#7989]: https://github.com/tokio-rs/tokio/pull/7989
+[#7991]: https://github.com/tokio-rs/tokio/pull/7991
+[#7996]: https://github.com/tokio-rs/tokio/pull/7996
+[#7997]: https://github.com/tokio-rs/tokio/pull/7997
+
+# 1.50.0 (Mar 3rd, 2026)
+
+### Added
+
+- net: add `TcpStream::set_zero_linger` ([#7837])
+- rt: add `is_rt_shutdown_err` ([#7771])
+
+### Changed
+
+- io: add optimizer hint that `memchr` returns in-bounds pointer ([#7792])
+- io: implement vectored writes for `write_buf` ([#7871])
+- runtime: panic when `event_interval` is set to 0 ([#7838])
+- runtime: shorten default thread name to fit in Linux limit ([#7880])
+- signal: remember the result of `SetConsoleCtrlHandler` ([#7833])
+- signal: specialize windows `Registry` ([#7885])
+
+### Fixed
+
+- io: always cleanup `AsyncFd` registration list on deregister ([#7773])
+- macros: remove (most) local `use` declarations in `tokio::select!` ([#7929])
+- net: fix `GET_BUF_SIZE` constant for `target_os = "android"` ([#7889])
+- runtime: avoid redundant unpark in current_thread scheduler ([#7834])
+- runtime: don't park in `current_thread` if `before_park` defers waker ([#7835])
+- io: fix write readiness on ESP32 on short writes ([#7872])
+- runtime: wake deferred tasks before entering `block_in_place` ([#7879])
+- sync: drop rx waker when oneshot receiver is dropped ([#7886])
+- runtime: fix double increment of `num_idle_threads` on shutdown ([#7910], [#7918], [#7922])
+
+### Unstable
+
+- fs: check for io-uring opcode support ([#7815])
+- runtime: avoid lock acquisition after uring init ([#7850])
+
+### Documented
+
+- docs: update outdated unstable features section ([#7839])
+- io: clarify the behavior of `AsyncWriteExt::shutdown()` ([#7908])
+- io: explain how to flush stdout/stderr ([#7904])
+- io: fix incorrect and confusing `AsyncWrite` documentation ([#7875])
+- rt: clarify the documentation of `Runtime::spawn` ([#7803])
+- rt: fix missing quotation in docs ([#7925])
+- runtime: correct the default thread name in docs ([#7896])
+- runtime: fix `event_interval` doc ([#7932])
+- sync: clarify RwLock fairness documentation ([#7919])
+- sync: clarify that `recv` returns `None` once closed and no more messages ([#7920])
+- task: clarify when to use `spawn_blocking` vs dedicated threads ([#7923])
+- task: doc that task drops before `JoinHandle` completion ([#7825])
+- signal: guarantee that listeners never return `None` ([#7869])
+- task: fix task module feature flags in docs ([#7891])
+- task: fix two typos ([#7913])
+- task: improve the docs of `Builder::spawn_local` ([#7828])
+- time: add docs about auto-advance and when to use sleep ([#7858])
+- util: fix typo in docs ([#7926])
+
+[#7771]: https://github.com/tokio-rs/tokio/pull/7771
+[#7773]: https://github.com/tokio-rs/tokio/pull/7773
+[#7792]: https://github.com/tokio-rs/tokio/pull/7792
+[#7803]: https://github.com/tokio-rs/tokio/pull/7803
+[#7815]: https://github.com/tokio-rs/tokio/pull/7815
+[#7825]: https://github.com/tokio-rs/tokio/pull/7825
+[#7828]: https://github.com/tokio-rs/tokio/pull/7828
+[#7833]: https://github.com/tokio-rs/tokio/pull/7833
+[#7834]: https://github.com/tokio-rs/tokio/pull/7834
+[#7835]: https://github.com/tokio-rs/tokio/pull/7835
+[#7837]: https://github.com/tokio-rs/tokio/pull/7837
+[#7838]: https://github.com/tokio-rs/tokio/pull/7838
+[#7839]: https://github.com/tokio-rs/tokio/pull/7839
+[#7850]: https://github.com/tokio-rs/tokio/pull/7850
+[#7858]: https://github.com/tokio-rs/tokio/pull/7858
+[#7869]: https://github.com/tokio-rs/tokio/pull/7869
+[#7871]: https://github.com/tokio-rs/tokio/pull/7871
+[#7872]: https://github.com/tokio-rs/tokio/pull/7872
+[#7875]: https://github.com/tokio-rs/tokio/pull/7875
+[#7879]: https://github.com/tokio-rs/tokio/pull/7879
+[#7880]: https://github.com/tokio-rs/tokio/pull/7880
+[#7885]: https://github.com/tokio-rs/tokio/pull/7885
+[#7886]: https://github.com/tokio-rs/tokio/pull/7886
+[#7889]: https://github.com/tokio-rs/tokio/pull/7889
+[#7891]: https://github.com/tokio-rs/tokio/pull/7891
+[#7896]: https://github.com/tokio-rs/tokio/pull/7896
+[#7904]: https://github.com/tokio-rs/tokio/pull/7904
+[#7908]: https://github.com/tokio-rs/tokio/pull/7908
+[#7910]: https://github.com/tokio-rs/tokio/pull/7910
+[#7913]: https://github.com/tokio-rs/tokio/pull/7913
+[#7918]: https://github.com/tokio-rs/tokio/pull/7918
+[#7919]: https://github.com/tokio-rs/tokio/pull/7919
+[#7920]: https://github.com/tokio-rs/tokio/pull/7920
+[#7922]: https://github.com/tokio-rs/tokio/pull/7922
+[#7923]: https://github.com/tokio-rs/tokio/pull/7923
+[#7925]: https://github.com/tokio-rs/tokio/pull/7925
+[#7926]: https://github.com/tokio-rs/tokio/pull/7926
+[#7929]: https://github.com/tokio-rs/tokio/pull/7929
+[#7932]: https://github.com/tokio-rs/tokio/pull/7932
+
+# 1.49.0 (January 3rd, 2026)
+
+### Added
+
+* net: add support for `TCLASS` option on IPv6 ([#7781])
+* runtime: stabilize `runtime::id::Id` ([#7125])
+* task: implement `Extend` for `JoinSet` ([#7195])
+* task: stabilize the `LocalSet::id()` ([#7776])
+
+### Changed
+
+* net: deprecate `{TcpStream,TcpSocket}::set_linger` ([#7752])
+
+### Fixed
+
+* macros: fix the hygiene issue of `join!` and `try_join!` ([#7766])
+* runtime: revert "replace manual vtable definitions with Wake" ([#7699])
+* sync: return `TryRecvError::Disconnected` from `Receiver::try_recv` after `Receiver::close` ([#7686])
+* task: remove unnecessary trait bounds on the `Debug` implementation ([#7720])
+
+### Unstable
+
+* fs: handle `EINTR` in `fs::write` for io-uring ([#7786])
+* fs: support io-uring with `tokio::fs::read` ([#7696])
+* runtime: disable io-uring on `EPERM` ([#7724])
+* time: add alternative timer for better multicore scalability ([#7467])
+
+### Documented
+
+* docs: fix a typos in `bounded.rs` and `park.rs` ([#7817])
+* io: add `SyncIoBridge` cross-references to `copy` and `copy_buf` ([#7798])
+* io: doc that `AsyncWrite` does not inherit from `std::io::Write` ([#7705])
+* metrics: clarify that `num_alive_tasks` is not strongly consistent ([#7614])
+* net: clarify the cancellation safety of the `TcpStream::peek` ([#7305])
+* net: clarify the drop behavior of `unix::OwnedWriteHalf` ([#7742])
+* net: clarify the platform-dependent backlog in `TcpSocket` docs ([#7738])
+* runtime: mention `LocalRuntime` in `new_current_thread` docs ([#7820])
+* sync: add missing period to `mpsc::Sender::try_send` docs ([#7721])
+* sync: clarify the cancellation safety of `oneshot::Receiver` ([#7780])
+* sync: improve the docs for the `errors` of mpsc ([#7722])
+* task: add example for `spawn_local` usage on local runtime ([#7689])
+
+[#7125]: https://github.com/tokio-rs/tokio/pull/7125
+[#7195]: https://github.com/tokio-rs/tokio/pull/7195
+[#7305]: https://github.com/tokio-rs/tokio/pull/7305
+[#7467]: https://github.com/tokio-rs/tokio/pull/7467
+[#7614]: https://github.com/tokio-rs/tokio/pull/7614
+[#7686]: https://github.com/tokio-rs/tokio/pull/7686
+[#7689]: https://github.com/tokio-rs/tokio/pull/7689
+[#7696]: https://github.com/tokio-rs/tokio/pull/7696
+[#7699]: https://github.com/tokio-rs/tokio/pull/7699
+[#7705]: https://github.com/tokio-rs/tokio/pull/7705
+[#7720]: https://github.com/tokio-rs/tokio/pull/7720
+[#7721]: https://github.com/tokio-rs/tokio/pull/7721
+[#7722]: https://github.com/tokio-rs/tokio/pull/7722
+[#7724]: https://github.com/tokio-rs/tokio/pull/7724
+[#7738]: https://github.com/tokio-rs/tokio/pull/7738
+[#7742]: https://github.com/tokio-rs/tokio/pull/7742
+[#7752]: https://github.com/tokio-rs/tokio/pull/7752
+[#7766]: https://github.com/tokio-rs/tokio/pull/7766
+[#7776]: https://github.com/tokio-rs/tokio/pull/7776
+[#7780]: https://github.com/tokio-rs/tokio/pull/7780
+[#7781]: https://github.com/tokio-rs/tokio/pull/7781
+[#7786]: https://github.com/tokio-rs/tokio/pull/7786
+[#7798]: https://github.com/tokio-rs/tokio/pull/7798
+[#7817]: https://github.com/tokio-rs/tokio/pull/7817
+[#7820]: https://github.com/tokio-rs/tokio/pull/7820
+
 # 1.48.0 (October 14th, 2025)
 
 The MSRV is increased to 1.71.
@@ -111,6 +305,20 @@ The MSRV is increased to 1.71.
 [#7672]: https://github.com/tokio-rs/tokio/pull/7672
 [#7675]: https://github.com/tokio-rs/tokio/pull/7675
 
+# 1.47.4 (April 2nd, 2026)
+
+### Fixed
+
+* sync: fix panic in `Chan::recv_many` when called with non-empty vector on closed channel ([#7991])
+
+[#7991]: https://github.com/tokio-rs/tokio/pull/7991
+
+# 1.47.3 (January 3rd, 2026)
+
+### Fixed
+
+* sync: return `TryRecvError::Disconnected` from `Receiver::try_recv` after `Receiver::close` ([#7686])
+
 # 1.47.2 (October 14th, 2025)
 
 ### Fixed
@@ -378,6 +586,14 @@ comment on [#7172].
 [#7186]: https://github.com/tokio-rs/tokio/pull/7186
 [#7192]: https://github.com/tokio-rs/tokio/pull/7192
 
+# 1.43.4 (January 3rd, 2026)
+
+### Fixed
+
+* sync: return `TryRecvError::Disconnected` from `Receiver::try_recv` after `Receiver::close` ([#7686])
+
+[#7686]: https://github.com/tokio-rs/tokio/pull/7686
+
 # 1.43.3 (October 14th, 2025)
 
 ### Fixed

tokio/Cargo.toml

@@ -1,12 +1,12 @@
 [package]
 name = "tokio"
 # When releasing to crates.io:
-# - Remove path dependencies
+# - Remove path dependencies (if any)
 # - Update doc url
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.48.0"
+version = "1.51.0"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -90,17 +90,17 @@ io-uring = ["dep:io-uring", "libc", "mio/os-poll", "mio/os-ext", "dep:slab"]
 taskdump = ["dep:backtrace"]
 
 [dependencies]
-tokio-macros = { version = "~2.6.0", path = "../tokio-macros", optional = true }
+tokio-macros = { version = "~2.7.0", path = "../tokio-macros", optional = true }
 
 pin-project-lite = "0.2.11"
 
 # Everything else is optional...
 bytes = { version = "1.2.1", optional = true }
-mio = { version = "1.0.1", optional = true, default-features = false }
+mio = { version = "1.2.0", optional = true, default-features = false }
 parking_lot = { version = "0.12.0", optional = true }
 
-[target.'cfg(not(target_family = "wasm"))'.dependencies]
-socket2 = { version = "0.6.0", optional = true, features = ["all"] }
+[target.'cfg(any(not(target_family = "wasm"), all(target_os = "wasi", not(target_env = "p1"))))'.dependencies]
+socket2 = { version = "0.6.3", optional = true, features = ["all"] }
 
 # Currently unstable. The API exposed by these features may be broken at any time.
 # Requires `--cfg tokio_unstable` to enable.
@@ -110,9 +110,9 @@ tracing = { version = "0.1.29", default-features = false, features = ["std"], op
 # Currently unstable. The API exposed by these features may be broken at any time.
 # Requires `--cfg tokio_unstable` to enable.
 [target.'cfg(all(tokio_unstable, target_os = "linux"))'.dependencies]
-io-uring = { version = "0.7.6", default-features = false, optional = true }
+io-uring = { version = "0.7.11", default-features = false, optional = true }
 libc = { version = "0.2.168", optional = true }
-mio = { version = "1.0.1", default-features = false, features = ["os-poll", "os-ext"], optional = true }
+mio = { version = "1.2.0", default-features = false, features = ["os-poll", "os-ext"], optional = true }
 slab = { version = "0.4.9", optional = true }
 backtrace = { version = "0.3.58", optional = true }
 
@@ -120,6 +120,9 @@ backtrace = { version = "0.3.58", optional = true }
 libc = { version = "0.2.168", optional = true }
 signal-hook-registry = { version = "1.1.1", optional = true }
 
+[target.'cfg(target_os = "wasi")'.dependencies]
+libc = { version = "0.2.168", optional = true }
+
 [target.'cfg(unix)'.dev-dependencies]
 libc = { version = "0.2.168" }
 nix = { version = "0.29.0", default-features = false, features = ["aio", "fs", "socket"] }
@@ -136,10 +139,11 @@ features = [
 ]
 
 [dev-dependencies]
-tokio-test = { version = "0.4.0", path = "../tokio-test" }
-tokio-stream = { version = "0.1", path = "../tokio-stream" }
-tokio-util = { version = "0.7", path = "../tokio-util", features = ["rt"] }
+tokio-test = "0.4.0"
+tokio-stream = "0.1"
+tokio-util = { version = "0.7", features = ["rt"] }
 futures = { version = "0.3.0", features = ["async-await"] }
+futures-test = "0.3.31"
 mockall = "0.13.0"
 async-stream = "0.3"
 futures-concurrency = "7.6.3"

tokio/README.md

@@ -1,3 +1,7 @@
+*[TokioConf 2026 program and tickets are now available!](https://tokioconf.com)*
+
+---
+
 # Tokio
 
 A runtime for writing reliable, asynchronous, and slim applications with
@@ -52,11 +56,11 @@ an asynchronous application.
 
 A basic TCP echo server with Tokio.
 
-Make sure you activated the full features of the tokio crate on Cargo.toml:
+Make sure you enable the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.48.0", features = ["full"] }
+tokio = { version = "1.51.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -103,7 +107,7 @@ More examples can be found [here][examples]. For a larger "real world" example,
 [examples]: https://github.com/tokio-rs/tokio/tree/master/examples
 [mini-redis]: https://github.com/tokio-rs/mini-redis/
 
-To see a list of the available features flags that can be enabled, check our
+To see a list of the available feature flags that can be enabled, check our
 [docs][feature-flag-docs].
 
 ## Getting Help
@@ -125,7 +129,7 @@ question. You can also ask your question on [the discussions page][discussions].
 you! We have a [contributing guide][guide] to help you get involved in the Tokio
 project.
 
-[guide]: https://github.com/tokio-rs/tokio/blob/master/CONTRIBUTING.md
+[guide]: https://github.com/tokio-rs/tokio/blob/master/docs/contributing/README.md
 
 ## Related Projects
 
@@ -217,18 +221,18 @@ 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.43.x` - LTS release until March 2026. (MSRV 1.70)
  * `1.47.x` - LTS release until September 2026. (MSRV 1.70)
+ * `1.51.x` - LTS release until March 2027. (MSRV 1.71)
 
 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
 use an LTS release.
 
 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.43.x` patch release, you
+example, to specify that you wish to use the newest `1.47.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.43", features = [...] }
+tokio = { version = "~1.47", features = [...] }
 ```
 
 ### Previous LTS releases
@@ -241,6 +245,7 @@ tokio = { version = "~1.43", features = [...] }
  * `1.32.x` - LTS release until September 2024.
  * `1.36.x` - LTS release until March 2025.
  * `1.38.x` - LTS release until July 2025.
+ * `1.43.x` - LTS release until March 2026.
 
 ## License
 
@@ -251,5 +256,5 @@ This project is licensed under the [MIT license].
 ### Contribution
 
 Unless you explicitly state otherwise, any contribution intentionally submitted
-for inclusion in Tokio by you, shall be licensed as MIT, without any additional
+for inclusion in Tokio by you shall be licensed as MIT, without any additional
 terms or conditions.

tokio/src/fs/file.rs

@@ -917,7 +917,9 @@ impl std::os::unix::io::AsFd for File {
 #[cfg(unix)]
 impl std::os::unix::io::FromRawFd for File {
     unsafe fn from_raw_fd(fd: std::os::unix::io::RawFd) -> Self {
-        StdFile::from_raw_fd(fd).into()
+        // Safety: exactly the same safety contract as
+        // `std::os::unix::io::FromRawFd::from_raw_fd`.
+        unsafe { StdFile::from_raw_fd(fd).into() }
     }
 }
 
@@ -942,7 +944,9 @@ cfg_windows! {
 
     impl FromRawHandle for File {
         unsafe fn from_raw_handle(handle: RawHandle) -> Self {
-            StdFile::from_raw_handle(handle).into()
+            // Safety: exactly the same safety contract as
+            // `FromRawHandle::from_raw_handle`.
+            unsafe { StdFile::from_raw_handle(handle).into() }
         }
     }
 }

tokio/src/fs/mod.rs

@@ -237,9 +237,6 @@ pub use self::metadata::metadata;
 
 mod open_options;
 pub use self::open_options::OpenOptions;
-cfg_io_uring! {
-    pub(crate) use self::open_options::UringOpenOptions;
-}
 
 mod read;
 pub use self::read::read;
@@ -298,6 +295,13 @@ cfg_windows! {
     pub use self::symlink_file::symlink_file;
 }
 
+cfg_io_uring! {
+    pub(crate) mod read_uring;
+    pub(crate) use self::read_uring::read_uring;
+
+    pub(crate) use self::open_options::UringOpenOptions;
+}
+
 use std::io;
 
 #[cfg(not(test))]

tokio/src/fs/open_options.rs

@@ -531,7 +531,10 @@ impl OpenOptions {
                 let handle = crate::runtime::Handle::current();
                 let driver_handle = handle.inner.driver().io();
 
-                if driver_handle.check_and_init()? {
+                if driver_handle
+                    .check_and_init(io_uring::opcode::OpenAt::CODE)
+                    .await?
+                {
                     Op::open(path.as_ref(), opts)?.await
                 } else {
                     let opts = opts.clone().into();

tokio/src/fs/read.rs

@@ -30,6 +30,16 @@ use std::{io, path::Path};
 ///
 /// [`ErrorKind::Interrupted`]: std::io::ErrorKind::Interrupted
 ///
+/// # io_uring support
+///
+/// On Linux, you can also use io_uring for executing system calls. To enable
+/// io_uring, you need to specify the `--cfg tokio_unstable` flag at compile time,
+/// enable the io-uring cargo feature, and set the `Builder::enable_io_uring`
+/// runtime option.
+///
+/// Support for io_uring is currently experimental, so its behavior may change
+/// or it may be removed in future versions.
+///
 /// # Examples
 ///
 /// ```no_run
@@ -45,5 +55,26 @@ use std::{io, path::Path};
 /// ```
 pub async fn read(path: impl AsRef<Path>) -> io::Result<Vec<u8>> {
     let path = path.as_ref().to_owned();
+
+    #[cfg(all(
+        tokio_unstable,
+        feature = "io-uring",
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux"
+    ))]
+    {
+        use crate::fs::read_uring;
+
+        let handle = crate::runtime::Handle::current();
+        let driver_handle = handle.inner.driver().io();
+        if driver_handle
+            .check_and_init(io_uring::opcode::Read::CODE)
+            .await?
+        {
+            return read_uring(&path).await;
+        }
+    }
+
     asyncify(move || std::fs::read(path)).await
 }

tokio/src/fs/read_uring.rs

@@ -0,0 +1,134 @@
+use crate::fs::OpenOptions;
+use crate::runtime::driver::op::Op;
+
+use std::io;
+use std::io::ErrorKind;
+use std::os::fd::OwnedFd;
+use std::path::Path;
+
+// this algorithm is inspired from rust std lib version 1.90.0
+// https://doc.rust-lang.org/1.90.0/src/std/io/mod.rs.html#409
+const PROBE_SIZE: usize = 32;
+const PROBE_SIZE_U32: u32 = PROBE_SIZE as u32;
+
+// Max bytes we can read using io uring submission at a time
+// SAFETY: cannot be higher than u32::MAX for safe cast
+// Set to read max 64 MiB at time
+const MAX_READ_SIZE: usize = 64 * 1024 * 1024;
+
+pub(crate) async fn read_uring(path: &Path) -> io::Result<Vec<u8>> {
+    let file = OpenOptions::new().read(true).open(path).await?;
+
+    // TODO: use io uring in the future to obtain metadata
+    let size_hint: Option<usize> = file.metadata().await.map(|m| m.len() as usize).ok();
+
+    let fd: OwnedFd = file
+        .try_into_std()
+        .expect("unexpected in-flight operation detected")
+        .into();
+
+    let mut buf = Vec::new();
+
+    if let Some(size_hint) = size_hint {
+        buf.try_reserve(size_hint)?;
+    }
+
+    read_to_end_uring(fd, buf).await
+}
+
+async fn read_to_end_uring(mut fd: OwnedFd, mut buf: Vec<u8>) -> io::Result<Vec<u8>> {
+    let mut offset = 0;
+    let start_cap = buf.capacity();
+
+    loop {
+        if buf.len() == buf.capacity() && buf.capacity() == start_cap && buf.len() >= PROBE_SIZE {
+            // The buffer might be an exact fit. Let's read into a probe buffer
+            // and see if it returns `Ok(0)`. If so, we've avoided an
+            // unnecessary increasing of the capacity. But if not, append the
+            // probe buffer to the primary buffer and let its capacity grow.
+            let (r_fd, r_buf, is_eof) = small_probe_read(fd, buf, &mut offset).await?;
+
+            if is_eof {
+                return Ok(r_buf);
+            }
+
+            buf = r_buf;
+            fd = r_fd;
+        }
+
+        // buf is full, need more capacity
+        if buf.len() == buf.capacity() {
+            buf.try_reserve(PROBE_SIZE)?;
+        }
+
+        // prepare the spare capacity to be read into
+        let buf_len = usize::min(buf.spare_capacity_mut().len(), MAX_READ_SIZE);
+
+        // buf_len cannot be greater than u32::MAX because MAX_READ_SIZE
+        // is less than u32::MAX
+        let read_len = u32::try_from(buf_len).expect("buf_len must always fit in u32");
+
+        // read into spare capacity
+        let (r_fd, r_buf, is_eof) = op_read(fd, buf, &mut offset, read_len).await?;
+
+        if is_eof {
+            return Ok(r_buf);
+        }
+
+        fd = r_fd;
+        buf = r_buf;
+    }
+}
+
+async fn small_probe_read(
+    fd: OwnedFd,
+    mut buf: Vec<u8>,
+    offset: &mut u64,
+) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
+    let read_len = PROBE_SIZE_U32;
+
+    let mut temp_arr = [0; PROBE_SIZE];
+    // we don't call this function if the buffer's length < PROBE_SIZE
+    let back_bytes_len = buf.len() - PROBE_SIZE;
+
+    temp_arr.copy_from_slice(&buf[back_bytes_len..]);
+
+    // We're decreasing the length of the buffer and len is greater
+    // than PROBE_SIZE. So we can read into the discarded length
+    buf.truncate(back_bytes_len);
+
+    let (r_fd, mut r_buf, is_eof) = op_read(fd, buf, offset, read_len).await?;
+    // If `size_read` returns zero due to reasons such as the buffer's exact fit,
+    // then this `try_reserve` does not perform allocation.
+    r_buf.try_reserve(PROBE_SIZE)?;
+    r_buf.splice(back_bytes_len..back_bytes_len, temp_arr);
+
+    Ok((r_fd, r_buf, is_eof))
+}
+
+// Takes a length to read and returns a single read in the buffer
+//
+// Returns the file descriptor, buffer and EOF reached or not
+async fn op_read(
+    mut fd: OwnedFd,
+    mut buf: Vec<u8>,
+    offset: &mut u64,
+    read_len: u32,
+) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
+    loop {
+        let (res, r_fd, r_buf) = Op::read(fd, buf, read_len, *offset).await;
+
+        match res {
+            Err(e) if e.kind() == ErrorKind::Interrupted => {
+                buf = r_buf;
+                fd = r_fd;
+            }
+            Err(e) => return Err(e),
+            Ok(size_read) => {
+                *offset += size_read as u64;
+
+                return Ok((r_fd, r_buf, size_read == 0));
+            }
+        }
+    }
+}

tokio/src/fs/write.rs

@@ -37,7 +37,10 @@ pub async fn write(path: impl AsRef<Path>, contents: impl AsRef<[u8]>) -> io::Re
     {
         let handle = crate::runtime::Handle::current();
         let driver_handle = handle.inner.driver().io();
-        if driver_handle.check_and_init()? {
+        if driver_handle
+            .check_and_init(io_uring::opcode::Write::CODE)
+            .await?
+        {
             return write_uring(path, contents).await;
         }
     }
@@ -72,10 +75,14 @@ async fn write_uring(path: &Path, mut buf: OwnedBuf) -> io::Result<()> {
     let mut buf_offset: usize = 0;
     let mut file_offset: u64 = 0;
     while buf_offset < total {
-        let (n, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await?;
-        if n == 0 {
-            return Err(io::ErrorKind::WriteZero.into());
-        }
+        let (res, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;
+
+        let n = match res {
+            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
+            Ok(n) => n,
+            Err(e) if e.kind() == io::ErrorKind::Interrupted => 0,
+            Err(e) => return Err(e),
+        };
 
         buf = _buf;
         fd = _fd;

tokio/src/io/async_write.rs

@@ -5,9 +5,11 @@ use std::task::{Context, Poll};
 
 /// Writes bytes asynchronously.
 ///
-/// The trait inherits from [`std::io::Write`] and indicates that an I/O object is
-/// **nonblocking**. All non-blocking I/O objects must return an error when
-/// bytes cannot be written instead of blocking the current thread.
+/// This trait is analogous to the [`std::io::Write`] trait, but integrates with
+/// the asynchronous task system. In particular, the [`poll_write`] method,
+/// unlike [`Write::write`], will automatically queue the current task for wakeup
+/// and return if data is not yet available, rather than blocking the calling
+/// thread.
 ///
 /// Specifically, this means that the [`poll_write`] function will return one of
 /// the following:
@@ -25,22 +27,14 @@ use std::task::{Context, Poll};
 /// * `Poll::Ready(Err(e))` for other errors are standard I/O errors coming from the
 ///   underlying object.
 ///
-/// This trait importantly means that the [`write`][stdwrite] method only works in
-/// the context of a future's task. The object may panic if used outside of a task.
-///
-/// Note that this trait also represents that the  [`Write::flush`][stdflush] method
-/// works very similarly to the `write` method, notably that `Ok(())` means that the
-/// writer has successfully been flushed, a "would block" error means that the
-/// current task is ready to receive a notification when flushing can make more
-/// progress, and otherwise normal errors can happen as well.
-///
 /// Utilities for working with `AsyncWrite` values are provided by
-/// [`AsyncWriteExt`].
+/// [`AsyncWriteExt`]. Most users will interact with `AsyncWrite` types through
+/// these extension methods, which provide ergonomic async functions such as
+/// `write_all` and `flush`.
 ///
 /// [`std::io::Write`]: std::io::Write
+/// [`Write::write`]: std::io::Write::write()
 /// [`poll_write`]: AsyncWrite::poll_write()
-/// [stdwrite]: std::io::Write::write()
-/// [stdflush]: std::io::Write::flush()
 /// [`AsyncWriteExt`]: crate::io::AsyncWriteExt
 pub trait AsyncWrite {
     /// Attempt to write bytes from `buf` into the object.
@@ -59,7 +53,7 @@ pub trait AsyncWrite {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &[u8],
-    ) -> Poll<Result<usize, io::Error>>;
+    ) -> Poll<io::Result<usize>>;
 
     /// Attempts to flush the object, ensuring that any buffered data reach
     /// their destination.
@@ -70,7 +64,7 @@ pub trait AsyncWrite {
     /// `Poll::Pending` and arranges for the current task (via
     /// `cx.waker()`) to receive a notification when the object can make
     /// progress towards flushing.
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>>;
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
 
     /// Initiates or attempts to shut down this writer, returning success when
     /// the I/O connection has completely shut down.
@@ -130,7 +124,7 @@ pub trait AsyncWrite {
     ///
     /// This function will panic if not called within the context of a future's
     /// task.
-    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>>;
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
 
     /// Like [`poll_write`], except that it writes from a slice of buffers.
     ///
@@ -159,7 +153,7 @@ pub trait AsyncWrite {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
-    ) -> Poll<Result<usize, io::Error>> {
+    ) -> Poll<io::Result<usize>> {
         let buf = bufs
             .iter()
             .find(|b| !b.is_empty())

tokio/src/io/poll_evented.rs

@@ -171,7 +171,7 @@ feature! {
             loop {
                 let evt = ready!(self.registration.poll_read_ready(cx))?;
 
-                let b = &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]);
+                let b = unsafe { &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
 
                 // used only when the cfgs below apply
                 #[allow(unused_variables)]
@@ -188,6 +188,7 @@ feature! {
                         // Read more:
                         // https://github.com/tokio-rs/tokio/issues/5866
                         #[cfg(all(
+                            // keep in sync with poll_write
                             not(mio_unsupported_force_poll_poll),
                             any(
                                 // epoll
@@ -213,7 +214,7 @@ feature! {
 
                         // Safety: We trust `TcpStream::read` to have filled up `n` bytes in the
                         // buffer.
-                        buf.assume_init(n);
+                        unsafe { buf.assume_init(n) };
                         buf.advance(n);
                         return Poll::Ready(Ok(()));
                     },
@@ -240,7 +241,28 @@ feature! {
                         // that the socket buffer is full.  Unfortunately this assumption
                         // fails for level-triggered selectors (like on Windows or poll even for
                         // UNIX): https://github.com/tokio-rs/tokio/issues/5866
-                        if n > 0 && (!cfg!(windows) && !cfg!(mio_unsupported_force_poll_poll) && n < buf.len()) {
+                        #[cfg(all(
+                            // keep in sync with poll_read
+                            not(mio_unsupported_force_poll_poll),
+                            any(
+                                // epoll
+                                target_os = "android",
+                                target_os = "illumos",
+                                target_os = "linux",
+                                target_os = "redox",
+                                // kqueue
+                                target_os = "dragonfly",
+                                target_os = "freebsd",
+                                target_os = "ios",
+                                target_os = "macos",
+                                target_os = "netbsd",
+                                target_os = "openbsd",
+                                target_os = "tvos",
+                                target_os = "visionos",
+                                target_os = "watchos",
+                            )
+                        ))]
+                        if 0 < n && n < buf.len() {
                             self.registration.clear_readiness(evt);
                         }
 

tokio/src/io/read_buf.rs

@@ -283,7 +283,9 @@ unsafe impl<'a> bytes::BufMut for ReadBuf<'a> {
 
     // SAFETY: The caller guarantees that at least `cnt` unfilled bytes have been initialized.
     unsafe fn advance_mut(&mut self, cnt: usize) {
-        self.assume_init(cnt);
+        unsafe {
+            self.assume_init(cnt);
+        }
         self.advance(cnt);
     }
 
@@ -311,16 +313,32 @@ impl fmt::Debug for ReadBuf<'_> {
     }
 }
 
+/// # Safety
+///
+/// The caller must ensure that `slice` is fully initialized
+/// and never writes uninitialized bytes to the returned slice.
 unsafe fn slice_to_uninit_mut(slice: &mut [u8]) -> &mut [MaybeUninit<u8>] {
-    &mut *(slice as *mut [u8] as *mut [MaybeUninit<u8>])
+    // SAFETY: `MaybeUninit<u8>` has the same memory layout as u8, and the caller
+    // promises to not write uninitialized bytes to the returned slice.
+    unsafe { &mut *(slice as *mut [u8] as *mut [MaybeUninit<u8>]) }
 }
 
+/// # Safety
+///
+/// The caller must ensure that `slice` is fully initialized.
 // TODO: This could use `MaybeUninit::slice_assume_init` when it is stable.
 unsafe fn slice_assume_init(slice: &[MaybeUninit<u8>]) -> &[u8] {
-    &*(slice as *const [MaybeUninit<u8>] as *const [u8])
+    // SAFETY: `MaybeUninit<u8>` has the same memory layout as u8, and the caller
+    // promises that `slice` is fully initialized.
+    unsafe { &*(slice as *const [MaybeUninit<u8>] as *const [u8]) }
 }
 
+/// # Safety
+///
+/// The caller must ensure that `slice` is fully initialized.
 // TODO: This could use `MaybeUninit::slice_assume_init_mut` when it is stable.
 unsafe fn slice_assume_init_mut(slice: &mut [MaybeUninit<u8>]) -> &mut [u8] {
-    &mut *(slice as *mut [MaybeUninit<u8>] as *mut [u8])
+    // SAFETY: `MaybeUninit<u8>` has the same memory layout as `u8`, and the caller
+    // promises that `slice` is fully initialized.
+    unsafe { &mut *(slice as *mut [MaybeUninit<u8>] as *mut [u8]) }
 }

tokio/src/io/stderr.rs

@@ -29,7 +29,7 @@ cfg_io_std! {
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {
-    ///     let mut stderr = io::stdout();
+    ///     let mut stderr = io::stderr();
     ///     stderr.write_all(b"Print some error here.").await?;
     ///     Ok(())
     /// }
@@ -50,6 +50,20 @@ cfg_io_std! {
     /// to occur as a single write, so multiple threads writing data with
     /// [`write_all`] may result in interleaved output.
     ///
+    /// Note that unlike [`std::io::stderr`], each call to this `stderr()`
+    /// produces a new writer, so for example, this program does **not** flush stderr:
+    ///
+    /// ```no_run
+    /// # use tokio::io::AsyncWriteExt;
+    /// # #[tokio::main]
+    /// # async fn main() -> std::io::Result<()> {
+    /// tokio::io::stderr().write_all(b"aa").await?;
+    /// tokio::io::stderr().flush().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// [`std::io::stderr`]: std::io::stderr
     /// [`AsyncWrite`]: AsyncWrite
     /// [`write_all`]: crate::io::AsyncWriteExt::write_all()
     ///

tokio/src/io/stdio_common.rs

@@ -177,7 +177,7 @@ mod tests {
     }
 
     #[test]
-    #[cfg_attr(miri, ignore)]
+    #[cfg_attr(miri, ignore)] // takes a really long time with miri
     fn test_splitter() {
         let data = str::repeat("█", DEFAULT_MAX_BUF_SIZE);
         let mut wr = super::SplitByUtf8BoundaryIfWindows::new(TextMockWriter);
@@ -191,7 +191,7 @@ mod tests {
     }
 
     #[test]
-    #[cfg_attr(miri, ignore)]
+    #[cfg_attr(miri, ignore)] // takes a really long time with miri
     fn test_pseudo_text() {
         // In this test we write a piece of binary data, whose beginning is
         // text though. We then validate that even in this corner case buffer

tokio/src/io/stdout.rs

@@ -74,6 +74,20 @@ cfg_io_std! {
     /// to occur as a single write, so multiple threads writing data with
     /// [`write_all`] may result in interleaved output.
     ///
+    /// Note that unlike [`std::io::stdout`], each call to this `stdout()`
+    /// produces a new writer, so for example, this program does **not** flush stdout:
+    ///
+    /// ```no_run
+    /// # use tokio::io::AsyncWriteExt;
+    /// # #[tokio::main]
+    /// # async fn main() -> std::io::Result<()> {
+    /// tokio::io::stdout().write_all(b"aa").await?;
+    /// tokio::io::stdout().flush().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// [`std::io::stdout`]: std::io::stdout
     /// [`AsyncWrite`]: AsyncWrite
     /// [`write_all`]: crate::io::AsyncWriteExt::write_all()
     ///