chore: update tokio-util version number (#7215)

Introduce tokio_test::io::Builder::name to configure
name of the mock object, to include in panic messages.

Also show number of remaining actions or action index
in some cases to help debugging failed tests.

.circleci/config.yml

@@ -1,25 +0,0 @@
-version: 2.1        
-jobs:
-  test-arm:
-    machine:
-      image: default
-    resource_class: arm.medium
-    environment:
-      # Change to pin rust version
-      RUST_STABLE: stable
-    steps:
-      - checkout
-      - run:
-          name: Install Rust
-          command: |
-            curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs -o rustup.sh
-            chmod +x rustup.sh
-            ./rustup.sh -y --default-toolchain $RUST_STABLE
-            source "$HOME"/.cargo/env
-      # Only run Tokio tests
-      - run: cargo test --all-features -p tokio
-
-workflows:
-  ci:
-    jobs:
-      - test-arm

.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-1
+  image_family: freebsd-14-2
 env:
-  RUST_STABLE: 1.81
-  RUST_NIGHTLY: nightly-2024-05-05
+  RUST_STABLE: stable
+  RUST_NIGHTLY: nightly-2025-01-25
   RUSTFLAGS: -D warnings
 
 # Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the

.github/buildomat/README.md

@@ -12,7 +12,7 @@ reproduce the failure on other operating systems, don't worry! The
 [tokio-rs/illumos] team is responsible for maintaining Tokio's illumos support,
 and can be called on to assist contributors with illumos-specific issues. Please
 feel free to tag @tokio-rs/illumos to ask for help resolving build failures on
-illumos 
+illumos.
 
 [illumos]: https://www.illumos.org/
 [Buildomat]: https://github.com/oxidecomputer/buildomat

.github/workflows/ci.yml

@@ -16,9 +16,9 @@ env:
   RUSTUP_WINDOWS_PATH_ADD_BIN: 1
   # Change to specific Rust release to pin
   rust_stable: stable
-  rust_nightly: nightly-2024-05-05
+  rust_nightly: nightly-2025-01-25
   # Pin a specific miri version
-  rust_miri_nightly: nightly-2024-09-19
+  rust_miri_nightly: nightly-2025-01-25
   rust_clippy: '1.77'
   # When updating this, also update:
   # - README.md
@@ -47,10 +47,11 @@ jobs:
       - test-workspace-all-features
       - test-integration-tests-per-feature
       - test-parking_lot
-      - test-tracing-instrumentation
       - valgrind
       - test-unstable
-      - miri
+      - miri-lib
+      - miri-test
+      - miri-doc
       - asan
       - cross-check
       - cross-check-tier3
@@ -180,7 +181,7 @@ jobs:
         run: |
           set -euxo pipefail
           RUSTFLAGS="$RUSTFLAGS -C panic=abort -Zpanic-abort-tests" cargo nextest run --workspace --exclude tokio-macros --exclude tests-build --all-features --tests
-  
+
   test-integration-tests-per-feature:
     needs: basics
     name: Run integration tests for each feature
@@ -246,34 +247,6 @@ jobs:
       - name: Check tests with all features enabled
         run: cargo check --workspace --all-features --tests
 
-  test-tracing-instrumentation:
-    # These tests use the as-yet unpublished `tracing-mock` crate to test the
-    # tracing instrumentation present in Tokio. As such they are placed in
-    # their own test crate outside of the workspace.
-    needs: basics
-    name: test tokio instrumentation
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@stable
-        with:
-          toolchain: ${{ env.rust_stable }}
-      - name: Install cargo-nextest
-        uses: taiki-e/install-action@v2
-        with:
-          tool: cargo-nextest
-
-      - uses: Swatinem/rust-cache@v2
-
-      - name: test tracing-instrumentation
-        run: |
-          set -euxo pipefail
-          cargo nextest run
-        working-directory: tokio/tests/tracing-instrumentation
-        env:
-          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
-
   valgrind:
     name: valgrind
     needs: basics
@@ -283,7 +256,7 @@ jobs:
       - name: Install Rust ${{ env.rust_stable }}
         uses: dtolnay/rust-toolchain@stable
         with:
-            toolchain: ${{ env.rust_stable }}
+            toolchain: 1.82
 
       - name: Install Valgrind
         uses: taiki-e/install-action@valgrind
@@ -409,8 +382,8 @@ jobs:
           # the unstable cfg to RustDoc
           RUSTDOCFLAGS: --cfg tokio_unstable --cfg tokio_internal_mt_counters
 
-  miri:
-    name: miri
+  miri-lib:
+    name: miri-lib
     needs: basics
     runs-on: ubuntu-latest
     steps:
@@ -427,7 +400,49 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - name: miri
         run: |
-          cargo miri nextest run --features full --lib --tests --no-fail-fast
+          cargo miri nextest run --features full --lib --no-fail-fast
+        working-directory: tokio
+        env:
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
+
+  miri-test:
+    name: miri-test
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust ${{ env.rust_miri_nightly }}
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: ${{ env.rust_miri_nightly }}
+          components: miri
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-nextest
+      - uses: Swatinem/rust-cache@v2
+      - name: miri
+        run: |
+          cargo miri nextest run --features full --test '*' --no-fail-fast
+        working-directory: tokio
+        env:
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
+
+  miri-doc:
+    name: miri-doc
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust ${{ env.rust_miri_nightly }}
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: ${{ env.rust_miri_nightly }}
+          components: miri
+      - uses: Swatinem/rust-cache@v2
+      - name: miri-doc-test
+        run: |
+          cargo miri test --doc --all-features --no-fail-fast
         working-directory: tokio
         env:
           MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
@@ -496,6 +511,7 @@ jobs:
     strategy:
       matrix:
         target:
+          - name: x86_64-unknown-haiku
           - name: armv7-sony-vita-newlibeabihf
             exclude_features: "process,signal,rt-process-signal,full"
     steps:
@@ -516,15 +532,19 @@ jobs:
 
   cross-test-with-parking_lot:
     needs: basics
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.os }}
     strategy:
       matrix:
         include:
           - target: i686-unknown-linux-gnu
+            os: ubuntu-latest
             rustflags: --cfg tokio_taskdump
           - target: armv5te-unknown-linux-gnueabi
+            os: ubuntu-latest
           - target: armv7-unknown-linux-gnueabihf
+            os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
           - target: aarch64-unknown-linux-gnu
+            os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
             rustflags: --cfg tokio_taskdump
     steps:
       - uses: actions/checkout@v4
@@ -556,15 +576,19 @@ jobs:
 
   cross-test-without-parking_lot:
     needs: basics
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.os }}
     strategy:
       matrix:
         include:
           - target: i686-unknown-linux-gnu
+            os: ubuntu-latest
             rustflags: --cfg tokio_taskdump
           - target: armv5te-unknown-linux-gnueabi
+            os: ubuntu-latest
           - target: armv7-unknown-linux-gnueabihf
+            os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
           - target: aarch64-unknown-linux-gnu
+            os: ubuntu-22.04-arm # TODO: update to 24.04 when https://github.com/rust-lang/rust/issues/135867 solved
             rustflags: --cfg tokio_taskdump
     steps:
       - uses: actions/checkout@v4
@@ -768,7 +792,15 @@ jobs:
 
   docs:
     name: docs
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.run.os }}
+    strategy:
+      matrix:
+        run:
+          - os: windows-latest
+          - os: ubuntu-latest
+            RUSTFLAGS: --cfg tokio_taskdump
+            RUSTDOCFLAGS: --cfg tokio_taskdump
+
     steps:
       - uses: actions/checkout@v4
       - name: Install Rust ${{ env.rust_nightly }}
@@ -780,8 +812,8 @@ jobs:
         run: |
           cargo doc --lib --no-deps --all-features --document-private-items
         env:
-          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable --cfg tokio_taskdump
-          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable --cfg tokio_taskdump -Dwarnings
+          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable ${{ matrix.run.RUSTFLAGS }}
+          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings ${{ matrix.run.RUSTDOCFLAGS }}
 
   loom-compile:
     name: build loom tests
@@ -985,7 +1017,7 @@ jobs:
       - name: Install cargo-hack, wasmtime, and cargo-wasi
         uses: taiki-e/install-action@v2
         with:
-          tool: cargo-hack,wasmtime,cargo-wasi
+          tool: cargo-hack,wasmtime
 
       - uses: Swatinem/rust-cache@v2
       - name: WASI test tokio full
@@ -1011,9 +1043,12 @@ jobs:
 
       - name: test tests-integration --features wasi-rt
         # TODO: this should become: `cargo hack wasi test --each-feature`
-        run: cargo wasi test --test rt_yield --features wasi-rt
+        run: cargo test --target ${{ matrix.target }} --test rt_yield --features wasi-rt
         if: matrix.target == 'wasm32-wasip1'
         working-directory: tests-integration
+        env:
+          CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
+          RUSTFLAGS: -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: test tests-integration --features wasi-threads-rt
         run: cargo test --target ${{ matrix.target }} --features wasi-threads-rt
@@ -1051,23 +1086,6 @@ jobs:
         run: cargo check-external-types --all-features
         working-directory: tokio
 
-  check-unexpected-lints-cfgs:
-    name: check unexpected lints and cfgs
-    needs: basics
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_nightly }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-          toolchain: ${{ env.rust_nightly }}
-      - name: don't allow warnings
-        run: sed -i '/#!\[allow(unknown_lints, unexpected_cfgs)\]/d' */src/lib.rs */tests/*.rs
-      - name: check for unknown lints and cfgs
-        run: cargo check --all-features --tests
-        env:
-          RUSTFLAGS: -Dwarnings --check-cfg=cfg(loom,tokio_unstable,tokio_taskdump,fuzzing,mio_unsupported_force_poll_poll,tokio_internal_mt_counters,fs,tokio_no_parking_lot,tokio_no_tuning_tests) -Funexpected_cfgs -Funknown_lints
-
   check-fuzzing:
     name: check-fuzzing
     needs: basics
@@ -1105,23 +1123,57 @@ jobs:
       - uses: actions/checkout@v4
       - name: Make sure dictionary words are sorted and unique
         run: |
-          # `sed` removes the first line (number of words) and
-          # the last line (new line). 
-          # 
+          FILE="spellcheck.dic"
+
+          # Verify the first line is an integer.
+          first_line=$(head -n 1 "$FILE")
+          if ! [[ "$first_line" =~ ^[0-9]+$ ]]; then
+            echo "Error: The first line of $FILE must be an integer, but got: '$first_line'"
+            exit 1
+          fi
+          expected_count="$first_line"
+
+          # Check that the number of lines matches the integer.
+          # xargs (with no arguments) will strip leading/trailing whitespacefrom wc's output.
+          actual_count=$(sed '1d' "$FILE" | wc -l | xargs)
+          if [ "$expected_count" -ne "$actual_count" ]; then
+            echo "Error: The number of lines ($actual_count) does not match $expected_count."
+            exit 1
+          fi
+
+          # `sed` removes the first line (number of words).
+          #
           # `sort` makes sure everything in between is sorted
           # and contains no duplicates.
-          # 
+          #
           # Since `sort` is sensitive to locale, we set it
           # using LC_ALL to en_US.UTF8 to be consistent in different
           # environments.
 
           (
-            sed '1d; $d' spellcheck.dic | LC_ALL=en_US.UTF8 sort -uc
+            sed '1d' $FILE | LC_ALL=en_US.UTF8 sort -uc
           ) || {
             echo "Dictionary is not in sorted order. Correct order is:"
-            LC_ALL=en_US.UTF8 sort -u <(sed '1d; $d' spellcheck.dic)
+            LC_ALL=en_US.UTF8 sort -u <(sed '1d' $FILE)
             false
           }
       - name: Run cargo-spellcheck
-        run: cargo spellcheck --code 1
-
+        run: |
+          if ! cargo spellcheck --code 1
+          then
+              echo ''
+              echo ''
+              echo 'If this is a Rust method/type/variable name, then you should'
+              echo 'enclose it in backticks like this: `MyRustType`.'
+              echo ''
+              echo 'If this is a real word, then you can add it to spellcheck.dic'
+              exit 1
+          fi
+      - name: Detect trailing whitespace
+        run: |
+          if grep --exclude-dir=.git --exclude-dir=target -rne '\s$' .
+          then
+              echo ''
+              echo 'Please remove trailing whitespace from these lines.'
+              exit 1
+          fi

.github/workflows/loom.yml

@@ -91,7 +91,7 @@ jobs:
             toolchain: ${{ env.rust_stable }}
       - uses: Swatinem/rust-cache@v2
       - name: loom ${{ matrix.scope }}
-        run: cargo test --lib --release --features full -- $SCOPE
+        run: cargo test --lib --release --features full -- --nocapture $SCOPE
         working-directory: tokio
         env:
           SCOPE: ${{ matrix.scope }}
@@ -116,7 +116,7 @@ jobs:
             toolchain: ${{ env.rust_stable }}
       - uses: Swatinem/rust-cache@v2
       - name: loom ${{ matrix.scope }}
-        run: cargo test --lib --release --features full -- $SCOPE
+        run: cargo test --lib --release --features full -- --nocapture $SCOPE
         working-directory: tokio
         env:
           SCOPE: ${{ matrix.scope }}

.github/workflows/stress-test.yml

@@ -13,7 +13,7 @@ env:
   RUSTFLAGS: -Dwarnings
   RUST_BACKTRACE: 1
   # Change to specific Rust release to pin
-  rust_stable: stable
+  rust_stable: 1.82
 
 permissions:
   contents: read

CONTRIBUTING.md

@@ -196,12 +196,12 @@ LOOM_MAX_PREEMPTIONS=1 LOOM_MAX_BRANCHES=10000 RUSTFLAGS="--cfg loom -C debug_as
     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. 
+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
+    cargo +nightly miri test --features full --lib --tests
 ```
 
 ### Performing spellcheck on tokio codebase
@@ -216,8 +216,8 @@ cargo install --locked cargo-spellcheck
 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. 
+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

Cargo.toml

@@ -17,3 +17,16 @@ members = [
 
 [workspace.metadata.spellcheck]
 config = "spellcheck.toml"
+
+[workspace.lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = [
+  'cfg(fuzzing)',
+  'cfg(loom)',
+  'cfg(mio_unsupported_force_poll_poll)',
+  'cfg(tokio_allow_from_blocking_fd)',
+  'cfg(tokio_internal_mt_counters)',
+  'cfg(tokio_no_parking_lot)',
+  'cfg(tokio_no_tuning_tests)',
+  'cfg(tokio_taskdump)',
+  'cfg(tokio_unstable)',
+] }

README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.41.1", features = ["full"] }
+tokio = { version = "1.44.1", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -78,7 +78,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             loop {
                 let n = match socket.read(&mut buf).await {
                     // socket closed
-                    Ok(n) if n == 0 => return,
+                    Ok(0) => return,
                     Ok(n) => n,
                     Err(e) => {
                         eprintln!("failed to read from socket; err = {:?}", e);
@@ -205,8 +205,8 @@ works with the MSRV of that minor release.
 
 ## Release schedule
 
-Tokio doesn't follow a fixed release schedule, but we typically make one to two
-new minor releases each month. We make patch releases for bugfixes as necessary.
+Tokio doesn't follow a fixed release schedule, but we typically make one minor
+release each month. We make patch releases for bugfixes as necessary.
 
 ## Bug patching policy
 
@@ -216,9 +216,8 @@ warrants a patch release with a fix for the bug, it will be backported and
 released as a new patch release for each LTS minor version. Our current LTS
 releases are:
 
- * `1.32.x` - LTS release until September 2024. (MSRV 1.63)
- * `1.36.x` - LTS release until March 2025. (MSRV 1.63)
  * `1.38.x` - LTS release until July 2025. (MSRV 1.63)
+ * `1.43.x` - LTS release until March 2026. (MSRV 1.70)
 
 Each LTS release will continue to receive backported fixes for at least a year.
 If you wish to use a fixed minor release in your project, we recommend that you
@@ -228,7 +227,7 @@ To use a fixed minor version, you can specify the version with a tilde. For
 example, to specify that you wish to use the newest `1.32.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.32", features = [...] }
+tokio = { version = "~1.38", features = [...] }
 ```
 
 ### Previous LTS releases
@@ -238,6 +237,8 @@ tokio = { version = "~1.32", features = [...] }
  * `1.18.x` - LTS release until June 2023.
  * `1.20.x` - LTS release until September 2023.
  * `1.25.x` - LTS release until March 2024.
+ * `1.32.x` - LTS release until September 2024.
+ * `1.36.x` - LTS release until March 2025.
 
 ## License
 

benches/Cargo.toml

@@ -95,3 +95,6 @@ harness = false
 name = "time_timeout"
 path = "time_timeout.rs"
 harness = false
+
+[lints]
+workspace = true

benches/sync_mpsc.rs

@@ -37,7 +37,7 @@ fn create_medium<const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>) {
 fn send_data<T: Default, const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>, prefix: &str) {
     let rt = rt();
 
-    g.bench_function(format!("{}_{}", prefix, SIZE), |b| {
+    g.bench_function(format!("{prefix}_{SIZE}"), |b| {
         b.iter(|| {
             let (tx, mut rx) = mpsc::channel::<T>(SIZE);
 

deny.toml

@@ -9,7 +9,7 @@ allow = [
     "Apache-2.0",
 ]
 exceptions = [
-    { allow = ["Unicode-DFS-2016"], crate = "unicode-ident" },
+    { allow = ["Unicode-3.0", "Unicode-DFS-2016"], crate = "unicode-ident" },
 ]
 
 [bans]

examples/Cargo.toml

@@ -95,3 +95,6 @@ path = "named-pipe-multi-client.rs"
 [[example]]
 name = "dump"
 path = "dump.rs"
+
+[lints]
+workspace = true

examples/chat.rs

@@ -193,7 +193,7 @@ async fn process(
     // A client has connected, let's let everyone know.
     {
         let mut state = state.lock().await;
-        let msg = format!("{} has joined the chat", username);
+        let msg = format!("{username} has joined the chat");
         tracing::info!("{}", msg);
         state.broadcast(addr, &msg).await;
     }
@@ -210,7 +210,7 @@ async fn process(
                 // broadcast this message to the other users.
                 Some(Ok(msg)) => {
                     let mut state = state.lock().await;
-                    let msg = format!("{}: {}", username, msg);
+                    let msg = format!("{username}: {msg}");
 
                     state.broadcast(addr, &msg).await;
                 }
@@ -234,7 +234,7 @@ async fn process(
         let mut state = state.lock().await;
         state.peers.remove(&addr);
 
-        let msg = format!("{} has left the chat", username);
+        let msg = format!("{username} has left the chat");
         tracing::info!("{}", msg);
         state.broadcast(addr, &msg).await;
     }

examples/connect.rs

@@ -77,7 +77,7 @@ mod tcp {
                 //BytesMut into Bytes
                 Ok(i) => future::ready(Some(i.freeze())),
                 Err(e) => {
-                    println!("failed to read from socket; error={}", e);
+                    println!("failed to read from socket; error={e}");
                     future::ready(None)
                 }
             })

examples/dump.rs

@@ -1,5 +1,3 @@
-#![allow(unknown_lints, unexpected_cfgs)]
-
 //! This example demonstrates tokio's experimental task dumping functionality.
 //! This application deadlocks. Input CTRL+C to display traces of each task, or
 //! input CTRL+C twice within 1 second to quit.

examples/echo-udp.rs

@@ -38,7 +38,7 @@ impl Server {
             if let Some((size, peer)) = to_send {
                 let amt = socket.send_to(&buf[..size], &peer).await?;
 
-                println!("Echoed {}/{} bytes to {}", amt, size, peer);
+                println!("Echoed {amt}/{size} bytes to {peer}");
             }
 
             // If we're here then `to_send` is `None`, so we take a look for the

examples/echo.rs

@@ -40,7 +40,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     // connections. This TCP listener is bound to the address we determined
     // above and must be associated with an event loop.
     let listener = TcpListener::bind(&addr).await?;
-    println!("Listening on: {}", addr);
+    println!("Listening on: {addr}");
 
     loop {
         // Asynchronously wait for an inbound socket.

examples/print_each_packet.rs

@@ -75,7 +75,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // to our event loop. After the socket's created we inform that we're ready
     // to go and start accepting connections.
     let listener = TcpListener::bind(&addr).await?;
-    println!("Listening on: {}", addr);
+    println!("Listening on: {addr}");
 
     loop {
         // Asynchronously wait for an inbound socket.
@@ -96,8 +96,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             // The stream will return None once the client disconnects.
             while let Some(message) = framed.next().await {
                 match message {
-                    Ok(bytes) => println!("bytes: {:?}", bytes),
-                    Err(err) => println!("Socket closed with error: {:?}", err),
+                    Ok(bytes) => println!("bytes: {bytes:?}"),
+                    Err(err) => println!("Socket closed with error: {err:?}"),
                 }
             }
             println!("Socket received FIN packet and closed connection");

examples/proxy.rs

@@ -38,8 +38,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .nth(2)
         .unwrap_or_else(|| "127.0.0.1:8080".to_string());
 
-    println!("Listening on: {}", listen_addr);
-    println!("Proxying to: {}", server_addr);
+    println!("Listening on: {listen_addr}");
+    println!("Proxying to: {server_addr}");
 
     let listener = TcpListener::bind(listen_addr).await?;
 
@@ -50,7 +50,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
             copy_bidirectional(&mut inbound, &mut outbound)
                 .map(|r| {
                     if let Err(e) = r {
-                        println!("Failed to transfer; error={}", e);
+                        println!("Failed to transfer; error={e}");
                     }
                 })
                 .await

examples/tinydb.rs

@@ -90,7 +90,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .unwrap_or_else(|| "127.0.0.1:8080".to_string());
 
     let listener = TcpListener::bind(&addr).await?;
-    println!("Listening on: {}", addr);
+    println!("Listening on: {addr}");
 
     // Create the shared state of this server that will be shared amongst all
     // clients. We populate the initial database and then create the `Database`
@@ -131,11 +131,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
                                 let response = response.serialize();
 
                                 if let Err(e) = lines.send(response.as_str()).await {
-                                    println!("error on sending response; error = {:?}", e);
+                                    println!("error on sending response; error = {e:?}");
                                 }
                             }
                             Err(e) => {
-                                println!("error on decoding from socket; error = {:?}", e);
+                                println!("error on decoding from socket; error = {e:?}");
                             }
                         }
                     }
@@ -143,7 +143,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                     // The connection will be closed at this point as `lines.next()` has returned `None`.
                 });
             }
-            Err(e) => println!("error accepting socket; error = {:?}", e),
+            Err(e) => println!("error accepting socket; error = {e:?}"),
         }
     }
 }
@@ -162,7 +162,7 @@ fn handle_request(line: &str, db: &Arc<Database>) -> Response {
                 value: value.clone(),
             },
             None => Response::Error {
-                msg: format!("no key {}", key),
+                msg: format!("no key {key}"),
             },
         },
         Request::Set { key, value } => {
@@ -203,7 +203,7 @@ impl Request {
                     value: value.to_string(),
                 })
             }
-            Some(cmd) => Err(format!("unknown command: {}", cmd)),
+            Some(cmd) => Err(format!("unknown command: {cmd}")),
             None => Err("empty input".into()),
         }
     }
@@ -212,13 +212,13 @@ impl Request {
 impl Response {
     fn serialize(&self) -> String {
         match *self {
-            Response::Value { ref key, ref value } => format!("{} = {}", key, value),
+            Response::Value { ref key, ref value } => format!("{key} = {value}"),
             Response::Set {
                 ref key,
                 ref value,
                 ref previous,
-            } => format!("set {} = `{}`, previous: {:?}", key, value, previous),
-            Response::Error { ref msg } => format!("error: {}", msg),
+            } => format!("set {key} = `{value}`, previous: {previous:?}"),
+            Response::Error { ref msg } => format!("error: {msg}"),
         }
     }
 }

examples/tinyhttp.rs

@@ -31,13 +31,13 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .nth(1)
         .unwrap_or_else(|| "127.0.0.1:8080".to_string());
     let server = TcpListener::bind(&addr).await?;
-    println!("Listening on: {}", addr);
+    println!("Listening on: {addr}");
 
     loop {
         let (stream, _) = server.accept().await?;
         tokio::spawn(async move {
             if let Err(e) = process(stream).await {
-                println!("failed to process connection; error = {}", e);
+                println!("failed to process connection; error = {e}");
             }
         });
     }
@@ -159,7 +159,7 @@ impl Decoder for Http {
             let mut parsed_headers = [httparse::EMPTY_HEADER; 16];
             let mut r = httparse::Request::new(&mut parsed_headers);
             let status = r.parse(src).map_err(|e| {
-                let msg = format!("failed to parse http request: {:?}", e);
+                let msg = format!("failed to parse http request: {e:?}");
                 io::Error::new(io::ErrorKind::Other, msg)
             })?;
 

examples/udp-codec.rs

@@ -46,7 +46,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     // Run both futures simultaneously of `a` and `b` sending messages back and forth.
     match tokio::try_join!(a, b) {
-        Err(e) => println!("an error occurred; error = {:?}", e),
+        Err(e) => println!("an error occurred; error = {e:?}"),
         _ => println!("done!"),
     }
 

spellcheck.dic

@@ -1,4 +1,4 @@
-292
+300
 &
 +
 <
@@ -30,11 +30,13 @@
 450ms
 50ms
 8MB
+ABI
+accessors
 adaptor
 adaptors
 Adaptors
 AIO
-ambiant
+ambient
 amongst
 api
 APIs
@@ -75,8 +77,12 @@ datagrams
 deallocate
 deallocated
 Deallocates
+debuginfo
+decrement
 decrementing
+demangled
 dequeued
+dereferenced
 deregister
 deregistered
 deregistering
@@ -133,6 +139,7 @@ implementers
 implementor
 implementors
 incrementing
+inlining
 interoperate
 invariants
 Invariants
@@ -274,6 +281,7 @@ unparks
 Unparks
 unreceived
 unsafety
+unsets
 Unsets
 unsynchronized
 untrusted
@@ -291,4 +299,3 @@ Wakers
 wakeup
 wakeups
 workstealing
-

stress-test/Cargo.toml

@@ -13,3 +13,6 @@ tokio = { version = "1.0.0", path = "../tokio/", features = ["full"] }
 
 [dev-dependencies]
 rand = "0.8"
+
+[lints]
+workspace = true

tests-build/Cargo.toml

@@ -15,3 +15,6 @@ tokio = { version = "1.0.0", path = "../tokio", optional = true }
 
 [dev-dependencies]
 trybuild = "1.0"
+
+[lints]
+workspace = true

tests-build/README.md

@@ -6,5 +6,5 @@ To run all of the tests in this directory, run the following commands:
 cargo test --features full
 cargo test --features rt
 ```
-If one of the tests fail, you can pass `TRYBUILD=overwrite` to the `cargo test`
+If any of the tests fail, you can pass `TRYBUILD=overwrite` to the `cargo test`
 command that failed to have it regenerate the test output.

tests-build/tests/fail/macros_type_mismatch.rs

@@ -23,6 +23,40 @@ async fn extra_semicolon() -> Result<(), ()> {
     Ok(());
 }
 
+/// This test is a characterization test for the `?` operator.
+///
+/// See <https://github.com/tokio-rs/tokio/issues/6930#issuecomment-2572502517> for more details.
+///
+/// It should fail with a single error message about the return type of the function, but instead
+/// if fails with an extra error message due to the `?` operator being used within the async block
+/// rather than the original function.
+///
+/// ```text
+/// 28 |     None?;
+///    |         ^ cannot use the `?` operator in an async block that returns `()`
+/// ```
+#[tokio::main]
+async fn question_mark_operator_with_invalid_option() -> Option<()> {
+    None?;
+}
+
+/// This test is a characterization test for the `?` operator.
+///
+/// See <https://github.com/tokio-rs/tokio/issues/6930#issuecomment-2572502517> for more details.
+///
+/// It should fail with a single error message about the return type of the function, but instead
+/// if fails with an extra error message due to the `?` operator being used within the async block
+/// rather than the original function.
+///
+/// ```text
+/// 33 |     Ok(())?;
+///    |           ^ cannot use the `?` operator in an async block that returns `()`
+/// ```
+#[tokio::main]
+async fn question_mark_operator_with_invalid_result() -> Result<(), ()> {
+    Ok(())?;
+}
+
 // https://github.com/tokio-rs/tokio/issues/4635
 #[allow(redundant_semicolons)]
 #[rustfmt::skip]

tests-build/tests/fail/macros_type_mismatch.stderr

@@ -49,11 +49,68 @@ help: try adding an expression at the end of the block
 24 +     Ok(())
    |
 
-error[E0308]: mismatched types
-  --> tests/fail/macros_type_mismatch.rs:32:5
+error[E0277]: the `?` operator can only be used in an async block that returns `Result` or `Option` (or another type that implements `FromResidual`)
+  --> tests/fail/macros_type_mismatch.rs:40:9
    |
-30 | async fn issue_4635() {
+38 | #[tokio::main]
+   | -------------- this function should return `Result` or `Option` to accept `?`
+39 | async fn question_mark_operator_with_invalid_option() -> Option<()> {
+40 |     None?;
+   |         ^ cannot use the `?` operator in an async block that returns `()`
+   |
+   = help: the trait `FromResidual<Option<Infallible>>` is not implemented for `()`
+
+error[E0308]: mismatched types
+  --> tests/fail/macros_type_mismatch.rs:40:5
+   |
+39 | async fn question_mark_operator_with_invalid_option() -> Option<()> {
+   |                                                          ---------- expected `Option<()>` because of return type
+40 |     None?;
+   |     ^^^^^^ expected `Option<()>`, found `()`
+   |
+   = note:   expected enum `Option<()>`
+           found unit type `()`
+help: try adding an expression at the end of the block
+   |
+40 ~     None?;;
+41 +     None
+   |
+40 ~     None?;;
+41 +     Some(())
+   |
+
+error[E0277]: the `?` operator can only be used in an async block that returns `Result` or `Option` (or another type that implements `FromResidual`)
+  --> tests/fail/macros_type_mismatch.rs:57:11
+   |
+55 | #[tokio::main]
+   | -------------- this function should return `Result` or `Option` to accept `?`
+56 | async fn question_mark_operator_with_invalid_result() -> Result<(), ()> {
+57 |     Ok(())?;
+   |           ^ cannot use the `?` operator in an async block that returns `()`
+   |
+   = help: the trait `FromResidual<Result<Infallible, _>>` is not implemented for `()`
+
+error[E0308]: mismatched types
+  --> tests/fail/macros_type_mismatch.rs:57:5
+   |
+56 | async fn question_mark_operator_with_invalid_result() -> Result<(), ()> {
+   |                                                          -------------- expected `Result<(), ()>` because of return type
+57 |     Ok(())?;
+   |     ^^^^^^^^ expected `Result<(), ()>`, found `()`
+   |
+   = note:   expected enum `Result<(), ()>`
+           found unit type `()`
+help: try adding an expression at the end of the block
+   |
+57 ~     Ok(())?;;
+58 +     Ok(())
+   |
+
+error[E0308]: mismatched types
+  --> tests/fail/macros_type_mismatch.rs:66:5
+   |
+64 | async fn issue_4635() {
    |                      - help: try adding a return type: `-> i32`
-31 |     return 1;
-32 |     ;
+65 |     return 1;
+66 |     ;
    |     ^ expected `()`, found integer

tests-integration/Cargo.toml

@@ -61,3 +61,6 @@ tokio-test = { version = "0.4", path = "../tokio-test", optional = true }
 doc-comment = "0.3.1"
 futures = { version = "0.3.0", features = ["async-await"] }
 bytes = "1.0.0"
+
+[lints]
+workspace = true

tokio-macros/CHANGELOG.md

@@ -1,3 +1,9 @@
+# 2.5.0 (Jan 8th, 2025)
+
+- macros: suppress `clippy::needless_return` in `#[tokio::main]` ([#6874])
+
+[#6874]: https://github.com/tokio-rs/tokio/pull/6874
+
 # 2.4.0 (July 22nd, 2024)
 
 - msrv: increase MSRV to 1.70 ([#6645])

tokio-macros/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-macros"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-macros-1.x.y" git tag.
-version = "2.4.0"
+version = "2.5.0"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -31,3 +31,6 @@ tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
 
 [package.metadata.docs.rs]
 all-features = true
+
+[lints]
+workspace = true

tokio-macros/src/entry.rs

@@ -20,7 +20,7 @@ impl RuntimeFlavor {
             "single_thread" => Err("The single threaded runtime flavor is called `current_thread`.".to_string()),
             "basic_scheduler" => Err("The `basic_scheduler` runtime flavor has been renamed to `current_thread`.".to_string()),
             "threaded_scheduler" => Err("The `threaded_scheduler` runtime flavor has been renamed to `multi_thread`.".to_string()),
-            _ => Err(format!("No such runtime flavor `{}`. The runtime flavors are `current_thread` and `multi_thread`.", s)),
+            _ => Err(format!("No such runtime flavor `{s}`. The runtime flavors are `current_thread` and `multi_thread`.")),
         }
     }
 }
@@ -36,7 +36,7 @@ impl UnhandledPanic {
         match s {
             "ignore" => Ok(UnhandledPanic::Ignore),
             "shutdown_runtime" => Ok(UnhandledPanic::ShutdownRuntime),
-            _ => Err(format!("No such unhandled panic behavior `{}`. The unhandled panic behaviors are `ignore` and `shutdown_runtime`.", s)),
+            _ => Err(format!("No such unhandled panic behavior `{s}`. The unhandled panic behaviors are `ignore` and `shutdown_runtime`.")),
         }
     }
 
@@ -239,12 +239,12 @@ fn parse_int(int: syn::Lit, span: Span, field: &str) -> Result<usize, syn::Error
             Ok(value) => Ok(value),
             Err(e) => Err(syn::Error::new(
                 span,
-                format!("Failed to parse value of `{}` as integer: {}", field, e),
+                format!("Failed to parse value of `{field}` as integer: {e}"),
             )),
         },
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as integer.", field),
+            format!("Failed to parse value of `{field}` as integer."),
         )),
     }
 }
@@ -255,7 +255,7 @@ fn parse_string(int: syn::Lit, span: Span, field: &str) -> Result<String, syn::E
         syn::Lit::Verbatim(s) => Ok(s.to_string()),
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as string.", field),
+            format!("Failed to parse value of `{field}` as string."),
         )),
     }
 }
@@ -275,7 +275,7 @@ fn parse_path(lit: syn::Lit, span: Span, field: &str) -> Result<Path, syn::Error
         }
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as path.", field),
+            format!("Failed to parse value of `{field}` as path."),
         )),
     }
 }
@@ -285,7 +285,7 @@ fn parse_bool(bool: syn::Lit, span: Span, field: &str) -> Result<bool, syn::Erro
         syn::Lit::Bool(b) => Ok(b.value),
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as bool.", field),
+            format!("Failed to parse value of `{field}` as bool."),
         )),
     }
 }
@@ -342,8 +342,7 @@ fn build_config(
                     }
                     name => {
                         let msg = format!(
-                            "Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`",
-                            name,
+                            "Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`",
                         );
                         return Err(syn::Error::new_spanned(namevalue, msg));
                     }
@@ -358,21 +357,19 @@ fn build_config(
                 let msg = match name.as_str() {
                     "threaded_scheduler" | "multi_thread" => {
                         format!(
-                            "Set the runtime flavor with #[{}(flavor = \"multi_thread\")].",
-                            macro_name
+                            "Set the runtime flavor with #[{macro_name}(flavor = \"multi_thread\")]."
                         )
                     }
                     "basic_scheduler" | "current_thread" | "single_threaded" => {
                         format!(
-                            "Set the runtime flavor with #[{}(flavor = \"current_thread\")].",
-                            macro_name
+                            "Set the runtime flavor with #[{macro_name}(flavor = \"current_thread\")]."
                         )
                     }
                     "flavor" | "worker_threads" | "start_paused" | "crate" | "unhandled_panic" => {
-                        format!("The `{}` attribute requires an argument.", name)
+                        format!("The `{name}` attribute requires an argument.")
                     }
                     name => {
-                        format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.", name)
+                        format!("Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.")
                     }
                 };
                 return Err(syn::Error::new_spanned(path, msg));

tokio-macros/src/lib.rs

@@ -1,4 +1,3 @@
-#![allow(unknown_lints, unexpected_cfgs)]
 #![allow(clippy::needless_doctest_main)]
 #![warn(
     missing_debug_implementations,
@@ -211,7 +210,6 @@ use proc_macro::TokenStream;
 /// This option is only compatible with the `current_thread` runtime.
 ///
 /// ```no_run
-/// # #![allow(unknown_lints, unexpected_cfgs)]
 /// #[cfg(tokio_unstable)]
 /// #[tokio::main(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
 /// async fn main() {
@@ -226,7 +224,6 @@ use proc_macro::TokenStream;
 /// Equivalent code not using `#[tokio::main]`
 ///
 /// ```no_run
-/// # #![allow(unknown_lints, unexpected_cfgs)]
 /// #[cfg(tokio_unstable)]
 /// fn main() {
 ///     tokio::runtime::Builder::new_current_thread()
@@ -480,7 +477,6 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// This option is only compatible with the `current_thread` runtime.
 ///
 /// ```no_run
-/// # #![allow(unknown_lints, unexpected_cfgs)]
 /// #[cfg(tokio_unstable)]
 /// #[tokio::test(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
 /// async fn my_test() {
@@ -495,7 +491,6 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// Equivalent code not using `#[tokio::test]`
 ///
 /// ```no_run
-/// # #![allow(unknown_lints, unexpected_cfgs)]
 /// #[cfg(tokio_unstable)]
 /// #[test]
 /// fn my_test() {

tokio-macros/src/select.rs

@@ -11,7 +11,7 @@ pub(crate) fn declare_output_enum(input: TokenStream) -> TokenStream {
     };
 
     let variants = (0..branches)
-        .map(|num| Ident::new(&format!("_{}", num), Span::call_site()))
+        .map(|num| Ident::new(&format!("_{num}"), Span::call_site()))
         .collect::<Vec<_>>();
 
     // Use a bitfield to track which futures completed

tokio-stream/CHANGELOG.md

@@ -1,3 +1,14 @@
+# 0.1.17 (December 6th, 2024)
+
+- deps: fix dev-dependency on tokio-test ([#6931], [#7019])
+- stream: fix link on `Peekable` ([#6861])
+- sync: fix `Stream` link in broadcast docs ([#6873])
+
+[#6861]: https://github.com/tokio-rs/tokio/pull/6861
+[#6873]: https://github.com/tokio-rs/tokio/pull/6873
+[#6931]: https://github.com/tokio-rs/tokio/pull/6931
+[#7019]: https://github.com/tokio-rs/tokio/pull/7019
+
 # 0.1.16 (September 5th, 2024)
 
 This release bumps the MSRV of tokio-stream to 1.70.

tokio-stream/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-stream"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.16"
+version = "0.1.17"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -56,3 +56,6 @@ rustdoc-args = ["--cfg", "docsrs"]
 # This should allow `docsrs` to be read across projects, so that `tokio-stream`
 # can pick up stubbed types exported by `tokio`.
 rustc-args = ["--cfg", "docsrs"]
+
+[lints]
+workspace = true

tokio-stream/src/lib.rs

@@ -1,4 +1,3 @@
-#![allow(unknown_lints, unexpected_cfgs)]
 #![allow(
     clippy::cognitive_complexity,
     clippy::large_enum_variant,

tokio-stream/src/wrappers/broadcast.rs

@@ -10,6 +10,29 @@ use std::task::{ready, Context, Poll};
 
 /// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```
+/// use tokio::sync::broadcast;
+/// use tokio_stream::wrappers::BroadcastStream;
+/// use tokio_stream::StreamExt;
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> Result<(), tokio::sync::broadcast::error::SendError<u8>> {
+/// let (tx, rx) = broadcast::channel(16);
+/// tx.send(10)?;
+/// tx.send(20)?;
+/// # // prevent the doc test from hanging
+/// drop(tx);
+///
+/// let mut stream = BroadcastStream::new(rx);
+/// assert_eq!(stream.next().await, Some(Ok(10)));
+/// assert_eq!(stream.next().await, Some(Ok(20)));
+/// assert_eq!(stream.next().await, None);
+/// # Ok(())
+/// # }
+/// ```
+///
 /// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
 /// [`Stream`]: trait@futures_core::Stream
 #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]

tokio-stream/src/wrappers/interval.rs

@@ -5,6 +5,26 @@ use tokio::time::{Instant, Interval};
 
 /// A wrapper around [`Interval`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```
+/// use tokio::time::{Duration, Instant, interval};
+/// use tokio_stream::wrappers::IntervalStream;
+/// use tokio_stream::StreamExt;
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() {
+/// let start = Instant::now();
+/// let interval = interval(Duration::from_millis(10));
+/// let mut stream = IntervalStream::new(interval);
+/// for _ in 0..3 {
+///     if let Some(instant) = stream.next().await {
+///         println!("elapsed: {:.1?}", instant.duration_since(start));
+///     }
+/// }
+/// # }
+/// ```
+///
 /// [`Interval`]: struct@tokio::time::Interval
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/lines.rs

@@ -8,6 +8,24 @@ use tokio::io::{AsyncBufRead, Lines};
 pin_project! {
     /// A wrapper around [`tokio::io::Lines`] that implements [`Stream`].
     ///
+    /// # Example
+    ///
+    /// ```
+    /// use tokio::io::AsyncBufReadExt;
+    /// use tokio_stream::wrappers::LinesStream;
+    /// use tokio_stream::StreamExt;
+    ///
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() -> std::io::Result<()> {
+    /// let input = b"Hello\nWorld\n";
+    /// let mut stream = LinesStream::new(input.lines());
+    /// while let Some(line) = stream.next().await {
+    ///     println!("{}", line?);
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
     /// [`tokio::io::Lines`]: struct@tokio::io::Lines
     /// [`Stream`]: trait@crate::Stream
     #[derive(Debug)]

tokio-stream/src/wrappers/mpsc_bounded.rs

@@ -5,6 +5,29 @@ use tokio::sync::mpsc::Receiver;
 
 /// A wrapper around [`tokio::sync::mpsc::Receiver`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```
+/// use tokio::sync::mpsc;
+/// use tokio_stream::wrappers::ReceiverStream;
+/// use tokio_stream::StreamExt;
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> Result<(), tokio::sync::mpsc::error::SendError<u8>> {
+/// let (tx, rx) = mpsc::channel(2);
+/// tx.send(10).await?;
+/// tx.send(20).await?;
+/// # // prevent the doc test from hanging
+/// drop(tx);
+///
+/// let mut stream = ReceiverStream::new(rx);
+/// assert_eq!(stream.next().await, Some(10));
+/// assert_eq!(stream.next().await, Some(20));
+/// assert_eq!(stream.next().await, None);
+/// # Ok(())
+/// # }
+/// ```
+///
 /// [`tokio::sync::mpsc::Receiver`]: struct@tokio::sync::mpsc::Receiver
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/mpsc_unbounded.rs

@@ -5,6 +5,29 @@ use tokio::sync::mpsc::UnboundedReceiver;
 
 /// A wrapper around [`tokio::sync::mpsc::UnboundedReceiver`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```
+/// use tokio::sync::mpsc;
+/// use tokio_stream::wrappers::UnboundedReceiverStream;
+/// use tokio_stream::StreamExt;
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> Result<(), tokio::sync::mpsc::error::SendError<u8>> {
+/// let (tx, rx) = mpsc::unbounded_channel();
+/// tx.send(10)?;
+/// tx.send(20)?;
+/// # // prevent the doc test from hanging
+/// drop(tx);
+///
+/// let mut stream = UnboundedReceiverStream::new(rx);
+/// assert_eq!(stream.next().await, Some(10));
+/// assert_eq!(stream.next().await, Some(20));
+/// assert_eq!(stream.next().await, None);
+/// # Ok(())
+/// # }
+/// ```
+///
 /// [`tokio::sync::mpsc::UnboundedReceiver`]: struct@tokio::sync::mpsc::UnboundedReceiver
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/read_dir.rs

@@ -6,6 +6,24 @@ use tokio::fs::{DirEntry, ReadDir};
 
 /// A wrapper around [`tokio::fs::ReadDir`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```
+/// use tokio::fs::read_dir;
+/// use tokio_stream::{StreamExt, wrappers::ReadDirStream};
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+/// let dirs = read_dir(".").await?;
+/// let mut dirs = ReadDirStream::new(dirs);
+/// while let Some(dir) = dirs.next().await {
+///     let dir = dir?;
+///     println!("{}", dir.path().display());
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
 /// [`tokio::fs::ReadDir`]: struct@tokio::fs::ReadDir
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/signal_unix.rs

@@ -5,6 +5,22 @@ use tokio::signal::unix::Signal;
 
 /// A wrapper around [`Signal`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```no_run
+/// use tokio::signal::unix::{signal, SignalKind};
+/// use tokio_stream::{StreamExt, wrappers::SignalStream};
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+/// let signals = signal(SignalKind::hangup())?;
+/// let mut stream = SignalStream::new(signals);
+/// while stream.next().await.is_some() {
+///     println!("hangup signal received");
+/// }
+/// # Ok(())
+/// # }
+/// ```
 /// [`Signal`]: struct@tokio::signal::unix::Signal
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/signal_windows.rs

@@ -7,6 +7,23 @@ use tokio::signal::windows::{CtrlBreak, CtrlC};
 ///
 /// [`CtrlC`]: struct@tokio::signal::windows::CtrlC
 /// [`Stream`]: trait@crate::Stream
+///
+/// # Example
+///
+/// ```no_run
+/// use tokio::signal::windows::ctrl_c;
+/// use tokio_stream::{StreamExt, wrappers::CtrlCStream};
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+/// let signals = ctrl_c()?;
+/// let mut stream = CtrlCStream::new(signals);
+/// while stream.next().await.is_some() {
+///     println!("ctrl-c received");
+/// }
+/// # Ok(())
+/// # }
+/// ```
 #[derive(Debug)]
 #[cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]
 pub struct CtrlCStream {
@@ -47,6 +64,23 @@ impl AsMut<CtrlC> for CtrlCStream {
 
 /// A wrapper around [`CtrlBreak`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```no_run
+/// use tokio::signal::windows::ctrl_break;
+/// use tokio_stream::{StreamExt, wrappers::CtrlBreakStream};
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+/// let signals = ctrl_break()?;
+/// let mut stream = CtrlBreakStream::new(signals);
+/// while stream.next().await.is_some() {
+///     println!("ctrl-break received");
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
 /// [`CtrlBreak`]: struct@tokio::signal::windows::CtrlBreak
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/split.rs

@@ -8,6 +8,24 @@ use tokio::io::{AsyncBufRead, Split};
 pin_project! {
     /// A wrapper around [`tokio::io::Split`] that implements [`Stream`].
     ///
+    /// # Example
+    ///
+    /// ```
+    /// use tokio::io::AsyncBufReadExt;
+    /// use tokio_stream::{StreamExt, wrappers::SplitStream};
+    ///
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() -> std::io::Result<()> {
+    /// let input = "Hello\nWorld\n".as_bytes();
+    /// let lines = AsyncBufReadExt::split(input, b'\n');
+    ///
+    /// let mut stream = SplitStream::new(lines);
+    /// while let Some(line) = stream.next().await {
+    ///     println!("length = {}", line?.len())
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
     /// [`tokio::io::Split`]: struct@tokio::io::Split
     /// [`Stream`]: trait@crate::Stream
     #[derive(Debug)]

tokio-stream/src/wrappers/tcp_listener.rs

@@ -6,6 +6,33 @@ use tokio::net::{TcpListener, TcpStream};
 
 /// A wrapper around [`TcpListener`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// Accept connections from both IPv4 and IPv6 listeners in the same loop:
+///
+/// ```no_run
+/// use std::net::{Ipv4Addr, Ipv6Addr};
+///
+/// use tokio::net::TcpListener;
+/// use tokio_stream::{StreamExt, wrappers::TcpListenerStream};
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+/// let ipv4_listener = TcpListener::bind((Ipv6Addr::LOCALHOST, 8080)).await?;
+/// let ipv6_listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 8080)).await?;
+/// let ipv4_connections = TcpListenerStream::new(ipv4_listener);
+/// let ipv6_connections = TcpListenerStream::new(ipv6_listener);
+///
+/// let mut connections = ipv4_connections.chain(ipv6_connections);
+/// while let Some(tcp_stream) = connections.next().await {
+///     let stream = tcp_stream?;
+///     let peer_addr = stream.peer_addr()?;
+///     println!("accepted connection; peer address = {peer_addr}");
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
 /// [`TcpListener`]: struct@tokio::net::TcpListener
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/src/wrappers/unix_listener.rs

@@ -6,6 +6,25 @@ use tokio::net::{UnixListener, UnixStream};
 
 /// A wrapper around [`UnixListener`] that implements [`Stream`].
 ///
+/// # Example
+///
+/// ```no_run
+/// use tokio::net::UnixListener;
+/// use tokio_stream::{StreamExt, wrappers::UnixListenerStream};
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+/// let listener = UnixListener::bind("/tmp/sock")?;
+/// let mut incoming = UnixListenerStream::new(listener);
+///
+/// while let Some(stream) = incoming.next().await {
+///     let stream = stream?;
+///     let peer_addr = stream.peer_addr()?;
+///     println!("Accepted connection from: {peer_addr:?}");
+/// }
+/// # Ok(())
+/// # }
+/// ```
 /// [`UnixListener`]: struct@tokio::net::UnixListener
 /// [`Stream`]: trait@crate::Stream
 #[derive(Debug)]

tokio-stream/tests/stream_chain.rs

@@ -49,6 +49,7 @@ where
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // Block on https://github.com/tokio-rs/tokio/issues/6860
 async fn pending_first() {
     let (tx1, rx1) = mpsc::unbounded_channel_stream();
     let (tx2, rx2) = mpsc::unbounded_channel_stream();

tokio-test/Cargo.toml

@@ -19,9 +19,6 @@ 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" }
-async-stream = "0.3.3"
-
-bytes = "1.0.0"
 futures-core = "0.3.0"
 
 [dev-dependencies]
@@ -30,3 +27,6 @@ futures-util = "0.3.0"
 
 [package.metadata.docs.rs]
 all-features = true
+
+[lints]
+workspace = true

tokio-test/src/io.rs

@@ -52,6 +52,7 @@ pub struct Handle {
 pub struct Builder {
     // Sequence of actions for the Mock to take
     actions: VecDeque<Action>,
+    name: String,
 }
 
 #[derive(Debug, Clone)]
@@ -71,6 +72,7 @@ struct Inner {
     sleep: Option<Pin<Box<Sleep>>>,
     read_wait: Option<Waker>,
     rx: UnboundedReceiverStream<Action>,
+    name: String,
 }
 
 impl Builder {
@@ -127,6 +129,12 @@ impl Builder {
         self
     }
 
+    /// Set name of the mock IO object to include in panic messages and debug output
+    pub fn name(&mut self, name: impl Into<String>) -> &mut Self {
+        self.name = name.into();
+        self
+    }
+
     /// Build a `Mock` value according to the defined script.
     pub fn build(&mut self) -> Mock {
         let (mock, _) = self.build_with_handle();
@@ -135,7 +143,7 @@ impl Builder {
 
     /// Build a `Mock` value paired with a handle
     pub fn build_with_handle(&mut self) -> (Mock, Handle) {
-        let (inner, handle) = Inner::new(self.actions.clone());
+        let (inner, handle) = Inner::new(self.actions.clone(), self.name.clone());
 
         let mock = Mock { inner };
 
@@ -184,7 +192,7 @@ impl Handle {
 }
 
 impl Inner {
-    fn new(actions: VecDeque<Action>) -> (Inner, Handle) {
+    fn new(actions: VecDeque<Action>, name: String) -> (Inner, Handle) {
         let (tx, rx) = mpsc::unbounded_channel();
 
         let rx = UnboundedReceiverStream::new(rx);
@@ -195,6 +203,7 @@ impl Inner {
             read_wait: None,
             rx,
             waiting: None,
+            name,
         };
 
         let handle = Handle { tx };
@@ -256,7 +265,7 @@ impl Inner {
                 Action::Write(ref mut expect) => {
                     let n = cmp::min(src.len(), expect.len());
 
-                    assert_eq!(&src[..n], &expect[..n]);
+                    assert_eq!(&src[..n], &expect[..n], "name={} i={}", self.name, i);
 
                     // Drop data that was matched
                     expect.drain(..n);
@@ -418,7 +427,7 @@ impl AsyncWrite for Mock {
                         self.inner.actions.push_back(action);
                     }
                     Poll::Ready(None) => {
-                        panic!("unexpected write");
+                        panic!("unexpected write {}", self.pmsg());
                     }
                 }
             }
@@ -429,7 +438,7 @@ impl AsyncWrite for Mock {
                         let until = Instant::now() + rem;
                         self.inner.sleep = Some(Box::pin(time::sleep_until(until)));
                     } else {
-                        panic!("unexpected WouldBlock");
+                        panic!("unexpected WouldBlock {}", self.pmsg());
                     }
                 }
                 Ok(0) => {
@@ -445,7 +454,7 @@ impl AsyncWrite for Mock {
                             continue;
                         }
                         None => {
-                            panic!("unexpected write");
+                            panic!("unexpected write {}", self.pmsg());
                         }
                     }
                 }
@@ -475,8 +484,16 @@ impl Drop for Mock {
         }
 
         self.inner.actions.iter().for_each(|a| match a {
-            Action::Read(data) => assert!(data.is_empty(), "There is still data left to read."),
-            Action::Write(data) => assert!(data.is_empty(), "There is still data left to write."),
+            Action::Read(data) => assert!(
+                data.is_empty(),
+                "There is still data left to read. {}",
+                self.pmsg()
+            ),
+            Action::Write(data) => assert!(
+                data.is_empty(),
+                "There is still data left to write. {}",
+                self.pmsg()
+            ),
             _ => (),
         });
     }
@@ -505,6 +522,33 @@ fn is_task_ctx() -> bool {
 
 impl fmt::Debug for Inner {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "Inner {{...}}")
+        if self.name.is_empty() {
+            write!(f, "Inner {{...}}")
+        } else {
+            write!(f, "Inner {{name={}, ...}}", self.name)
+        }
+    }
+}
+
+struct PanicMsgSnippet<'a>(&'a Inner);
+
+impl<'a> fmt::Display for PanicMsgSnippet<'a> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if self.0.name.is_empty() {
+            write!(f, "({} actions remain)", self.0.actions.len())
+        } else {
+            write!(
+                f,
+                "(name {}, {} actions remain)",
+                self.0.name,
+                self.0.actions.len()
+            )
+        }
+    }
+}
+
+impl Mock {
+    fn pmsg(&self) -> PanicMsgSnippet<'_> {
+        PanicMsgSnippet(&self.inner)
     }
 }

tokio-test/src/lib.rs

@@ -1,4 +1,3 @@
-#![allow(unknown_lints, unexpected_cfgs)]
 #![warn(
     missing_debug_implementations,
     missing_docs,

tokio-test/src/stream_mock.rs

@@ -161,8 +161,7 @@ impl<T: Unpin> Drop for StreamMock<T> {
 
         assert!(
             undropped_count == 0,
-            "StreamMock was dropped before all actions were consumed, {} actions were not consumed",
-            undropped_count
+            "StreamMock was dropped before all actions were consumed, {undropped_count} actions were not consumed"
         );
     }
 }

tokio-test/tests/io.rs

@@ -34,7 +34,7 @@ async fn read_error() {
     match mock.read(&mut buf).await {
         Err(error) => {
             assert_eq!(error.kind(), io::ErrorKind::Other);
-            assert_eq!("cruel", format!("{}", error));
+            assert_eq!("cruel", format!("{error}"));
         }
         Ok(_) => panic!("error not received"),
     }
@@ -87,7 +87,7 @@ async fn write_error() {
     match mock.write_all(b"whoa").await {
         Err(error) => {
             assert_eq!(error.kind(), io::ErrorKind::Other);
-            assert_eq!("cruel", format!("{}", error));
+            assert_eq!("cruel", format!("{error}"));
         }
         Ok(_) => panic!("error not received"),
     }

tokio-util/CHANGELOG.md

@@ -1,3 +1,49 @@
+# 0.7.14 (March 12th, 2025)
+
+### Added
+
+- io: add `get_ref` and `get_mut` for `SyncIoBridge` ([#7128])
+- io: add `read_exact_arc` ([#7165])
+- sync: add `CancellationToken::run_until_cancelled_owned` ([#7081])
+
+### Changed
+
+- codec: optimize buffer reserve for `AnyDelimiterCodec::encode` ([#7188])
+- either: enable `Either` to use underlying `AsyncWrite` implementation ([#7025])
+
+### Fixed
+
+- codec: fix typo in API docs ([#7044])
+- util: fix example in `StreamReader` docs ([#7167])
+
+### Documented
+
+- io: add docs for `SyncIoBridge` with examples and alternatives ([#6815])
+
+### Internal
+
+- io: clean up buffer casts ([#7142])
+- task: run `spawn_pinned` tests with miri ([#7023])
+
+[#6815]: https://github.com/tokio-rs/tokio/pull/6815
+[#7023]: https://github.com/tokio-rs/tokio/pull/7023
+[#7025]: https://github.com/tokio-rs/tokio/pull/7025
+[#7044]: https://github.com/tokio-rs/tokio/pull/7044
+[#7081]: https://github.com/tokio-rs/tokio/pull/7081
+[#7128]: https://github.com/tokio-rs/tokio/pull/7128
+[#7142]: https://github.com/tokio-rs/tokio/pull/7142
+[#7165]: https://github.com/tokio-rs/tokio/pull/7165
+[#7167]: https://github.com/tokio-rs/tokio/pull/7167
+[#7188]: https://github.com/tokio-rs/tokio/pull/7188
+
+# 0.7.13 (December 4th, 2024)
+
+### Fixed
+
+- codec: fix incorrect handling of invalid utf-8 in `LinesCodec::decode_eof` ([#7011])
+
+[#7011]: https://github.com/tokio-rs/tokio/pull/7011
+
 # 0.7.12 (September 5th, 2024)
 
 This release bumps the MSRV to 1.70. ([#6645])
@@ -127,7 +173,7 @@ This release contains one performance improvement:
 [#5630]: https://github.com/tokio-rs/tokio/pull/5630
 [#5632]: https://github.com/tokio-rs/tokio/pull/5632
 
-# 0.7.7 (February 12, 2023)
+# 0.7.7 (February 12th, 2023)
 
 This release reverts the removal of the `Encoder` bound on the `FramedParts`
 constructor from [#5280] since it turned out to be a breaking change. ([#5450])

tokio-util/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-util"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.12"
+version = "0.7.14"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -35,7 +35,7 @@ __docs_rs = ["futures-util"]
 
 [dependencies]
 tokio = { version = "1.28.0", path = "../tokio", features = ["sync"] }
-bytes = "1.0.0"
+bytes = "1.5.0"
 futures-core = "0.3.0"
 futures-sink = "0.3.0"
 futures-io = { version = "0.3.0", optional = true }
@@ -68,3 +68,6 @@ rustc-args = ["--cfg", "docsrs", "--cfg", "tokio_unstable"]
 
 [package.metadata.playground]
 features = ["full"]
+
+[lints]
+workspace = true

tokio-util/src/codec/any_delimiter_codec.rs

@@ -217,7 +217,7 @@ where
 
     fn encode(&mut self, chunk: T, buf: &mut BytesMut) -> Result<(), AnyDelimiterCodecError> {
         let chunk = chunk.as_ref();
-        buf.reserve(chunk.len() + 1);
+        buf.reserve(chunk.len() + self.sequence_writer.len());
         buf.put(chunk.as_bytes());
         buf.put(self.sequence_writer.as_ref());
 
@@ -249,7 +249,7 @@ impl fmt::Display for AnyDelimiterCodecError {
             AnyDelimiterCodecError::MaxChunkLengthExceeded => {
                 write!(f, "max chunk length exceeded")
             }
-            AnyDelimiterCodecError::Io(e) => write!(f, "{}", e),
+            AnyDelimiterCodecError::Io(e) => write!(f, "{e}"),
         }
     }
 }

tokio-util/src/codec/lines_codec.rs

@@ -169,6 +169,7 @@ impl Decoder for LinesCodec {
         Ok(match self.decode(buf)? {
             Some(frame) => Some(frame),
             None => {
+                self.next_index = 0;
                 // No terminating newline - return remaining data, if any
                 if buf.is_empty() || buf == &b"\r"[..] {
                     None
@@ -176,7 +177,6 @@ impl Decoder for LinesCodec {
                     let line = buf.split_to(buf.len());
                     let line = without_carriage_return(&line);
                     let line = utf8(line)?;
-                    self.next_index = 0;
                     Some(line.to_string())
                 }
             }
@@ -218,7 +218,7 @@ impl fmt::Display for LinesCodecError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             LinesCodecError::MaxLineLengthExceeded => write!(f, "max line length exceeded"),
-            LinesCodecError::Io(e) => write!(f, "{}", e),
+            LinesCodecError::Io(e) => write!(f, "{e}"),
         }
     }
 }

tokio-util/src/codec/mod.rs

@@ -224,7 +224,7 @@
 //! The main method on the `Encoder` trait is the [`encode`] method. This method
 //! takes an item that is being written, and a buffer to write the item to. The
 //! buffer may already contain data, and in this case, the encoder should append
-//! the new frame the to buffer rather than overwrite the existing data.
+//! the new frame to the buffer rather than overwrite the existing data.
 //!
 //! It is guaranteed that, from one call to `encode` to another, the provided
 //! buffer will contain the exact same data as before, except that some of the

tokio-util/src/either.rs

@@ -150,6 +150,21 @@ where
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
         delegate_call!(self.poll_shutdown(cx))
     }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<std::result::Result<usize, std::io::Error>> {
+        delegate_call!(self.poll_write_vectored(cx, bufs))
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        match self {
+            Self::Left(l) => l.is_write_vectored(),
+            Self::Right(r) => r.is_write_vectored(),
+        }
+    }
 }
 
 impl<L, R> futures_core::stream::Stream for Either<L, R>

tokio-util/src/io/mod.rs

@@ -18,6 +18,9 @@ mod sink_writer;
 mod stream_reader;
 
 cfg_io_util! {
+    mod read_arc;
+    pub use self::read_arc::read_exact_arc;
+
     mod sync_bridge;
     pub use self::sync_bridge::SyncIoBridge;
 }

tokio-util/src/io/read_arc.rs

@@ -0,0 +1,44 @@
+use std::io;
+use std::mem::MaybeUninit;
+use std::sync::Arc;
+use tokio::io::{AsyncRead, AsyncReadExt};
+
+/// Read data from an `AsyncRead` into an `Arc`.
+///
+/// This uses `Arc::new_uninit_slice` and reads into the resulting uninitialized `Arc`.
+///
+/// # Example
+///
+/// ```
+/// # #[tokio::main]
+/// # async fn main() -> std::io::Result<()> {
+/// use tokio_util::io::read_exact_arc;
+///
+/// let read = tokio::io::repeat(42);
+///
+/// let arc = read_exact_arc(read, 4).await?;
+///
+/// assert_eq!(&arc[..], &[42; 4]);
+/// # Ok(())
+/// # }
+/// ```
+pub async fn read_exact_arc<R: AsyncRead>(read: R, len: usize) -> io::Result<Arc<[u8]>> {
+    tokio::pin!(read);
+    // TODO(MSRV 1.82): When bumping MSRV, switch to `Arc::new_uninit_slice(len)`. The following is
+    // equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
+    let arc: Arc<[MaybeUninit<u8>]> = (0..len).map(|_| MaybeUninit::uninit()).collect();
+    // TODO(MSRV future): Use `Arc::get_mut_unchecked` once it's stabilized.
+    // SAFETY: We're the only owner of the `Arc`, and we keep the `Arc` valid throughout this loop
+    // as we write through this reference.
+    let mut buf = unsafe { &mut *(Arc::as_ptr(&arc) as *mut [MaybeUninit<u8>]) };
+    while !buf.is_empty() {
+        if read.read_buf(&mut buf).await? == 0 {
+            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
+        }
+    }
+    // TODO(MSRV 1.82): When bumping MSRV, switch to `arc.assume_init()`. The following is
+    // equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
+    // SAFETY: This changes `[MaybeUninit<u8>]` to `[u8]`, and we've initialized all the bytes in
+    // the loop above.
+    Ok(unsafe { Arc::from_raw(Arc::into_raw(arc) as *const [u8]) })
+}

tokio-util/src/io/stream_reader.rs

@@ -169,7 +169,7 @@ where
     /// Convert a stream of byte chunks into an [`AsyncRead`].
     ///
     /// The item should be a [`Result`] with the ok variant being something that
-    /// implements the [`Buf`] trait (e.g. `Vec<u8>` or `Bytes`). The error
+    /// implements the [`Buf`] trait (e.g. `Cursor<Vec<u8>>` or `Bytes`). The error
     /// should be convertible into an [io error].
     ///
     /// [`Result`]: std::result::Result

tokio-util/src/io/sync_bridge.rs

@@ -5,7 +5,258 @@ use tokio::io::{
 };
 
 /// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
-/// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
+/// a [`tokio::io::AsyncWrite`] synchronously as a [`std::io::Write`].
+///
+/// # Alternatives
+///
+/// In many cases, there are better alternatives to using `SyncIoBridge`, especially
+/// if you want to avoid blocking the async runtime. Consider the following scenarios:
+///
+/// When hashing data, using `SyncIoBridge` can lead to suboptimal performance and
+/// might not fully leverage the async capabilities of the system.
+///
+/// ### Why It Matters:
+///
+/// `SyncIoBridge` allows you to use asynchronous I/O operations in an synchronous
+/// context by blocking the current thread. However, this can be inefficient because:
+/// - **Inefficient Resource Usage**: `SyncIoBridge` takes up an entire OS thread,
+///   which is inefficient compared to asynchronous code that can multiplex many
+///   tasks on a single thread.
+/// - **Thread Pool Saturation**: Excessive use of `SyncIoBridge` can exhaust the
+///   async runtime's thread pool, reducing the number of threads available for
+///   other tasks and impacting overall performance.
+/// - **Missed Concurrency Benefits**: By using synchronous operations with
+///   `SyncIoBridge`, you lose the ability to interleave tasks efficiently,
+///   which is a key advantage of asynchronous programming.
+///
+/// ## Example 1: Hashing Data
+///
+/// The use of `SyncIoBridge` is unnecessary when hashing data. Instead, you can
+/// process the data asynchronously by reading it into memory, which avoids blocking
+/// the async runtime.
+///
+/// There are two strategies for avoiding `SyncIoBridge` when hashing data. When
+/// the data fits into memory, the easiest is to read the data into a `Vec<u8>`
+/// and hash it:
+///
+/// Explanation: This example demonstrates how to asynchronously read data from a
+/// reader into memory and hash it using a synchronous hashing function. The
+/// `SyncIoBridge` is avoided, ensuring that the async runtime is not blocked.
+/// ```rust
+/// use tokio::io::AsyncReadExt;
+/// use tokio::io::AsyncRead;
+/// use std::io::Cursor;
+/// # mod blake3 { pub fn hash(_: &[u8]) {} }
+///
+/// async fn hash_contents(mut reader: impl AsyncRead + Unpin) -> Result<(), std::io::Error> {
+///    // Read all data from the reader into a Vec<u8>.
+///    let mut data = Vec::new();
+///    reader.read_to_end(&mut data).await?;
+///
+///    // Hash the data using the blake3 hashing function.
+///    let hash = blake3::hash(&data);
+///
+///    Ok(hash)
+///}
+///
+/// #[tokio::main]
+/// async fn main() -> Result<(), std::io::Error> {
+///     // Example: In-memory data.
+///     let data = b"Hello, world!"; // A byte slice.
+///     let reader = Cursor::new(data); // Create an in-memory AsyncRead.
+///     hash_contents(reader).await
+/// }
+/// ```
+///
+/// When the data doesn't fit into memory, the hashing library will usually
+/// provide a `hasher` that you can repeatedly call `update` on to hash the data
+/// one chunk at the time.
+///
+/// Explanation: This example demonstrates how to asynchronously stream data in
+/// chunks for hashing. Each chunk is read asynchronously, and the hash is updated
+/// incrementally. This avoids blocking and improves performance over using
+/// `SyncIoBridge`.
+///
+/// ```rust
+/// use tokio::io::AsyncReadExt;
+/// use tokio::io::AsyncRead;
+/// use std::io::Cursor;
+/// # struct Hasher;
+/// # impl Hasher { pub fn update(&mut self, _: &[u8]) {} pub fn finalize(&self) {} }
+///
+/// /// Asynchronously streams data from an async reader, processes it in chunks,
+/// /// and hashes the data incrementally.
+/// async fn hash_stream(mut reader: impl AsyncRead + Unpin, mut hasher: Hasher) -> Result<(), std::io::Error> {
+///    // Create a buffer to read data into, sized for performance.
+///    let mut data = vec![0; 64 * 1024];
+///    loop {
+///        // Read data from the reader into the buffer.
+///        let len = reader.read(&mut data).await?;
+///        if len == 0 { break; } // Exit loop if no more data.
+///
+///        // Update the hash with the data read.
+///        hasher.update(&data[..len]);
+///    }
+///
+///    // Finalize the hash after all data has been processed.
+///    let hash = hasher.finalize();
+///
+///    Ok(hash)
+///}
+///
+/// #[tokio::main]
+/// async fn main() -> Result<(), std::io::Error> {
+///     // Example: In-memory data.
+///     let data = b"Hello, world!"; // A byte slice.
+///     let reader = Cursor::new(data); // Create an in-memory AsyncRead.
+///     let hasher = Hasher;
+///     hash_stream(reader, hasher).await
+/// }
+/// ```
+///
+///
+/// ## Example 2: Compressing Data
+///
+/// When compressing data, the use of `SyncIoBridge` is unnecessary as it introduces
+/// blocking and inefficient code. Instead, you can utilize an async compression library
+/// such as the [`async-compression`](https://docs.rs/async-compression/latest/async_compression/)
+/// crate, which is built to handle asynchronous data streams efficiently.
+///
+/// Explanation: This example shows how to asynchronously compress data using an
+/// async compression library. By reading and writing asynchronously, it avoids
+/// blocking and is more efficient than using `SyncIoBridge` with a non-async
+/// compression library.
+///
+/// ```ignore
+/// use async_compression::tokio::write::GzipEncoder;
+/// use std::io::Cursor;
+/// use tokio::io::AsyncRead;
+///
+/// /// Asynchronously compresses data from an async reader using Gzip and an async encoder.
+/// async fn compress_data(mut reader: impl AsyncRead + Unpin) -> Result<(), std::io::Error> {
+///    let writer = tokio::io::sink();
+///
+///    // Create a Gzip encoder that wraps the writer.
+///    let mut encoder = GzipEncoder::new(writer);
+///
+///    // Copy data from the reader to the encoder, compressing it.
+///    tokio::io::copy(&mut reader, &mut encoder).await?;
+///
+///    Ok(())
+///}
+///
+/// #[tokio::main]
+/// async fn main() -> Result<(), std::io::Error> {
+///     // Example: In-memory data.
+///     let data = b"Hello, world!"; // A byte slice.
+///     let reader = Cursor::new(data); // Create an in-memory AsyncRead.
+///     compress_data(reader).await?;
+///
+///   Ok(())
+/// }
+/// ```
+///
+///
+/// ## Example 3: Parsing Data Formats
+///
+///
+/// `SyncIoBridge` is not ideal when parsing data formats such as `JSON`, as it
+/// blocks async operations. A more efficient approach is to read data asynchronously
+/// into memory and then `deserialize` it, avoiding unnecessary synchronization overhead.
+///
+/// Explanation: This example shows how to asynchronously read data into memory
+/// and then parse it as `JSON`. By avoiding `SyncIoBridge`, the asynchronous runtime
+/// remains unblocked, leading to better performance when working with asynchronous
+/// I/O streams.
+///
+/// ```rust,no_run
+/// use tokio::io::AsyncRead;
+/// use tokio::io::AsyncReadExt;
+/// use std::io::Cursor;
+/// # mod serde {
+/// #     pub trait DeserializeOwned: 'static {}
+/// #     impl<T: 'static> DeserializeOwned for T {}
+/// # }
+/// # mod serde_json {
+/// #     use super::serde::DeserializeOwned;
+/// #     pub fn from_slice<T: DeserializeOwned>(_: &[u8]) -> Result<T, std::io::Error> {
+/// #         unimplemented!()
+/// #     }
+/// # }
+/// # #[derive(Debug)] struct MyStruct;
+///
+///
+/// async fn parse_json(mut reader: impl AsyncRead + Unpin) -> Result<MyStruct, std::io::Error> {
+///    // Read all data from the reader into a Vec<u8>.
+///    let mut data = Vec::new();
+///    reader.read_to_end(&mut data).await?;
+///
+///    // Deserialize the data from the Vec<u8> into a MyStruct instance.
+///    let value: MyStruct = serde_json::from_slice(&data)?;
+///
+///    Ok(value)
+///}
+///
+/// #[tokio::main]
+/// async fn main() -> Result<(), std::io::Error> {
+///     // Example: In-memory data.
+///     let data = b"Hello, world!"; // A byte slice.
+///     let reader = Cursor::new(data); // Create an in-memory AsyncRead.
+///     parse_json(reader).await?;
+///     Ok(())
+/// }
+/// ```
+///
+/// ## Correct Usage of `SyncIoBridge` inside `spawn_blocking`
+///
+/// `SyncIoBridge` is mainly useful when you need to interface with synchronous
+/// libraries from an asynchronous context.
+///
+/// Explanation: This example shows how to use `SyncIoBridge` inside a `spawn_blocking`
+/// task to safely perform synchronous I/O without blocking the async runtime. The
+/// `spawn_blocking` ensures that the synchronous code is offloaded to a dedicated
+/// thread pool, preventing it from interfering with the async tasks.
+///
+/// ```rust
+/// use tokio::task::spawn_blocking;
+/// use tokio_util::io::SyncIoBridge;
+/// use tokio::io::AsyncRead;
+/// use std::marker::Unpin;
+/// use std::io::Cursor;
+///
+/// /// Wraps an async reader with `SyncIoBridge` and performs synchronous I/O operations in a blocking task.
+/// async fn process_sync_io(reader: impl AsyncRead + Unpin + Send + 'static) -> Result<Vec<u8>, std::io::Error> {
+///    // Wrap the async reader with `SyncIoBridge` to allow synchronous reading.
+///    let mut sync_reader = SyncIoBridge::new(reader);
+///
+///    // Spawn a blocking task to perform synchronous I/O operations.
+///    let result = spawn_blocking(move || {
+///        // Create an in-memory buffer to hold the copied data.
+///        let mut buffer = Vec::new();
+///        // Copy data from the sync_reader to the buffer.
+///        std::io::copy(&mut sync_reader, &mut buffer)?;
+///        // Return the buffer containing the copied data.
+///        Ok::<_, std::io::Error>(buffer)
+///    })
+///    .await??;
+///
+///    // Return the result from the blocking task.
+///    Ok(result)
+///}
+///
+/// #[tokio::main]
+/// async fn main() -> Result<(), std::io::Error> {
+///     // Example: In-memory data.
+///     let data = b"Hello, world!"; // A byte slice.
+///     let reader = Cursor::new(data); // Create an in-memory AsyncRead.
+///     let result = process_sync_io(reader).await?;
+///
+///     // You can use `result` here as needed.
+///
+///     Ok(())
+/// }
+/// ```
+///
 #[derive(Debug)]
 pub struct SyncIoBridge<T> {
     src: T,
@@ -154,3 +405,15 @@ impl<T: Unpin> SyncIoBridge<T> {
         self.src
     }
 }
+
+impl<T> AsMut<T> for SyncIoBridge<T> {
+    fn as_mut(&mut self) -> &mut T {
+        &mut self.src
+    }
+}
+
+impl<T> AsRef<T> for SyncIoBridge<T> {
+    fn as_ref(&self) -> &T {
+        &self.src
+    }
+}

tokio-util/src/lib.rs

@@ -1,4 +1,3 @@
-#![allow(unknown_lints, unexpected_cfgs)]
 #![allow(clippy::needless_doctest_main)]
 #![warn(
     missing_debug_implementations,

tokio-util/src/sync/cancellation_token.rs

@@ -287,6 +287,22 @@ impl CancellationToken {
         }
         .await
     }
+
+    /// Runs a future to completion and returns its result wrapped inside of an `Option`
+    /// unless the `CancellationToken` is cancelled. In that case the function returns
+    /// `None` and the future gets dropped.
+    ///
+    /// The function takes self by value and returns a future that owns the token.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is only cancel safe if `fut` is cancel safe.
+    pub async fn run_until_cancelled_owned<F>(self, fut: F) -> Option<F::Output>
+    where
+        F: Future,
+    {
+        self.run_until_cancelled(fut).await
+    }
 }
 
 // ===== impl WaitForCancellationFuture =====

tokio-util/src/task/spawn_pinned.rs

@@ -39,9 +39,9 @@ use tokio::task::{spawn_local, JoinHandle, LocalSet};
 ///             task::spawn_local(async move {
 ///                 println!("{}", data_clone);
 ///             });
-///     
+///
 ///             data.to_string()
-///         }   
+///         }
 ///     }).await.unwrap();
 ///     println!("output: {}", output);
 /// }
@@ -249,7 +249,7 @@ impl LocalPool {
             // Send the callback to the LocalSet task
             if let Err(e) = worker_spawner.send(spawn_task) {
                 // Propagate the error as a panic in the join handle.
-                panic!("Failed to send job to worker: {}", e);
+                panic!("Failed to send job to worker: {e}");
             }
 
             // Wait for the task's join handle
@@ -260,7 +260,7 @@ impl LocalPool {
                     // join handle... We assume something happened to the worker
                     // and the task was not spawned. Propagate the error as a
                     // panic in the join handle.
-                    panic!("Worker failed to send join handle: {}", e);
+                    panic!("Worker failed to send join handle: {e}");
                 }
             };
 
@@ -284,12 +284,12 @@ impl LocalPool {
                         // No one else should have the join handle, so this is
                         // unexpected. Forward this error as a panic in the join
                         // handle.
-                        panic!("spawn_pinned task was canceled: {}", e);
+                        panic!("spawn_pinned task was canceled: {e}");
                     } else {
                         // Something unknown happened (not a panic or
                         // cancellation). Forward this error as a panic in the
                         // join handle.
-                        panic!("spawn_pinned task failed: {}", e);
+                        panic!("spawn_pinned task failed: {e}");
                     }
                 }
             }

tokio-util/src/time/delay_queue.rs

@@ -665,7 +665,7 @@ impl<T> DelayQueue<T> {
                 // The delay is already expired, store it in the expired queue
                 self.expired.push(key, &mut self.slab);
             }
-            Err((_, err)) => panic!("invalid deadline; err={:?}", err),
+            Err((_, err)) => panic!("invalid deadline; err={err:?}"),
         }
     }
 

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

@@ -39,86 +39,10 @@ const LEVEL_MULT: usize = 64;
 
 impl<T: Stack> Level<T> {
     pub(crate) fn new(level: usize) -> Level<T> {
-        // Rust's derived implementations for arrays require that the value
-        // contained by the array be `Copy`. So, here we have to manually
-        // initialize every single slot.
-        macro_rules! s {
-            () => {
-                T::default()
-            };
-        }
-
         Level {
             level,
             occupied: 0,
-            slot: [
-                // It does not look like the necessary traits are
-                // derived for [T; 64].
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-                s!(),
-            ],
+            slot: std::array::from_fn(|_| T::default()),
         }
     }
 

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

@@ -280,13 +280,7 @@ mod test {
     #[test]
     fn test_level_for() {
         for pos in 0..64 {
-            assert_eq!(
-                0,
-                level_for(0, pos),
-                "level_for({}) -- binary = {:b}",
-                pos,
-                pos
-            );
+            assert_eq!(0, level_for(0, pos), "level_for({pos}) -- binary = {pos:b}");
         }
 
         for level in 1..5 {
@@ -295,9 +289,7 @@ mod test {
                 assert_eq!(
                     level,
                     level_for(0, a as u64),
-                    "level_for({}) -- binary = {:b}",
-                    a,
-                    a
+                    "level_for({a}) -- binary = {a:b}"
                 );
 
                 if pos > level {
@@ -305,9 +297,7 @@ mod test {
                     assert_eq!(
                         level,
                         level_for(0, a as u64),
-                        "level_for({}) -- binary = {:b}",
-                        a,
-                        a
+                        "level_for({a}) -- binary = {a:b}"
                     );
                 }
 
@@ -316,9 +306,7 @@ mod test {
                     assert_eq!(
                         level,
                         level_for(0, a as u64),
-                        "level_for({}) -- binary = {:b}",
-                        a,
-                        a
+                        "level_for({a}) -- binary = {a:b}"
                     );
                 }
             }

tokio-util/src/udp/frame.rs

@@ -5,13 +5,13 @@ use tokio::{io::ReadBuf, net::UdpSocket};
 
 use bytes::{BufMut, BytesMut};
 use futures_sink::Sink;
+use std::io;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};
 use std::{
     borrow::Borrow,
     net::{Ipv4Addr, SocketAddr, SocketAddrV4},
 };
-use std::{io, mem::MaybeUninit};
 
 /// A unified [`Stream`] and [`Sink`] interface to an underlying `UdpSocket`, using
 /// the `Encoder` and `Decoder` traits to encode and decode frames.
@@ -83,7 +83,7 @@ where
             let addr = {
                 // Safety: `chunk_mut()` returns a `&mut UninitSlice`, and `UninitSlice` is a
                 // transparent wrapper around `[MaybeUninit<u8>]`.
-                let buf = unsafe { &mut *(pin.rd.chunk_mut() as *mut _ as *mut [MaybeUninit<u8>]) };
+                let buf = unsafe { pin.rd.chunk_mut().as_uninit_slice_mut() };
                 let mut read = ReadBuf::uninit(buf);
                 let ptr = read.filled().as_ptr();
                 let res = ready!(pin.socket.borrow().poll_recv_from(cx, &mut read));
@@ -91,9 +91,10 @@ where
                 assert_eq!(ptr, read.filled().as_ptr());
                 let addr = res?;
 
+                let filled = read.filled().len();
                 // Safety: This is guaranteed to be the number of initialized (and read) bytes due
                 // to the invariants provided by `ReadBuf::filled`.
-                unsafe { pin.rd.advance_mut(read.filled().len()) };
+                unsafe { pin.rd.advance_mut(filled) };
 
                 addr
             };

tokio-util/src/util/poll_buf.rs

@@ -2,7 +2,6 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
 use bytes::{Buf, BufMut};
 use std::io::{self, IoSlice};
-use std::mem::MaybeUninit;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};
 
@@ -59,7 +58,7 @@ pub fn poll_read_buf<T: AsyncRead + ?Sized, B: BufMut>(
 
         // Safety: `chunk_mut()` returns a `&mut UninitSlice`, and `UninitSlice` is a
         // transparent wrapper around `[MaybeUninit<u8>]`.
-        let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
+        let dst = unsafe { dst.as_uninit_slice_mut() };
         let mut buf = ReadBuf::uninit(dst);
         let ptr = buf.filled().as_ptr();
         ready!(io.poll_read(cx, &mut buf)?);

tokio-util/tests/codecs.rs

@@ -62,6 +62,19 @@ fn lines_decoder() {
     assert_eq!(None, codec.decode_eof(buf).unwrap());
 }
 
+#[test]
+fn lines_decoder_invalid_utf8() {
+    let mut codec = LinesCodec::new();
+    let buf = &mut BytesMut::new();
+    buf.reserve(200);
+    buf.put_slice(b"line 1\xc3\x28");
+    assert_eq!(None, codec.decode(buf).unwrap());
+    assert!(codec.decode_eof(buf).is_err());
+    assert_eq!(None, codec.decode_eof(buf).unwrap());
+    buf.put_slice(b"line 22222222222222\n");
+    assert_eq!("line 22222222222222", codec.decode(buf).unwrap().unwrap());
+}
+
 #[test]
 fn lines_decoder_max_length() {
     const MAX_LENGTH: usize = 6;
@@ -75,32 +88,17 @@ fn lines_decoder_max_length() {
     assert!(codec.decode(buf).is_err());
 
     let line = codec.decode(buf).unwrap().unwrap();
-    assert!(
-        line.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        line,
-        MAX_LENGTH
-    );
+    assert!(line.len() <= MAX_LENGTH, "{line:?}.len() <= {MAX_LENGTH:?}");
     assert_eq!("line 2", line);
 
     assert!(codec.decode(buf).is_err());
 
     let line = codec.decode(buf).unwrap().unwrap();
-    assert!(
-        line.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        line,
-        MAX_LENGTH
-    );
+    assert!(line.len() <= MAX_LENGTH, "{line:?}.len() <= {MAX_LENGTH:?}");
     assert_eq!("line 4", line);
 
     let line = codec.decode(buf).unwrap().unwrap();
-    assert!(
-        line.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        line,
-        MAX_LENGTH
-    );
+    assert!(line.len() <= MAX_LENGTH, "{line:?}.len() <= {MAX_LENGTH:?}");
     assert_eq!("", line);
 
     assert_eq!(None, codec.decode(buf).unwrap());
@@ -109,12 +107,7 @@ fn lines_decoder_max_length() {
     assert_eq!(None, codec.decode(buf).unwrap());
 
     let line = codec.decode_eof(buf).unwrap().unwrap();
-    assert!(
-        line.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        line,
-        MAX_LENGTH
-    );
+    assert!(line.len() <= MAX_LENGTH, "{line:?}.len() <= {MAX_LENGTH:?}");
     assert_eq!("\rk", line);
 
     assert_eq!(None, codec.decode(buf).unwrap());
@@ -273,18 +266,14 @@ fn any_delimiters_decoder_max_length() {
     let chunk = codec.decode(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("chunk 2", chunk);
 
     let chunk = codec.decode(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("chunk 3", chunk);
 
@@ -292,36 +281,28 @@ fn any_delimiters_decoder_max_length() {
     let chunk = codec.decode(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("", chunk);
 
     let chunk = codec.decode(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("chunk 4", chunk);
 
     let chunk = codec.decode(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("", chunk);
 
     let chunk = codec.decode(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("", chunk);
 
@@ -333,9 +314,7 @@ fn any_delimiters_decoder_max_length() {
     let chunk = codec.decode_eof(buf).unwrap().unwrap();
     assert!(
         chunk.len() <= MAX_LENGTH,
-        "{:?}.len() <= {:?}",
-        chunk,
-        MAX_LENGTH
+        "{chunk:?}.len() <= {MAX_LENGTH:?}"
     );
     assert_eq!("k", chunk);
 

tokio-util/tests/framed_write.rs

@@ -180,7 +180,7 @@ impl Write for Mock {
                 Ok(data.len())
             }
             Some(Err(e)) => Err(e),
-            None => panic!("unexpected write; {:?}", src),
+            None => panic!("unexpected write; {src:?}"),
         }
     }
 

tokio-util/tests/io_reader_stream.rs

@@ -42,7 +42,7 @@ async fn correct_behavior_on_errors() {
     let mut had_error = false;
     loop {
         let item = stream.next().await.unwrap();
-        println!("{:?}", item);
+        println!("{item:?}");
         match item {
             Ok(bytes) => {
                 let bytes = &*bytes;

tokio-util/tests/length_delimited.rs

@@ -789,7 +789,7 @@ impl AsyncWrite for Mock {
         match self.calls.pop_front() {
             Some(Poll::Ready(Ok(Op::Data(data)))) => {
                 let len = data.len();
-                assert!(src.len() >= len, "expect={:?}; actual={:?}", data, src);
+                assert!(src.len() >= len, "expect={data:?}; actual={src:?}");
                 assert_eq!(&data[..], &src[..len]);
                 Poll::Ready(Ok(len))
             }

tokio-util/tests/spawn_pinned.rs

@@ -1,7 +1,5 @@
 #![warn(rust_2018_idioms)]
 #![cfg(not(target_os = "wasi"))] // Wasi doesn't support threads
-// Blocked on https://github.com/rust-lang/miri/issues/3911
-#![cfg(not(miri))]
 
 use std::rc::Rc;
 use std::sync::Arc;

tokio-util/tests/sync_cancellation_token.rs

@@ -493,3 +493,58 @@ fn run_until_cancelled_test() {
         );
     }
 }
+
+#[test]
+fn run_until_cancelled_owned_test() {
+    let (waker, _) = new_count_waker();
+
+    {
+        let token = CancellationToken::new();
+        let to_cancel = token.clone();
+
+        let takes_ownership = move |token: CancellationToken| {
+            token.run_until_cancelled_owned(std::future::pending::<()>())
+        };
+
+        let fut = takes_ownership(token);
+        pin!(fut);
+
+        assert_eq!(
+            Poll::Pending,
+            fut.as_mut().poll(&mut Context::from_waker(&waker))
+        );
+
+        to_cancel.cancel();
+
+        assert_eq!(
+            Poll::Ready(None),
+            fut.as_mut().poll(&mut Context::from_waker(&waker))
+        );
+    }
+
+    {
+        let (tx, rx) = oneshot::channel::<()>();
+
+        let token = CancellationToken::new();
+        let takes_ownership = move |token: CancellationToken, rx: oneshot::Receiver<()>| {
+            token.run_until_cancelled_owned(async move {
+                rx.await.unwrap();
+                42
+            })
+        };
+        let fut = takes_ownership(token, rx);
+        pin!(fut);
+
+        assert_eq!(
+            Poll::Pending,
+            fut.as_mut().poll(&mut Context::from_waker(&waker))
+        );
+
+        tx.send(()).unwrap();
+
+        assert_eq!(
+            Poll::Ready(Some(42)),
+            fut.as_mut().poll(&mut Context::from_waker(&waker))
+        );
+    }
+}

tokio-util/tests/task_join_map.rs

@@ -1,4 +1,3 @@
-#![allow(unknown_lints, unexpected_cfgs)]
 #![warn(rust_2018_idioms)]
 #![cfg(all(feature = "rt", tokio_unstable))]
 

tokio-util/tests/time_delay_queue.rs

@@ -111,7 +111,7 @@ async fn multi_delay_at_start() {
 
     let start = Instant::now();
     for elapsed in 0..1200 {
-        println!("elapsed: {:?}", elapsed);
+        println!("elapsed: {elapsed:?}");
         let elapsed = elapsed + 1;
         tokio::time::sleep_until(start + ms(elapsed)).await;
 
@@ -328,7 +328,7 @@ async fn remove_at_timer_wheel_threshold() {
             let entry = queue.remove(&key1).into_inner();
             assert_eq!(entry, "foo");
         }
-        other => panic!("other: {:?}", other),
+        other => panic!("other: {other:?}"),
     }
 }
 

tokio/CHANGELOG.md

@@ -1,3 +1,171 @@
+# 1.44.1 (March 13th, 2025)
+
+### Fixed
+
+- rt: skip defer queue in `block_in_place` context ([#7216])
+
+[#7216]: https://github.com/tokio-rs/tokio/pull/7216
+
+# 1.44.0 (March 7th, 2025)
+
+This release changes the `from_std` method on sockets to panic if a blocking
+socket is provided. We determined this change is not a breaking change as Tokio is not
+intended to operate using blocking sockets. Doing so results in runtime hangs and
+should be considered a bug. Accidentally passing a blocking socket to Tokio is one
+of the most common user mistakes. If this change causes an issue for you, please
+comment on [#7172].
+
+### Added
+
+ - coop: add `task::coop` module ([#7116])
+ - process: add `Command::get_kill_on_drop()` ([#7086])
+ - sync: add `broadcast::Sender::closed` ([#6685], [#7090])
+ - sync: add `broadcast::WeakSender` ([#7100])
+ - sync: add `oneshot::Receiver::is_empty()` ([#7153])
+ - sync: add `oneshot::Receiver::is_terminated()` ([#7152])
+
+### Fixed
+
+ - fs: empty reads on `File` should not start a background read ([#7139])
+ - process: calling `start_kill` on exited child should not fail ([#7160])
+ - signal: fix `CTRL_CLOSE`, `CTRL_LOGOFF`, `CTRL_SHUTDOWN` on windows ([#7122])
+ - sync: properly handle panic during mpsc drop ([#7094])
+
+### Changes
+
+ - runtime: clean up magic number in registration set ([#7112])
+ - coop: make coop yield using waker defer strategy ([#7185])
+ - macros: make `select!` budget-aware ([#7164])
+ - net: panic when passing a blocking socket to `from_std` ([#7166])
+ - io: clean up buffer casts ([#7142])
+
+### Changes to unstable APIs
+
+ - rt: add before and after task poll callbacks ([#7120])
+ - tracing: make the task tracing API unstable public ([#6972])
+
+### Documented
+
+ - docs: fix nesting of sections in top-level docs ([#7159])
+ - fs: rename symlink and hardlink parameter names ([#7143])
+ - io: swap reader/writer in simplex doc test ([#7176])
+ - macros: docs about `select!` alternatives ([#7110])
+ - net: rename the argument for `send_to` ([#7146])
+ - process: add example for reading `Child` stdout ([#7141])
+ - process: clarify `Child::kill` behavior ([#7162])
+ - process: fix grammar of the `ChildStdin` struct doc comment ([#7192])
+ - runtime: consistently use `worker_threads` instead of `core_threads` ([#7186])
+
+[#6685]: https://github.com/tokio-rs/tokio/pull/6685
+[#6972]: https://github.com/tokio-rs/tokio/pull/6972
+[#7086]: https://github.com/tokio-rs/tokio/pull/7086
+[#7090]: https://github.com/tokio-rs/tokio/pull/7090
+[#7094]: https://github.com/tokio-rs/tokio/pull/7094
+[#7100]: https://github.com/tokio-rs/tokio/pull/7100
+[#7110]: https://github.com/tokio-rs/tokio/pull/7110
+[#7112]: https://github.com/tokio-rs/tokio/pull/7112
+[#7116]: https://github.com/tokio-rs/tokio/pull/7116
+[#7120]: https://github.com/tokio-rs/tokio/pull/7120
+[#7122]: https://github.com/tokio-rs/tokio/pull/7122
+[#7139]: https://github.com/tokio-rs/tokio/pull/7139
+[#7141]: https://github.com/tokio-rs/tokio/pull/7141
+[#7142]: https://github.com/tokio-rs/tokio/pull/7142
+[#7143]: https://github.com/tokio-rs/tokio/pull/7143
+[#7146]: https://github.com/tokio-rs/tokio/pull/7146
+[#7152]: https://github.com/tokio-rs/tokio/pull/7152
+[#7153]: https://github.com/tokio-rs/tokio/pull/7153
+[#7159]: https://github.com/tokio-rs/tokio/pull/7159
+[#7160]: https://github.com/tokio-rs/tokio/pull/7160
+[#7162]: https://github.com/tokio-rs/tokio/pull/7162
+[#7164]: https://github.com/tokio-rs/tokio/pull/7164
+[#7166]: https://github.com/tokio-rs/tokio/pull/7166
+[#7172]: https://github.com/tokio-rs/tokio/pull/7172
+[#7176]: https://github.com/tokio-rs/tokio/pull/7176
+[#7185]: https://github.com/tokio-rs/tokio/pull/7185
+[#7186]: https://github.com/tokio-rs/tokio/pull/7186
+[#7192]: https://github.com/tokio-rs/tokio/pull/7192
+
+# 1.43.0 (Jan 8th, 2025)
+
+### Added
+
+- net: add `UdpSocket::peek` methods ([#7068])
+- net: add support for Haiku OS ([#7042])
+- process: add `Command::into_std()` ([#7014])
+- signal: add `SignalKind::info` on illumos ([#6995])
+- signal: add support for realtime signals on illumos ([#7029])
+
+### Fixed
+
+- io: don't call `set_len` before initializing vector in `Blocking` ([#7054])
+- macros: suppress `clippy::needless_return` in `#[tokio::main]` ([#6874])
+- runtime: fix thread parking on WebAssembly ([#7041])
+
+### Changes
+
+- chore: use unsync loads for `unsync_load` ([#7073])
+- io: use `Buf::put_bytes` in `Repeat` read impl ([#7055])
+- task: drop the join waker of a task eagerly ([#6986])
+
+### Changes to unstable APIs
+
+- metrics: improve flexibility of H2Histogram Configuration ([#6963])
+- taskdump: add accessor methods for backtrace ([#6975])
+
+### Documented
+
+- io: clarify `ReadBuf::uninit` allows initialized buffers as well ([#7053])
+- net: fix ambiguity in `TcpStream::try_write_vectored` docs ([#7067])
+- runtime: fix `LocalRuntime` doc links ([#7074])
+- sync: extend documentation for `watch::Receiver::wait_for` ([#7038])
+- sync: fix typos in `OnceCell` docs ([#7047])
+
+[#6874]: https://github.com/tokio-rs/tokio/pull/6874
+[#6963]: https://github.com/tokio-rs/tokio/pull/6963
+[#6975]: https://github.com/tokio-rs/tokio/pull/6975
+[#6986]: https://github.com/tokio-rs/tokio/pull/6986
+[#6995]: https://github.com/tokio-rs/tokio/pull/6995
+[#7014]: https://github.com/tokio-rs/tokio/pull/7014
+[#7029]: https://github.com/tokio-rs/tokio/pull/7029
+[#7038]: https://github.com/tokio-rs/tokio/pull/7038
+[#7041]: https://github.com/tokio-rs/tokio/pull/7041
+[#7042]: https://github.com/tokio-rs/tokio/pull/7042
+[#7047]: https://github.com/tokio-rs/tokio/pull/7047
+[#7053]: https://github.com/tokio-rs/tokio/pull/7053
+[#7054]: https://github.com/tokio-rs/tokio/pull/7054
+[#7055]: https://github.com/tokio-rs/tokio/pull/7055
+[#7067]: https://github.com/tokio-rs/tokio/pull/7067
+[#7068]: https://github.com/tokio-rs/tokio/pull/7068
+[#7073]: https://github.com/tokio-rs/tokio/pull/7073
+[#7074]: https://github.com/tokio-rs/tokio/pull/7074
+
+# 1.42.0 (Dec 3rd, 2024)
+
+### Added
+
+- io: add `AsyncFd::{try_io, try_io_mut}` ([#6967])
+
+### Fixed
+
+- io: avoid `ptr->ref->ptr` roundtrip in RegistrationSet ([#6929])
+- runtime: do not defer `yield_now` inside `block_in_place` ([#6999])
+
+### Changes
+
+- io: simplify io readiness logic ([#6966])
+
+### Documented
+
+- net: fix docs for `tokio::net::unix::{pid_t, gid_t, uid_t}` ([#6791])
+- time: fix a typo in `Instant` docs ([#6982])
+
+[#6791]: https://github.com/tokio-rs/tokio/pull/6791
+[#6929]: https://github.com/tokio-rs/tokio/pull/6929
+[#6966]: https://github.com/tokio-rs/tokio/pull/6966
+[#6967]: https://github.com/tokio-rs/tokio/pull/6967
+[#6982]: https://github.com/tokio-rs/tokio/pull/6982
+[#6999]: https://github.com/tokio-rs/tokio/pull/6999
+
 # 1.41.1 (Nov 7th, 2024)
 
 ### Fixed
@@ -10,7 +178,7 @@
 [#6938]: https://github.com/tokio-rs/tokio/pull/6938
 [#6944]: https://github.com/tokio-rs/tokio/pull/6944
 
-# 1.41.0 (Oct 22th, 2024)
+# 1.41.0 (Oct 22nd, 2024)
 
 ### Added
 

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.41.1"
+version = "1.44.1"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -86,12 +86,12 @@ test-util = ["rt", "sync", "time"]
 time = []
 
 [dependencies]
-tokio-macros = { version = "~2.4.0", path = "../tokio-macros", optional = true }
+tokio-macros = { version = "~2.5.0", path = "../tokio-macros", optional = true }
 
 pin-project-lite = "0.2.11"
 
 # Everything else is optional...
-bytes = { version = "1.0.0", optional = true }
+bytes = { version = "1.2.1", optional = true }
 mio = { version = "1.0.1", optional = true, default-features = false }
 parking_lot = { version = "0.12.0", optional = true }
 
@@ -109,11 +109,11 @@ tracing = { version = "0.1.29", default-features = false, features = ["std"], op
 backtrace = { version = "0.3.58" }
 
 [target.'cfg(unix)'.dependencies]
-libc = { version = "0.2.149", optional = true }
+libc = { version = "0.2.168", optional = true }
 signal-hook-registry = { version = "1.1.1", optional = true }
 
 [target.'cfg(unix)'.dev-dependencies]
-libc = { version = "0.2.149" }
+libc = { version = "0.2.168" }
 nix = { version = "0.29.0", default-features = false, features = ["aio", "fs", "socket"] }
 
 [target.'cfg(windows)'.dependencies.windows-sys]
@@ -133,6 +133,7 @@ tokio-stream = { version = "0.1", path = "../tokio-stream" }
 futures = { version = "0.3.0", features = ["async-await"] }
 mockall = "0.11.1"
 async-stream = "0.3"
+futures-concurrency = "7.6.3"
 
 [target.'cfg(not(target_family = "wasm"))'.dev-dependencies]
 socket2 = "0.5.5"
@@ -151,6 +152,9 @@ mio-aio = { version = "0.9.0", features = ["tokio"] }
 [target.'cfg(loom)'.dev-dependencies]
 loom = { version = "0.7", features = ["futures", "checkpoint"] }
 
+[target.'cfg(all(tokio_unstable, target_has_atomic = "64"))'.dev-dependencies]
+tracing-mock = "= 0.1.0-beta.1"
+
 [package.metadata.docs.rs]
 all-features = true
 # enable unstable features in the documentation
@@ -170,3 +174,6 @@ allowed_external_types = [
   "bytes::buf::buf_mut::BufMut",
   "tokio_macros::*",
 ]
+
+[lints]
+workspace = true

tokio/README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.41.1", features = ["full"] }
+tokio = { version = "1.44.1", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -78,7 +78,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             loop {
                 let n = match socket.read(&mut buf).await {
                     // socket closed
-                    Ok(n) if n == 0 => return,
+                    Ok(0) => return,
                     Ok(n) => n,
                     Err(e) => {
                         eprintln!("failed to read from socket; err = {:?}", e);
@@ -205,8 +205,8 @@ works with the MSRV of that minor release.
 
 ## Release schedule
 
-Tokio doesn't follow a fixed release schedule, but we typically make one to two
-new minor releases each month. We make patch releases for bugfixes as necessary.
+Tokio doesn't follow a fixed release schedule, but we typically make one minor
+release each month. We make patch releases for bugfixes as necessary.
 
 ## Bug patching policy
 
@@ -216,9 +216,8 @@ warrants a patch release with a fix for the bug, it will be backported and
 released as a new patch release for each LTS minor version. Our current LTS
 releases are:
 
- * `1.32.x` - LTS release until September 2024. (MSRV 1.63)
- * `1.36.x` - LTS release until March 2025. (MSRV 1.63)
  * `1.38.x` - LTS release until July 2025. (MSRV 1.63)
+ * `1.43.x` - LTS release until March 2026. (MSRV 1.70)
 
 Each LTS release will continue to receive backported fixes for at least a year.
 If you wish to use a fixed minor release in your project, we recommend that you
@@ -228,7 +227,7 @@ To use a fixed minor version, you can specify the version with a tilde. For
 example, to specify that you wish to use the newest `1.32.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.32", features = [...] }
+tokio = { version = "~1.38", features = [...] }
 ```
 
 ### Previous LTS releases
@@ -238,6 +237,8 @@ tokio = { version = "~1.32", features = [...] }
  * `1.18.x` - LTS release until June 2023.
  * `1.20.x` - LTS release until September 2023.
  * `1.25.x` - LTS release until March 2024.
+ * `1.32.x` - LTS release until September 2024.
+ * `1.36.x` - LTS release until March 2025.
 
 ## License
 

tokio/src/doc/mod.rs

@@ -24,21 +24,21 @@ pub enum NotDefinedHere {}
 impl mio::event::Source for NotDefinedHere {
     fn register(
         &mut self,
-        registry: &mio::Registry,
-        token: mio::Token,
-        interests: mio::Interest,
+        _registry: &mio::Registry,
+        _token: mio::Token,
+        _interests: mio::Interest,
     ) -> std::io::Result<()> {
         Ok(())
     }
     fn reregister(
         &mut self,
-        registry: &mio::Registry,
-        token: mio::Token,
-        interests: mio::Interest,
+        _registry: &mio::Registry,
+        _token: mio::Token,
+        _interests: mio::Interest,
     ) -> std::io::Result<()> {
         Ok(())
     }
-    fn deregister(&mut self, registry: &mio::Registry) -> std::io::Result<()> {
+    fn deregister(&mut self, _registry: &mio::Registry) -> std::io::Result<()> {
         Ok(())
     }
 }

tokio/src/fs/file.rs

@@ -7,6 +7,7 @@ use crate::io::blocking::{Buf, DEFAULT_MAX_BUF_SIZE};
 use crate::io::{AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
 use crate::sync::Mutex;
 
+use std::cmp;
 use std::fmt;
 use std::fs::{Metadata, Permissions};
 use std::future::Future;
@@ -586,6 +587,7 @@ impl AsyncRead for File {
         dst: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
         ready!(crate::trace::trace_leaf(cx));
+
         let me = self.get_mut();
         let inner = me.inner.get_mut();
 
@@ -594,17 +596,20 @@ impl AsyncRead for File {
                 State::Idle(ref mut buf_cell) => {
                     let mut buf = buf_cell.take().unwrap();
 
-                    if !buf.is_empty() {
+                    if !buf.is_empty() || dst.remaining() == 0 {
                         buf.copy_to(dst);
                         *buf_cell = Some(buf);
                         return Poll::Ready(Ok(()));
                     }
 
-                    buf.ensure_capacity_for(dst, me.max_buf_size);
                     let std = me.std.clone();
 
+                    let max_buf_size = cmp::min(dst.remaining(), me.max_buf_size);
                     inner.state = State::Busy(spawn_blocking(move || {
-                        let res = buf.read_from(&mut &*std);
+                        // SAFETY: the `Read` implementation of `std` does not
+                        // read from the buffer it is borrowing and correctly
+                        // reports the length of the data written into the buffer.
+                        let res = unsafe { buf.read_from(&mut &*std, max_buf_size) };
                         (Operation::Read(res), buf)
                     }));
                 }

tokio/src/fs/hard_link.rs

@@ -7,7 +7,7 @@ use std::path::Path;
 ///
 /// This is an async version of [`std::fs::hard_link`].
 ///
-/// The `dst` path will be a link pointing to the `src` path. Note that systems
+/// The `link` path will be a link pointing to the `original` path. Note that systems
 /// often require these two paths to both be located on the same filesystem.
 ///
 /// # Platform-specific behavior
@@ -23,7 +23,7 @@ use std::path::Path;
 /// This function will return an error in the following situations, but is not
 /// limited to just these cases:
 ///
-/// * The `src` path is not a file or doesn't exist.
+/// * The `original` path is not a file or doesn't exist.
 ///
 /// # Examples
 ///
@@ -36,9 +36,9 @@ use std::path::Path;
 ///     Ok(())
 /// }
 /// ```
-pub async fn hard_link(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
-    let src = src.as_ref().to_owned();
-    let dst = dst.as_ref().to_owned();
+pub async fn hard_link(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
+    let original = original.as_ref().to_owned();
+    let link = link.as_ref().to_owned();
 
-    asyncify(move || std::fs::hard_link(src, dst)).await
+    asyncify(move || std::fs::hard_link(original, link)).await
 }

tokio/src/fs/mocks.rs

@@ -129,7 +129,7 @@ impl<T> Future for JoinHandle<T> {
 
         match Pin::new(&mut self.rx).poll(cx) {
             Poll::Ready(Ok(v)) => Poll::Ready(Ok(v)),
-            Poll::Ready(Err(e)) => panic!("error = {:?}", e),
+            Poll::Ready(Err(e)) => panic!("error = {e:?}"),
             Poll::Pending => Poll::Pending,
         }
     }

tokio/src/fs/symlink.rs

@@ -5,12 +5,12 @@ use std::path::Path;
 
 /// Creates a new symbolic link on the filesystem.
 ///
-/// The `dst` path will be a symbolic link pointing to the `src` path.
+/// The `link` path will be a symbolic link pointing to the `original` path.
 ///
 /// This is an async version of [`std::os::unix::fs::symlink`].
-pub async fn symlink(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
-    let src = src.as_ref().to_owned();
-    let dst = dst.as_ref().to_owned();
+pub async fn symlink(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
+    let original = original.as_ref().to_owned();
+    let link = link.as_ref().to_owned();
 
-    asyncify(move || std::os::unix::fs::symlink(src, dst)).await
+    asyncify(move || std::os::unix::fs::symlink(original, link)).await
 }

tokio/src/fs/symlink_dir.rs

@@ -5,15 +5,15 @@ use std::path::Path;
 
 /// Creates a new directory symlink on the filesystem.
 ///
-/// The `dst` path will be a directory symbolic link pointing to the `src`
+/// The `link` path will be a directory symbolic link pointing to the `original`
 /// path.
 ///
 /// This is an async version of [`std::os::windows::fs::symlink_dir`][std]
 ///
 /// [std]: https://doc.rust-lang.org/std/os/windows/fs/fn.symlink_dir.html
-pub async fn symlink_dir(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
-    let src = src.as_ref().to_owned();
-    let dst = dst.as_ref().to_owned();
+pub async fn symlink_dir(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
+    let original = original.as_ref().to_owned();
+    let link = link.as_ref().to_owned();
 
-    asyncify(move || std::os::windows::fs::symlink_dir(src, dst)).await
+    asyncify(move || std::os::windows::fs::symlink_dir(original, link)).await
 }

tokio/src/fs/symlink_file.rs

@@ -5,15 +5,15 @@ use std::path::Path;
 
 /// Creates a new file symbolic link on the filesystem.
 ///
-/// The `dst` path will be a file symbolic link pointing to the `src`
+/// The `link` path will be a file symbolic link pointing to the `original`
 /// path.
 ///
 /// This is an async version of [`std::os::windows::fs::symlink_file`][std]
 ///
 /// [std]: https://doc.rust-lang.org/std/os/windows/fs/fn.symlink_file.html
-pub async fn symlink_file(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> io::Result<()> {
-    let src = src.as_ref().to_owned();
-    let dst = dst.as_ref().to_owned();
+pub async fn symlink_file(original: impl AsRef<Path>, link: impl AsRef<Path>) -> io::Result<()> {
+    let original = original.as_ref().to_owned();
+    let link = link.as_ref().to_owned();
 
-    asyncify(move || std::os::windows::fs::symlink_file(src, dst)).await
+    asyncify(move || std::os::windows::fs::symlink_file(original, link)).await
 }

tokio/src/io/async_buf_read.rs

@@ -84,15 +84,15 @@ impl<T: ?Sized + AsyncBufRead + Unpin> AsyncBufRead for &mut T {
 
 impl<P> AsyncBufRead for Pin<P>
 where
-    P: DerefMut + Unpin,
+    P: DerefMut,
     P::Target: AsyncBufRead,
 {
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
-        self.get_mut().as_mut().poll_fill_buf(cx)
+        crate::util::pin_as_deref_mut(self).poll_fill_buf(cx)
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        self.get_mut().as_mut().consume(amt);
+        crate::util::pin_as_deref_mut(self).consume(amt);
     }
 }
 

tokio/src/io/async_fd.rs

@@ -872,6 +872,56 @@ impl<T: AsRawFd> AsyncFd<T> {
             .async_io(interest, || f(self.inner.as_mut().unwrap()))
             .await
     }
+
+    /// Tries to read or write from the file descriptor using a user-provided IO operation.
+    ///
+    /// If the file descriptor is ready, the provided closure is called. The closure
+    /// should attempt to perform IO operation on the file descriptor by manually
+    /// calling the appropriate syscall. If the operation fails because the
+    /// file descriptor is not actually ready, then the closure should return a
+    /// `WouldBlock` error and the readiness flag is cleared. The return value
+    /// of the closure is then returned by `try_io`.
+    ///
+    /// If the file descriptor is not ready, then the closure is not called
+    /// and a `WouldBlock` error is returned.
+    ///
+    /// The closure should only return a `WouldBlock` error if it has performed
+    /// an IO operation on the file descriptor that failed due to the file descriptor not being
+    /// ready. Returning a `WouldBlock` error in any other situation will
+    /// incorrectly clear the readiness flag, which can cause the file descriptor to
+    /// behave incorrectly.
+    ///
+    /// The closure should not perform the IO operation using any of the methods
+    /// defined on the Tokio `AsyncFd` type, as this will mess with the
+    /// readiness flag and can cause the file descriptor to behave incorrectly.
+    ///
+    /// This method is not intended to be used with combined interests.
+    /// The closure should perform only one type of IO operation, so it should not
+    /// require more than one ready state. This method may panic or sleep forever
+    /// if it is called with a combined interest.
+    pub fn try_io<R>(
+        &self,
+        interest: Interest,
+        f: impl FnOnce(&T) -> io::Result<R>,
+    ) -> io::Result<R> {
+        self.registration
+            .try_io(interest, || f(self.inner.as_ref().unwrap()))
+    }
+
+    /// Tries to read or write from the file descriptor using a user-provided IO operation.
+    ///
+    /// The behavior is the same as [`try_io`], except that the closure can mutate the inner
+    /// value of the [`AsyncFd`].
+    ///
+    /// [`try_io`]: AsyncFd::try_io
+    pub fn try_io_mut<R>(
+        &mut self,
+        interest: Interest,
+        f: impl FnOnce(&mut T) -> io::Result<R>,
+    ) -> io::Result<R> {
+        self.registration
+            .try_io(interest, || f(self.inner.as_mut().unwrap()))
+    }
 }
 
 impl<T: AsRawFd> AsRawFd for AsyncFd<T> {

tokio/src/io/async_read.rs

@@ -80,7 +80,7 @@ impl<T: ?Sized + AsyncRead + Unpin> AsyncRead for &mut T {
 
 impl<P> AsyncRead for Pin<P>
 where
-    P: DerefMut + Unpin,
+    P: DerefMut,
     P::Target: AsyncRead,
 {
     fn poll_read(
@@ -88,7 +88,7 @@ where
         cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        self.get_mut().as_mut().poll_read(cx, buf)
+        crate::util::pin_as_deref_mut(self).poll_read(cx, buf)
     }
 }
 

tokio/src/io/async_seek.rs

@@ -68,15 +68,15 @@ impl<T: ?Sized + AsyncSeek + Unpin> AsyncSeek for &mut T {
 
 impl<P> AsyncSeek for Pin<P>
 where
-    P: DerefMut + Unpin,
+    P: DerefMut,
     P::Target: AsyncSeek,
 {
     fn start_seek(self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
-        self.get_mut().as_mut().start_seek(pos)
+        crate::util::pin_as_deref_mut(self).start_seek(pos)
     }
 
     fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
-        self.get_mut().as_mut().poll_complete(cx)
+        crate::util::pin_as_deref_mut(self).poll_complete(cx)
     }
 }
 

tokio/src/io/async_write.rs

@@ -224,7 +224,7 @@ impl<T: ?Sized + AsyncWrite + Unpin> AsyncWrite for &mut T {
 
 impl<P> AsyncWrite for Pin<P>
 where
-    P: DerefMut + Unpin,
+    P: DerefMut,
     P::Target: AsyncWrite,
 {
     fn poll_write(
@@ -232,7 +232,7 @@ where
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        self.get_mut().as_mut().poll_write(cx, buf)
+        crate::util::pin_as_deref_mut(self).poll_write(cx, buf)
     }
 
     fn poll_write_vectored(
@@ -240,7 +240,7 @@ where
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
     ) -> Poll<io::Result<usize>> {
-        self.get_mut().as_mut().poll_write_vectored(cx, bufs)
+        crate::util::pin_as_deref_mut(self).poll_write_vectored(cx, bufs)
     }
 
     fn is_write_vectored(&self) -> bool {
@@ -248,11 +248,11 @@ where
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.get_mut().as_mut().poll_flush(cx)
+        crate::util::pin_as_deref_mut(self).poll_flush(cx)
     }
 
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.get_mut().as_mut().poll_shutdown(cx)
+        crate::util::pin_as_deref_mut(self).poll_shutdown(cx)
     }
 }
 

tokio/src/io/blocking.rs

@@ -5,6 +5,7 @@ use std::cmp;
 use std::future::Future;
 use std::io;
 use std::io::prelude::*;
+use std::mem::MaybeUninit;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};
 
@@ -33,8 +34,13 @@ enum State<T> {
 
 cfg_io_blocking! {
     impl<T> Blocking<T> {
+        /// # Safety
+        ///
+        /// The `Read` implementation of `inner` must never read from the buffer
+        /// it is borrowing and must correctly report the length of the data
+        /// written into the buffer.
         #[cfg_attr(feature = "fs", allow(dead_code))]
-        pub(crate) fn new(inner: T) -> Blocking<T> {
+        pub(crate) unsafe fn new(inner: T) -> Blocking<T> {
             Blocking {
                 inner: Some(inner),
                 state: State::Idle(Some(Buf::with_capacity(0))),
@@ -64,11 +70,12 @@ where
                         return Poll::Ready(Ok(()));
                     }
 
-                    buf.ensure_capacity_for(dst, DEFAULT_MAX_BUF_SIZE);
                     let mut inner = self.inner.take().unwrap();
 
+                    let max_buf_size = cmp::min(dst.remaining(), DEFAULT_MAX_BUF_SIZE);
                     self.state = State::Busy(sys::run(move || {
-                        let res = buf.read_from(&mut inner);
+                        // SAFETY: the requirements are satisfied by `Blocking::new`.
+                        let res = unsafe { buf.read_from(&mut inner, max_buf_size) };
                         (res, buf, inner)
                     }));
                 }
@@ -227,25 +234,30 @@ impl Buf {
         &self.buf[self.pos..]
     }
 
-    pub(crate) fn ensure_capacity_for(&mut self, bytes: &ReadBuf<'_>, max_buf_size: usize) {
+    /// # Safety
+    ///
+    /// `rd` must not read from the buffer `read` is borrowing and must correctly
+    /// report the length of the data written into the buffer.
+    pub(crate) unsafe fn read_from<T: Read>(
+        &mut self,
+        rd: &mut T,
+        max_buf_size: usize,
+    ) -> io::Result<usize> {
         assert!(self.is_empty());
+        self.buf.reserve(max_buf_size);
 
-        let len = cmp::min(bytes.remaining(), max_buf_size);
-
-        if self.buf.len() < len {
-            self.buf.reserve(len - self.buf.len());
-        }
-
-        unsafe {
-            self.buf.set_len(len);
-        }
-    }
-
-    pub(crate) fn read_from<T: Read>(&mut self, rd: &mut T) -> io::Result<usize> {
-        let res = uninterruptibly!(rd.read(&mut self.buf));
+        let buf = &mut self.buf.spare_capacity_mut()[..max_buf_size];
+        // SAFETY: The memory may be uninitialized, but `rd.read` will only write to the buffer.
+        let buf = unsafe { &mut *(buf as *mut [MaybeUninit<u8>] as *mut [u8]) };
+        let res = uninterruptibly!(rd.read(buf));
 
         if let Ok(n) = res {
-            self.buf.truncate(n);
+            // SAFETY: the caller promises that `rd.read` initializes
+            // a section of `buf` and correctly reports that length.
+            // The `self.is_empty()` assertion verifies that `n`
+            // equals the length of the `buf` capacity that was written
+            // to (and that `buf` isn't being shrunk).
+            unsafe { self.buf.set_len(n) }
         } else {
             self.buf.clear();
         }

tokio/src/io/mod.rs

@@ -231,6 +231,9 @@ cfg_io_driver_impl! {
     pub(crate) use poll_evented::PollEvented;
 }
 
+// The bsd module can't be build on Windows, so we completely ignore it, even
+// when building documentation.
+#[cfg(unix)]
 cfg_aio! {
     /// BSD-specific I/O types.
     pub mod bsd {

tokio/src/io/read_buf.rs

@@ -39,9 +39,11 @@ impl<'a> ReadBuf<'a> {
         }
     }
 
-    /// Creates a new `ReadBuf` from a fully uninitialized buffer.
+    /// Creates a new `ReadBuf` from a buffer that may be uninitialized.
     ///
-    /// Use `assume_init` if part of the buffer is known to be already initialized.
+    /// The internal cursor will mark the entire buffer as uninitialized. If
+    /// the buffer is known to be partially initialized, then use `assume_init`
+    /// to move the internal cursor.
     #[inline]
     pub fn uninit(buf: &'a mut [MaybeUninit<u8>]) -> ReadBuf<'a> {
         ReadBuf {

tokio/src/io/stderr.rs

@@ -67,8 +67,12 @@ cfg_io_std! {
     /// ```
     pub fn stderr() -> Stderr {
         let std = io::stderr();
+        // SAFETY: The `Read` implementation of `std` does not read from the
+        // buffer it is borrowing and correctly reports the length of the data
+        // written into the buffer.
+        let blocking = unsafe { Blocking::new(std) };
         Stderr {
-            std: SplitByUtf8BoundaryIfWindows::new(Blocking::new(std)),
+            std: SplitByUtf8BoundaryIfWindows::new(blocking),
         }
     }
 }