mpsc: add array-backed internals for bounded channel

sync: add blocking_acquire methods to Semaphore (#8269 )
io: handle empty vectored writes in simplex (#8353 )
2026-09-09 00:00:08 +02:00 · 2026-08-17 15:51:41 -04:00 · 2026-08-11 09:51:50 +02:00 · 2026-08-10 12:19:30 +02:00 · 2026-08-10 12:18:22 +02:00 · 2026-08-09 16:35:12 +00:00
230 changed files with 8389 additions and 3168 deletions
@@ -15,6 +15,8 @@ R-loom-time-driver:
  - any-glob-to-any-file:
    - tokio/src/runtime/time/*
    - tokio/src/runtime/time/**/*
+    - tokio/src/runtime/time_alt/*
+    - tokio/src/runtime/time_alt/**/*

 R-loom-current-thread:
 - changed-files:
@@ -20,5 +20,5 @@ jobs:
      issues: write
    runs-on: ubuntu-latest
    steps:
-    - uses: actions/checkout@v6
+    - uses: actions/checkout@v7
    - uses: EmbarkStudios/cargo-deny-action@v2
@@ -18,7 +18,7 @@ env:
  rust_stable: stable
  rust_nightly: nightly-2025-10-12
  # Pin a specific miri version
-  rust_miri_nightly: nightly-2026-05-20
+  rust_miri_nightly: nightly-2026-06-29
  rust_clippy: '1.88'
  # When updating this, also update:
  # - README.md
@@ -64,7 +64,7 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -99,7 +99,7 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -141,7 +141,7 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -177,7 +177,7 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -220,7 +220,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -246,7 +246,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -287,7 +287,7 @@ jobs:
          - { os: ubuntu-latest, extra_features: io-uring }
          - { os: macos-latest, extra_features: "" }
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -315,6 +315,30 @@ jobs:
          # the unstable cfg to RustDoc
          RUSTDOCFLAGS: --cfg tokio_unstable

+  # Run the test suite with the sharded `spawn_blocking` queue.
+  test-sharded-blocking-queue:
+    name: test tokio full with sharded blocking queue
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+      - 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 tokio full with sharded blocking queue
+        run: cargo nextest run --features full
+        working-directory: tokio
+        env:
+          TOKIO_UNSTABLE_SHARDED_BLOCKING_QUEUE: "1"
+
  test-unstable-taskdump:
    name: test tokio full --unstable --taskdump
    needs: basics
@@ -324,7 +348,7 @@ jobs:
        include:
          - os: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -358,7 +382,7 @@ jobs:
        include:
          - os: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -387,7 +411,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_miri_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -410,7 +434,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_miri_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -433,7 +457,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_miri_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -452,7 +476,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install llvm
        # Required to resolve symbols in sanitizer output
        run: sudo apt-get install -y llvm
@@ -474,7 +498,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Check `tokio` semver
        uses: obi1kenobi/cargo-semver-checks-action@v2
        with:
@@ -503,7 +527,7 @@ jobs:
          - powerpc64-unknown-linux-gnu
          - arm-linux-androideabi
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -528,7 +552,7 @@ jobs:
 #         - name: armv7-sony-vita-newlibeabihf
 #           exclude_features: "process,signal,rt-process-signal,full,taskdump"
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@nightly
        with:
@@ -562,7 +586,7 @@ jobs:
          - target: aarch64-pc-windows-msvc
            os: windows-11-arm
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust stable
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -612,7 +636,7 @@ jobs:
          - target: aarch64-pc-windows-msvc
            os: windows-11-arm
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust stable
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -653,7 +677,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -685,7 +709,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -701,8 +725,9 @@ jobs:
      # https://github.com/tokio-rs/tokio/pull/5356
      # https://github.com/tokio-rs/tokio/issues/5373
      - name: Check
-        # We use `--skip io-uring` since io-uring crate doesn't provide a binding for the i686 target.
-        run: cargo hack check -Zbuild-std --target target-specs/i686-unknown-linux-gnu.json -p tokio --feature-powerset --skip io-uring --depth 2 --keep-going
+        # We use `--skip io-uring,schedule-latency` since io-uring crate doesn't provide a binding for the i686 target
+        # and schedule latency tracking is only supported on 64-bit targets.
+        run: cargo hack check -Zbuild-std --target target-specs/i686-unknown-linux-gnu.json -p tokio --feature-powerset --skip io-uring,schedule-latency --depth 2 --keep-going
        env:
          RUSTFLAGS: --cfg tokio_unstable -Dwarnings

@@ -715,15 +740,15 @@ jobs:
        include:
          - name: ""
            rustflags: ""
-            exclude_features: "io-uring,taskdump"
+            exclude_features: "io-uring,taskdump,schedule-latency"
          - name: "--unstable"
            rustflags: "--cfg tokio_unstable -Dwarnings"
-            exclude_features: "io-uring,taskdump"
-          - name: "--unstable io-uring,taskdump"
+            exclude_features: "io-uring,taskdump,schedule-latency"
+          - name: "--unstable io-uring,taskdump,schedule-latency"
            rustflags: "--cfg tokio_unstable -Dwarnings"
            exclude_features: ""
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -742,7 +767,7 @@ jobs:
    name: minrust
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_min }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -771,7 +796,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -804,7 +829,7 @@ jobs:
    name: fmt
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -824,7 +849,7 @@ jobs:
    name: clippy
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_clippy }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -864,7 +889,7 @@ jobs:
            extra_features: "tracing,io-uring,taskdump"

    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -881,7 +906,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -897,7 +922,7 @@ jobs:
    name: Check README
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Verify that both READMEs are identical
        run: diff README.md tokio/README.md

@@ -916,7 +941,7 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -970,7 +995,7 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1016,7 +1041,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1033,7 +1058,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -1057,7 +1082,7 @@ jobs:
          - name: macros sync time rt
            features: "macros sync time rt"
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1080,7 +1105,7 @@ jobs:
          - wasm32-wasip1
          - wasm32-wasip1-threads
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1091,7 +1116,10 @@ jobs:
      - name: Install cargo-hack, wasmtime
        uses: taiki-e/install-action@v2
        with:
-          tool: cargo-hack,wasmtime
+          # Wasmtime v46.0.1 is the last version to support
+          # `wasm32-wasip1-threads` (which was an experiment that was never
+          # standardized):
+          tool: cargo-hack,[email protected]

      - uses: Swatinem/rust-cache@v2
      - name: WASI test tokio full
@@ -1141,7 +1169,7 @@ jobs:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1151,7 +1179,9 @@ jobs:
      - name: Install cargo-nextest, wasmtime
        uses: taiki-e/install-action@v2
        with:
-          tool: cargo-nextest,wasmtime-cli
+          # Wasmtime v47.0.0 or later should work, but we stick with a known,
+          # specific version to avoid surprises:
+          tool: cargo-nextest,[email protected]

      - uses: Swatinem/rust-cache@v2

@@ -1176,7 +1206,7 @@ jobs:
            # includes all unstable features.
            extra_features: "tracing,io-uring,taskdump"
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ matrix.rust }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1200,7 +1230,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_nightly }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1220,7 +1250,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@stable
        with:
@@ -1229,7 +1259,7 @@ jobs:
        uses: taiki-e/install-action@v2
        with:
          tool: cargo-spellcheck
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Make sure dictionary words are sorted and unique
        run: |
          FILE="spellcheck.dic"
@@ -1317,7 +1347,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Test in FreeBSD
        uses: vmactions/freebsd-vm@v1
        with:
@@ -1343,7 +1373,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Build docs in FreeBSD
        uses: vmactions/freebsd-vm@v1
        env:
@@ -1371,7 +1401,7 @@ jobs:
    needs: basics
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Test in FreeBSD
        uses: vmactions/freebsd-vm@v1
        with:
@@ -19,7 +19,7 @@ jobs:
    runs-on: ubuntu-latest
    if: github.repository_owner == 'tokio-rs'
    steps:
-    - uses: actions/labeler@v6
+    - uses: actions/labeler@v7
      with:
        repo-token: "${{ secrets.GITHUB_TOKEN }}"
        sync-labels: true
@@ -28,8 +28,15 @@ jobs:
    # base_ref is null when it's not a pull request
    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-blocking') || (github.base_ref == null))
    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        # Run the blocking pool loom tests against both `spawn_blocking`
+        # queue implementations.
+        include:
+          - sharded_blocking_queue: "0"
+          - sharded_blocking_queue: "1"
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -38,6 +45,8 @@ jobs:
      - name: run tests
        run: cargo test --lib --release --features full -- --nocapture loom_blocking
        working-directory: tokio
+        env:
+          TOKIO_UNSTABLE_SHARDED_BLOCKING_QUEUE: ${{ matrix.sharded_blocking_queue }}

  loom-sync:
    name: loom tokio::sync
@@ -45,7 +54,7 @@ jobs:
    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-sync') || (github.base_ref == null))
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -61,7 +70,7 @@ jobs:
    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-time-driver') || (github.base_ref == null))
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -77,7 +86,7 @@ jobs:
    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-current-thread') || (github.base_ref == null))
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -100,7 +109,7 @@ jobs:
          - scope: loom_multi_thread::group_c
          - scope: loom_multi_thread::group_d
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -118,7 +127,7 @@ jobs:
    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-util') || (github.base_ref == null))
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -19,5 +19,5 @@ jobs:
  cargo-deny:
    runs-on: ubuntu-latest
    steps:
-    - uses: actions/checkout@v6
+    - uses: actions/checkout@v7
    - uses: EmbarkStudios/cargo-deny-action@v2
@@ -27,7 +27,7 @@ jobs:
        stress-test:
          - simple_echo_tcp
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
      - name: Install Rust ${{ env.rust_stable }}
        uses: dtolnay/rust-toolchain@master
        with:
@@ -14,7 +14,7 @@ jobs:
    env:
      KERNEL_VERSION: ${{ inputs.kernel_version }}
    steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7

      - name: Install system dependencies
        run: |
@@ -25,7 +25,7 @@ jobs:

      - name: Cache Linux source
        id: cache-kernel
-        uses: actions/cache@v5
+        uses: actions/cache@v6
        with:
          path: linux-${{ env.KERNEL_VERSION }}
          key: kernel-${{ env.KERNEL_VERSION }}
@@ -89,16 +89,31 @@ jobs:
            | cpio --null -ov --format=newc | gzip -9 > ../initramfs.cpio.gz)

      - name: Run tests in QEMU
+        shell: bash
        run: |
+          set -euo pipefail
+
+          set +e
          qemu-system-x86_64 \
            -kernel linux-${{ env.KERNEL_VERSION }}/arch/x86/boot/bzImage \
            -initrd initramfs.cpio.gz \
            -append "console=ttyS0 rootfstype=ramfs panic=1" \
            -nographic -no-reboot -m 1024 -action panic=exit-failure 2>&1 | tee qemu-output.log
+          qemu_status=${PIPESTATUS[0]} tee_status=${PIPESTATUS[1]}
+          set -e
+
+          if [ "$qemu_status" -ne 0 ]; then
+            echo "QEMU exited with status $qemu_status"
+            exit "$qemu_status"
+          fi
+
+          if [ "$tee_status" -ne 0 ]; then
+            echo "tee exited with status $tee_status"
+            exit "$tee_status"
+          fi

-          # qemu always exits with 0, so we check if the tests passed by using grep.
          if grep -q "test result: FAILED" qemu-output.log; then
-            echo "tests failed (QEMU exited abnormally)"
+            echo "tests reported failures"
            exit 1
          else
            echo "all tests passed"
@@ -1,7 +1,3 @@
-*[TokioConf 2026 program and tickets are now available!](https://tokioconf.com)*
-
---
-
 # Tokio

 A runtime for writing reliable, asynchronous, and slim applications with
@@ -60,7 +56,7 @@ Make sure you enable the full features of the tokio crate on Cargo.toml:

 ```toml
 [dependencies]
-tokio = { version = "1.52.3", features = ["full"] }
+tokio = { version = "1.53.1", features = ["full"] }
 ```
 Then, on your main.rs:

@@ -73,7 +73,7 @@ impl SlowHddWriter {
    fn write_bytes(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
-        writeable: usize,
+        writable: usize,
    ) -> std::task::Poll<Result<usize, std::io::Error>> {
        let service_res = self.as_mut().service_write(cx);

@@ -86,7 +86,7 @@ impl SlowHddWriter {
            assert!(service_res.is_pending());
            Poll::Pending
        } else {
-            let written = available.min(writeable);
+            let written = available.min(writable);
            self.buffer_used += written;
            Poll::Ready(Ok(written))
        }
@@ -123,8 +123,8 @@ impl AsyncWrite for SlowHddWriter {
        cx: &mut std::task::Context<'_>,
        bufs: &[std::io::IoSlice<'_>],
    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        let writeable = bufs.iter().fold(0, |acc, buf| acc + buf.len());
-        self.write_bytes(cx, writeable)
+        let writable = bufs.iter().fold(0, |acc, buf| acc + buf.len());
+        self.write_bytes(cx, writable)
    }

    fn is_write_vectored(&self) -> bool {
@@ -84,9 +84,7 @@ fn remote_spawn_contention(c: &mut Criterion) {
 }

 fn parallelism_levels() -> Vec<usize> {
-    let max_parallelism = std::thread::available_parallelism()
-        .map(|p| p.get())
-        .unwrap_or(1);
+    let max_parallelism = std::thread::available_parallelism().map_or(1, |p| p.get());

    [1, 2, 4, 8, 16, 32, 64]
        .into_iter()
@@ -15,9 +15,7 @@ const NUM_BATCHES: usize = 100;
 const BATCH_SIZE: usize = 16;

 fn spawn_blocking_concurrency(c: &mut Criterion) {
-    let max_parallelism = std::thread::available_parallelism()
-        .map(|p| p.get())
-        .unwrap_or(1);
+    let max_parallelism = std::thread::available_parallelism().map_or(1, |p| p.get());

    let parallelism_levels: Vec<usize> = [1, 2, 4, 8, 16, 32, 64]
        .into_iter()
@@ -19,6 +19,15 @@ impl Default for Large {
    }
 }

+#[cfg(target_pointer_width = "64")]
+const ARRAY_CAP: usize = 32;
+
+#[cfg(not(target_pointer_width = "64"))]
+const ARRAY_CAP: usize = 16;
+
+const LIST_CAP: usize = ARRAY_CAP + 1;
+const ROUNDTRIP_ITERS: usize = 1_000;
+
 fn rt() -> tokio::runtime::Runtime {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(6)
@@ -34,6 +43,14 @@ fn create_medium<const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>) {
    });
 }

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

@@ -48,6 +65,22 @@ fn send_data<T: Default, const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>, pr
    });
 }

+fn roundtrip_try_send_recv_data<T: Default, const SIZE: usize>(
+    g: &mut BenchmarkGroup<WallTime>,
+    prefix: &str,
+) {
+    let (tx, mut rx) = mpsc::channel::<T>(SIZE);
+
+    g.bench_function(format!("{prefix}_{SIZE}"), |b| {
+        b.iter(|| {
+            for _ in 0..ROUNDTRIP_ITERS {
+                tx.try_send(T::default()).unwrap();
+                black_box(rx.try_recv().unwrap());
+            }
+        })
+    });
+}
+
 fn contention_bounded(g: &mut BenchmarkGroup<WallTime>) {
    let rt = rt();

@@ -296,13 +329,39 @@ fn bench_create_medium(c: &mut Criterion) {
    group.finish();
 }

+fn bench_create_small(c: &mut Criterion) {
+    let mut group = c.benchmark_group("create_small");
+    create_data::<Medium, 1>(&mut group, "medium");
+    create_data::<Medium, 10>(&mut group, "medium");
+    create_data::<Medium, ARRAY_CAP>(&mut group, "medium");
+    create_data::<Medium, LIST_CAP>(&mut group, "medium");
+    create_data::<Large, 1>(&mut group, "large");
+    create_data::<Large, 10>(&mut group, "large");
+    create_data::<Large, ARRAY_CAP>(&mut group, "large");
+    create_data::<Large, LIST_CAP>(&mut group, "large");
+    group.finish();
+}
+
 fn bench_send(c: &mut Criterion) {
    let mut group = c.benchmark_group("send");
+    send_data::<Medium, 1>(&mut group, "medium");
+    send_data::<Medium, 10>(&mut group, "medium");
    send_data::<Medium, 1000>(&mut group, "medium");
+    send_data::<Large, 1>(&mut group, "large");
+    send_data::<Large, 10>(&mut group, "large");
    send_data::<Large, 1000>(&mut group, "large");
    group.finish();
 }

+fn bench_roundtrip_try_send_recv(c: &mut Criterion) {
+    let mut group = c.benchmark_group("roundtrip_try_send_recv");
+    roundtrip_try_send_recv_data::<Medium, 1>(&mut group, "medium");
+    roundtrip_try_send_recv_data::<Medium, 10>(&mut group, "medium");
+    roundtrip_try_send_recv_data::<Medium, ARRAY_CAP>(&mut group, "medium");
+    roundtrip_try_send_recv_data::<Medium, LIST_CAP>(&mut group, "medium");
+    group.finish();
+}
+
 fn bench_contention(c: &mut Criterion) {
    let mut group = c.benchmark_group("contention");
    contention_bounded(&mut group);
@@ -324,8 +383,17 @@ fn bench_uncontented(c: &mut Criterion) {
 }

 criterion_group!(create, bench_create_medium);
+criterion_group!(create_small, bench_create_small);
 criterion_group!(send, bench_send);
+criterion_group!(roundtrip_try_send_recv, bench_roundtrip_try_send_recv);
 criterion_group!(contention, bench_contention);
 criterion_group!(uncontented, bench_uncontented);

-criterion_main!(create, send, contention, uncontented);
+criterion_main!(
+    create,
+    create_small,
+    send,
+    roundtrip_try_send_recv,
+    contention,
+    uncontented
+);
@@ -22,6 +22,7 @@ If you are unsure where to begin, use the following guides:
    - [Triaging a Bug Report](contributing-in-issues.md#triaging-a-bug-report)
    - [Resolving a Bug Report](contributing-in-issues.md#resolving-a-bug-report)
 - [Pull Requests](pull-requests.md)
+    - [Step-by-Step Contribution Workflow](pull-requests.md#step-by-step-contribution-workflow)
    - [Cargo Commands](pull-requests.md#cargo-commands)
    - [Performing spellcheck on Tokio codebase](pull-requests.md#performing-spellcheck-on-tokio-codebase)
    - [Tests](pull-requests.md#tests)
@@ -11,6 +11,65 @@ documentation) are greatly appreciated.
 > issue describing the change to solicit feedback and guidance.
 > This will increase the likelihood of the PR getting merged.

+### Step-by-Step Contribution Workflow
+
+Once you've found an issue you'd like to work on, follow these steps before opening a Pull Request. This workflow provides a chronological overview of the contribution process and points to the relevant documentation where additional detail is available.
+
+#### 1. Fork and Clone the Repository
+
+Fork the repository to your GitHub account and clone your fork locally.
+
+If you've previously cloned the repository, ensure your local copy is up to date before creating a new branch.
+
+```bash
+git checkout master
+git pull upstream master
+```
+
+---
+
+#### 2. Create a Feature Branch
+
+Never commit directly to your local `master` branch. Instead, create a descriptive feature branch for each change you work on.
+
+```bash
+git checkout -b my-feature
+```
+
+Using a dedicated branch keeps your default branch clean and makes it easier to update your Pull Request during code review.
+
+---
+
+#### 3. Implement Your Changes
+
+Make the code or documentation changes needed. Check your work with `git diff` to confirm the changes look correct before moving on.
+
+If your changes introduce new functionality or modify existing behavior, consider whether additional tests or documentation should also be added.
+
+---
+
+#### 4. Verify Your Changes
+
+Before opening a Pull Request, run the project's verification steps. Refer to the sections below for details on when each command should be used.
+
+---
+
+#### 5. Commit and Push Your Changes
+
+Commit your changes following the [project's commit message guidelines](#commit-message-guidelines).
+
+```bash
+git add .
+git commit -m "module: describe your change"
+git push origin my-feature
+```
+
+---
+
+#### 6. Open a Pull Request
+
+Open a Pull Request from your feature branch to the main repository. See [Opening the Pull Request](#opening-the-pull-request) for what to include and how the review process works.
+
 ### Cargo Commands

 Due to the extensive use of features in Tokio, you will often need to add extra
@@ -5,7 +5,12 @@
 #[cfg(all(
    tokio_unstable,
    target_os = "linux",
-    any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
+    any(
+        target_arch = "aarch64",
+        target_arch = "x86",
+        target_arch = "x86_64",
+        target_arch = "s390x"
+    )
 ))]
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -82,7 +87,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 #[cfg(not(all(
    tokio_unstable,
    target_os = "linux",
-    any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
+    any(
+        target_arch = "aarch64",
+        target_arch = "x86",
+        target_arch = "x86_64",
+        target_arch = "s390x"
+    )
 )))]
 fn main() {
    println!("task dumps are not available")
@@ -9,6 +9,28 @@
 //!     cargo run --example connect-udp 127.0.0.1:8080
 //!
 //! Each line you type in to the `connect-udp` terminal should be echo'd back to you!
+//!
+//! # Binding to all interfaces
+//!
+//! By default this example binds to `127.0.0.1` so it is only reachable
+//! from the local machine.
+//!
+//! To listen on all interfaces instead:
+//!
+//! ```sh
+//! cargo run --example echo-udp -- 0.0.0.0:8080
+//! ```
+//!
+//! Binding to `0.0.0.0` exposes the server on all network interfaces.
+//! Only do this in trusted network environments.
+//!
+//! On multi-homed systems, a UDP socket bound to a wildcard address
+//! (`0.0.0.0` or `::`) cannot always send replies from the same local IP
+//! that received the packet. Replies may therefore originate from a
+//! different address than the client targeted. See [this Cloudflare blog
+//! post][udp-blog] for more details.
+//!
+//! [udp-blog]: https://blog.cloudflare.com/everything-you-ever-wanted-to-know-about-udp-sockets-but-were-afraid-to-ask-part-1

 #![warn(rust_2018_idioms)]

@@ -25,7 +25,6 @@
 use tokio::io::copy_bidirectional;
 use tokio::net::{TcpListener, TcpStream};

-use futures::FutureExt;
 use std::env;
 use std::error::Error;

@@ -44,16 +43,20 @@ async fn main() -> Result<(), Box<dyn Error>> {
    let listener = TcpListener::bind(listen_addr).await?;

    while let Ok((mut inbound, _)) = listener.accept().await {
-        let mut outbound = TcpStream::connect(server_addr.clone()).await?;
+        let server_addr = server_addr.clone();

        tokio::spawn(async move {
-            copy_bidirectional(&mut inbound, &mut outbound)
-                .map(|r| {
-                    if let Err(e) = r {
-                        println!("Failed to transfer; error={e}");
-                    }
-                })
-                .await
+            let mut outbound = match TcpStream::connect(server_addr).await {
+                Ok(outbound) => outbound,
+                Err(e) => {
+                    println!("Failed to connect; error={e}");
+                    return;
+                }
+            };
+
+            if let Err(e) = copy_bidirectional(&mut inbound, &mut outbound).await {
+                println!("Failed to transfer; error={e}");
+            }
        });
    }

@@ -264,8 +264,7 @@ mod date {
                let now = SystemTime::now();
                let now_unix = now
                    .duration_since(SystemTime::UNIX_EPOCH)
-                    .map(|since_epoch| since_epoch.as_secs())
-                    .unwrap_or(0);
+                    .map_or(0, |since_epoch| since_epoch.as_secs());
                if cache.unix_date != now_unix {
                    cache.update(now, now_unix);
                }
@@ -1,4 +1,4 @@
-324
+327
 &
 +
 <
@@ -141,6 +141,7 @@ hashsets
 HdrHistogram
 ICMP
 ie
+iff
 Illumos
 impl
 implementers
@@ -229,6 +230,7 @@ reregistering
 resize
 resized
 RMW
+RNG
 runtime
 runtime's
 runtimes
@@ -244,6 +246,7 @@ signalling
 SmallCrush
 Solaris
 spawner
+spawners
 Splitter
 spmc
 spsc
@@ -16,7 +16,6 @@ async fn spawning() -> usize {
    join.await.unwrap()
 }

-#[cfg(tokio_unstable)]
 #[tokio::main(flavor = "local")]
 async fn local_main() -> usize {
    let join = tokio::task::spawn_local(async { 1 });
@@ -33,6 +32,5 @@ fn shell() {
    assert_eq!(1, basic_main());
    assert_eq!(bool::default(), generic_fun::<bool>());

-    #[cfg(tokio_unstable)]
    assert_eq!(1, local_main());
 }
@@ -1,3 +1,15 @@
+# 2.7.2 (July 29th, 2026)
+
+- macros: upgrade syn to v3 ([#8304])
+
+[#8304]: https://github.com/tokio-rs/tokio/pull/8304
+
+# 2.7.1 (July 17th, 2026)
+
+- macros: clarify `tokio::main` expansion ([#8193])
+
+[#8193]: https://github.com/tokio-rs/tokio/pull/8193
+
 # 2.7.0 (April 3rd, 2026)

 - macros: stabilize `LocalRuntime` ([#7557])
@@ -4,7 +4,7 @@ name = "tokio-macros"
 # - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-macros-x.y.z" git tag.
-version = "2.7.0"
+version = "2.7.2"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -24,7 +24,7 @@ proc-macro = true
 [dependencies]
 proc-macro2 = "1.0.60"
 quote = "1"
-syn = { version = "2.0", features = ["full"] }
+syn = { version = "3.0", features = ["full"] }

 [dev-dependencies]
 tokio = { version = "1.0.0", features = ["full", "test-util"] }
@@ -326,7 +326,7 @@ fn contains_impl_trait(ty: &syn::Type) -> bool {
                    _ => false,
                }),
                syn::PathArguments::Parenthesized(args) => {
-                    args.inputs.iter().any(contains_impl_trait)
+                    args.inputs.iter().any(|arg| contains_impl_trait(&arg.ty))
                        || matches!(&args.output, syn::ReturnType::Type(_, t) if contains_impl_trait(t))
                }
                syn::PathArguments::None => false,
@@ -453,13 +453,13 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
        (start, end)
    };

-    let crate_path = config
-        .crate_name
-        .map(ToTokens::into_token_stream)
-        .unwrap_or_else(|| {
+    let crate_path = config.crate_name.map_or_else(
+        || {
            Ident::new("tokio", Span::call_site().located_at(last_stmt_start_span))
                .into_token_stream()
-        });
+        },
+        ToTokens::into_token_stream,
+    );

    let use_builder = quote_spanned! {Span::call_site().located_at(last_stmt_start_span)=>
        use #crate_path::runtime::Builder;
@@ -29,10 +29,11 @@ use proc_macro::TokenStream;
 /// powerful interface.
 ///
 /// Note: This macro can be used on any function and not just the `main`
-/// function. Using it on a non-main function makes the function behave as if it
-/// was synchronous by starting a new runtime each time it is called. If the
-/// function is called often, it is preferable to create the runtime using the
-/// runtime builder so the runtime can be reused across calls.
+/// function. Although the function is written with `async fn`, this macro
+/// expands it to a synchronous function that starts a runtime each time it is
+/// called. If the function is called often, it is preferable to create the
+/// runtime using the runtime builder so the runtime can be reused across calls.
+/// For details on the expansion, see [Bridging with sync code][bridging].
 ///
 /// # Non-worker async function
 ///
@@ -308,6 +309,7 @@ use proc_macro::TokenStream;
 /// [`Builder::unhandled_panic`]: ../tokio/runtime/struct.Builder.html#method.unhandled_panic
 /// [unstable]: ../tokio/index.html#unstable-features
 /// [local runtime]: ../tokio/runtime/struct.LocalRuntime.html
+/// [bridging]: https://tokio.rs/tokio/topics/bridging#what-tokiomain-expands-to
 #[proc_macro_attribute]
 pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::main(args.into(), item.into(), true).into()
@@ -1,3 +1,38 @@
+# 0.1.19 (July 22nd, 2026)
+
+### Added
+
+- stream: implement `FromStream` for standard collections ([#7954], [#7966])
+- stream: implement `FusedStream` for various stream adaptors ([#7854], [#8090], [#8096])
+- stream: implement `Peekable::size_hint` ([#8109])
+- task: add `JoinSetStream` wrapper for `JoinSet` ([#8189])
+
+### Changed
+
+- stream: bump minimum required `tokio` version to `1.38.0` ([#7887])
+- stream: use cooperative budgeting in `tokio_stream::iter` ([#8218])
+- stream: update coop handling for `empty` and `once` ([#8227])
+
+### Fixed
+
+- stream: fix overflow in `StreamMap::size_hint` ([#8216])
+- stream: honor `StreamMap::next_many` limit ([#8215])
+- stream: stop polling underlying stream once `map_while` yields `None` ([#8233])
+
+[#7854]: https://github.com/tokio-rs/tokio/pull/7854
+[#7887]: https://github.com/tokio-rs/tokio/pull/7887
+[#7954]: https://github.com/tokio-rs/tokio/pull/7954
+[#7966]: https://github.com/tokio-rs/tokio/pull/7966
+[#8090]: https://github.com/tokio-rs/tokio/pull/8090
+[#8096]: https://github.com/tokio-rs/tokio/pull/8096
+[#8109]: https://github.com/tokio-rs/tokio/pull/8109
+[#8189]: https://github.com/tokio-rs/tokio/pull/8189
+[#8215]: https://github.com/tokio-rs/tokio/pull/8215
+[#8216]: https://github.com/tokio-rs/tokio/pull/8216
+[#8218]: https://github.com/tokio-rs/tokio/pull/8218
+[#8227]: https://github.com/tokio-rs/tokio/pull/8227
+[#8233]: https://github.com/tokio-rs/tokio/pull/8233
+
 # 0.1.18 (January 4th, 2026)

 ### Added
@@ -4,7 +4,7 @@ name = "tokio-stream"
 # - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.18"
+version = "0.1.19"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -24,6 +24,7 @@ full = [
    "net",
    "io-util",
    "fs",
+    "rt",
    "sync",
    "signal"
 ]
@@ -32,6 +33,7 @@ time = ["tokio/time"]
 net = ["tokio/net"]
 io-util = ["tokio/io-util"]
 fs = ["tokio/fs"]
+rt = ["tokio/rt"]
 sync = ["tokio/sync", "tokio-util"]
 signal = ["tokio/signal"]

@@ -40,7 +40,15 @@ pub const fn empty<T>() -> Empty<T> {
 impl<T> Stream for Empty<T> {
    type Item = T;

-    fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> {
+    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<T>> {
+        #[cfg(feature = "rt")]
+        {
+            use tokio::task::coop;
+
+            let coop = std::task::ready!(coop::poll_proceed(_cx));
+            coop.made_progress();
+        }
+
        Poll::Ready(None)
    }

@@ -8,6 +8,7 @@ use core::task::{Context, Poll};
 #[must_use = "streams do nothing unless polled"]
 pub struct Iter<I> {
    iter: I,
+    #[cfg(not(feature = "rt"))]
    yield_amt: usize,
 }

@@ -36,6 +37,7 @@ where
 {
    Iter {
        iter: i.into_iter(),
+        #[cfg(not(feature = "rt"))]
        yield_amt: 0,
    }
 }
@@ -47,17 +49,33 @@ where
    type Item = I::Item;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<I::Item>> {
-        // TODO: add coop back
-        if self.yield_amt >= 32 {
-            self.yield_amt = 0;
+        #[cfg(feature = "rt")]
+        {
+            use tokio::task::coop;

-            cx.waker().wake_by_ref();
+            let coop = std::task::ready!(coop::poll_proceed(cx));
+            let item = self.iter.next();

-            Poll::Pending
-        } else {
-            self.yield_amt += 1;
+            coop.made_progress();

-            Poll::Ready(self.iter.next())
+            Poll::Ready(item)
+        }
+
+        #[cfg(not(feature = "rt"))]
+        {
+            if self.yield_amt >= 32 {
+                self.yield_amt = 0;
+
+                cx.waker().wake_by_ref();
+
+                Poll::Pending
+            } else {
+                let item = self.iter.next();
+
+                self.yield_amt += 1;
+
+                Poll::Ready(item)
+            }
        }
    }

@@ -57,3 +57,13 @@ macro_rules! cfg_signal {
        )*
    }
 }
+
+macro_rules! cfg_rt {
+    ($($item:item)*) => {
+        $(
+            #[cfg(feature = "rt")]
+            #[cfg_attr(docsrs, doc(cfg(feature = "rt")))]
+            $item
+        )*
+    }
+}
@@ -1,6 +1,5 @@
-use crate::{Iter, Stream};
+use crate::Stream;

-use core::option;
 use core::pin::Pin;
 use core::task::{Context, Poll};

@@ -8,7 +7,7 @@ use core::task::{Context, Poll};
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
 pub struct Once<T> {
-    iter: Iter<option::IntoIter<T>>,
+    value: Option<T>,
 }

 impl<I> Unpin for Once<I> {}
@@ -34,19 +33,30 @@ impl<I> Unpin for Once<I> {}
 /// # }
 /// ```
 pub fn once<T>(value: T) -> Once<T> {
-    Once {
-        iter: crate::iter(Some(value)),
-    }
+    Once { value: Some(value) }
 }

 impl<T> Stream for Once<T> {
    type Item = T;

-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
-        Pin::new(&mut self.iter).poll_next(cx)
+    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<T>> {
+        #[cfg(feature = "rt")]
+        {
+            use tokio::task::coop;
+
+            let coop = std::task::ready!(coop::poll_proceed(_cx));
+
+            coop.made_progress();
+        }
+
+        Poll::Ready(self.value.take())
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
-        self.iter.size_hint()
+        if self.value.is_some() {
+            (1, Some(1))
+        } else {
+            (0, Some(0))
+        }
    }
 }
@@ -31,6 +31,32 @@ impl<St, F> Filter<St, F> {
    pub(super) fn new(stream: St, f: F) -> Self {
        Self { stream, f }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St, F> Stream for Filter<St, F>
@@ -31,6 +31,32 @@ impl<St, F> FilterMap<St, F> {
    pub(super) fn new(stream: St, f: F) -> Self {
        Self { stream, f }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St, F, T> Stream for FilterMap<St, F>
@@ -29,6 +29,32 @@ impl<St, F> Map<St, F> {
    pub(super) fn new(stream: St, f: F) -> Self {
        Map { stream, f }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St, F, T> Stream for Map<St, F>
@@ -3,6 +3,7 @@ use crate::Stream;
 use core::fmt;
 use core::pin::Pin;
 use core::task::{Context, Poll};
+use futures_core::FusedStream;
 use pin_project_lite::pin_project;

 pin_project! {
@@ -12,6 +13,7 @@ pin_project! {
        #[pin]
        stream: St,
        f: F,
+        done: bool,
    }
 }

@@ -22,13 +24,44 @@ where
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MapWhile")
            .field("stream", &self.stream)
+            .field("done", &self.done)
            .finish()
    }
 }

 impl<St, F> MapWhile<St, F> {
    pub(super) fn new(stream: St, f: F) -> Self {
-        MapWhile { stream, f }
+        MapWhile {
+            stream,
+            f,
+            done: false,
+        }
+    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
    }
 }

@@ -41,12 +74,36 @@ where

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
        let me = self.project();
+        if *me.done {
+            return Poll::Ready(None);
+        }
+
        let f = me.f;
-        me.stream.poll_next(cx).map(|opt| opt.and_then(f))
+        let done = me.done;
+        me.stream.poll_next(cx).map(|opt| {
+            let mapped = opt.and_then(f);
+            if mapped.is_none() {
+                *done = true;
+            }
+            mapped
+        })
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.done {
+            return (0, Some(0));
+        }
+
        let (_, upper) = self.stream.size_hint();
        (0, upper)
    }
 }
+
+impl<St, F> FusedStream for MapWhile<St, F>
+where
+    Self: Stream,
+{
+    fn is_terminated(&self) -> bool {
+        self.done
+    }
+}
@@ -34,6 +34,33 @@ impl<T: Stream> Peekable<T> {
            self.peek.as_ref()
        }
    }
+
+    /// Peek at the next item in the stream as a mutable reference.
+    pub async fn peek_mut(&mut self) -> Option<&mut T::Item>
+    where
+        T: Unpin,
+    {
+        if let Some(ref mut it) = self.peek {
+            Some(it)
+        } else {
+            self.peek = self.next().await;
+            self.peek.as_mut()
+        }
+    }
+
+    /// Poll to peek at the next item in the stream as a mutable reference.
+    pub fn poll_peek(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<&mut T::Item>> {
+        let mut this = self.project();
+
+        if this.peek.is_none() {
+            match this.stream.as_mut().poll_next(cx) {
+                Poll::Ready(item) => *this.peek = item,
+                Poll::Pending => return Poll::Pending,
+            }
+        }
+
+        Poll::Ready(this.peek.as_mut())
+    }
 }

 impl<T: Stream> Stream for Peekable<T> {
@@ -47,4 +74,12 @@ impl<T: Stream> Stream for Peekable<T> {
            this.stream.poll_next(cx)
        }
    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let peek_len = if self.peek.is_some() { 1 } else { 0 };
+        let (lo, hi) = self.stream.size_hint();
+        let lo = lo.saturating_add(peek_len);
+        let hi = hi.and_then(|x| x.checked_add(peek_len));
+        (lo, hi)
+    }
 }
@@ -31,6 +31,32 @@ impl<St> Skip<St> {
    pub(super) fn new(stream: St, remaining: usize) -> Self {
        Self { stream, remaining }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St> Stream for Skip<St>
@@ -34,6 +34,32 @@ impl<St, F> SkipWhile<St, F> {
            predicate: Some(predicate),
        }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St, F> Stream for SkipWhile<St, F>
@@ -32,6 +32,32 @@ impl<St> Take<St> {
    pub(super) fn new(stream: St, remaining: usize) -> Self {
        Self { stream, remaining }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St> Stream for Take<St>
@@ -37,6 +37,32 @@ impl<St, F> TakeWhile<St, F> {
            done: false,
        }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St, F> Stream for TakeWhile<St, F>
@@ -49,13 +75,7 @@ where
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if !*self.as_mut().project().done {
            self.as_mut().project().stream.poll_next(cx).map(|ready| {
-                let ready = ready.and_then(|item| {
-                    if !(self.as_mut().project().predicate)(&item) {
-                        None
-                    } else {
-                        Some(item)
-                    }
-                });
+                let ready = ready.filter(self.as_mut().project().predicate);

                if ready.is_none() {
                    *self.as_mut().project().done = true;
@@ -38,6 +38,32 @@ impl<St, Fut, F> Then<St, Fut, F> {
            f,
        }
    }
+
+    /// Returns a reference to the inner stream.
+    pub fn get_ref(&self) -> &St {
+        &self.stream
+    }
+
+    /// Returns a mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut St {
+        &mut self.stream
+    }
+
+    /// Returns a pinned mutable reference to the inner stream.
+    ///
+    /// Mutating the inner stream may confuse this combinator.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut St> {
+        self.project().stream
+    }
+
+    /// Consumes this combinator and returns the inner stream.
+    ///
+    /// This may discard intermediate combinator state.
+    pub fn into_inner(self) -> St {
+        self.stream
+    }
 }

 impl<St, F, Fut> Stream for Then<St, Fut, F>
@@ -632,6 +632,10 @@ where
                        should_loop = true;

                        idx = idx.wrapping_add(1) % self.entries.len();
+
+                        if added == limit {
+                            break;
+                        }
                    }
                    Poll::Ready(None) => {
                        // Remove the entry
@@ -685,15 +689,15 @@ where
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
-        let mut ret = (0, Some(0));
+        let mut ret: (usize, Option<usize>) = (0, Some(0));

        for (_, stream) in &self.entries {
            let hint = stream.size_hint();

-            ret.0 += hint.0;
+            ret.0 = ret.0.saturating_add(hint.0);

            match (ret.1, hint.1) {
-                (Some(a), Some(b)) => ret.1 = Some(a + b),
+                (Some(a), Some(b)) => ret.1 = a.checked_add(b),
                (Some(_), None) => ret.1 = None,
                _ => {}
            }
@@ -13,6 +13,11 @@ pub use mpsc_bounded::ReceiverStream;
 mod mpsc_unbounded;
 pub use mpsc_unbounded::UnboundedReceiverStream;

+cfg_rt! {
+    mod task;
+    pub use task::JoinSetStream;
+}
+
 cfg_sync! {
    mod broadcast;
    pub use broadcast::BroadcastStream;
@@ -0,0 +1,78 @@
+use crate::Stream;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::task::{JoinError, JoinSet};
+
+/// A wrapper around [`tokio::task::JoinSet`] that implements [`Stream`].
+///
+/// # Example
+///
+/// ```
+/// use tokio::task::JoinSet;
+/// use tokio_stream::wrappers::JoinSetStream;
+/// use tokio_stream::StreamExt;
+///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> Result<(), tokio::task::JoinError> {
+/// let set: JoinSet<_> = (0..2).map(|i| async move { i }).collect();
+///
+/// let mut stream = JoinSetStream::new(set);
+/// assert_eq!(stream.next().await.transpose()?, Some(0));
+/// assert_eq!(stream.next().await.transpose()?, Some(1));
+/// assert_eq!(stream.next().await.transpose()?, None);
+/// # Ok(())
+/// # }
+/// ```
+///
+/// [`tokio::task::JoinSet`]: struct@tokio::task::JoinSet
+/// [`Stream`]: trait@crate::Stream
+#[derive(Debug)]
+pub struct JoinSetStream<T> {
+    inner: JoinSet<T>,
+}
+
+impl<T> JoinSetStream<T> {
+    /// Create a new `JoinSetStream`.
+    pub fn new(join_set: JoinSet<T>) -> Self {
+        Self { inner: join_set }
+    }
+
+    /// Get back the inner `JoinSet`.
+    pub fn into_inner(self) -> JoinSet<T> {
+        self.inner
+    }
+}
+
+impl<T: 'static> Stream for JoinSetStream<T> {
+    type Item = Result<T, JoinError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.inner.poll_join_next(cx)
+    }
+
+    /// Returns the bounds of the stream based on the underlying `JoinSet`.
+    ///
+    /// It returns `(set.len(), Some(set.len()))`.
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let size = self.inner.len();
+        (size, Some(size))
+    }
+}
+
+impl<T> AsRef<JoinSet<T>> for JoinSetStream<T> {
+    fn as_ref(&self) -> &JoinSet<T> {
+        &self.inner
+    }
+}
+
+impl<T> AsMut<JoinSet<T>> for JoinSetStream<T> {
+    fn as_mut(&mut self) -> &mut JoinSet<T> {
+        &mut self.inner
+    }
+}
+
+impl<T> From<JoinSet<T>> for JoinSetStream<T> {
+    fn from(join_set: JoinSet<T>) -> Self {
+        Self::new(join_set)
+    }
+}
@@ -0,0 +1,39 @@
+#![cfg(feature = "rt")]
+
+use futures::{Stream, StreamExt};
+use std::collections::HashSet;
+use tokio::task::JoinSet;
+use tokio_stream::wrappers::JoinSetStream;
+
+#[tokio::test]
+async fn size_hint_stream() {
+    let set: JoinSet<_> = (0..2).map(|i| async move { i }).collect();
+    let mut stream = JoinSetStream::new(set);
+
+    assert_eq!(stream.size_hint(), (2, Some(2)));
+    stream.next().await;
+    assert_eq!(stream.size_hint(), (1, Some(1)));
+    stream.next().await;
+    assert_eq!(stream.size_hint(), (0, Some(0)));
+}
+
+#[tokio::test]
+async fn join_set_as_stream() {
+    let set: JoinSet<_> = (0..2).map(|i| async move { i }).collect();
+    let stream = JoinSetStream::new(set);
+
+    let values: HashSet<_> = stream.map(|result| result.unwrap()).collect().await;
+    assert_eq!(values, HashSet::from([0, 1]));
+}
+
+// Cannot run this test when “unwind” is disabled
+// since `JoinSet` use it to catch futures that panics.
+#[cfg(panic = "unwind")]
+#[tokio::test]
+async fn join_set_as_stream_panics_with_error() {
+    let set: JoinSet<_> = std::iter::once(async move { panic!("boom!") }).collect();
+    let mut stream = JoinSetStream::new(set);
+
+    let result = stream.next().await.transpose();
+    assert!(matches!(result, Err(e) if e.is_panic()));
+}
@@ -0,0 +1,152 @@
+use std::pin::Pin;
+
+use tokio_stream::{self as stream, Iter, StreamExt};
+
+fn base() -> Iter<std::vec::IntoIter<i32>> {
+    stream::iter(vec![1, 2, 3])
+}
+
+#[tokio::test]
+async fn map_accessors() {
+    let mut map = base().map(|x| x * 2);
+
+    let _: &Iter<_> = map.get_ref();
+    let _: &mut Iter<_> = map.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut map).get_pin_mut();
+
+    assert_eq!(map.next().await, Some(2));
+
+    // The recovered inner stream continues from where the combinator left it.
+    let mut inner = map.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+    assert_eq!(inner.next().await, Some(3));
+    assert_eq!(inner.next().await, None);
+}
+
+#[tokio::test]
+async fn take_accessors() {
+    let mut take = base().take(2);
+
+    let _: &Iter<_> = take.get_ref();
+    let _: &mut Iter<_> = take.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut take).get_pin_mut();
+
+    assert_eq!(take.next().await, Some(1));
+
+    // `take(2)` would stop after item 2, but the inner stream still has
+    // everything that was not yet pulled.
+    let mut inner = take.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+    assert_eq!(inner.next().await, Some(3));
+}
+
+#[tokio::test]
+async fn skip_accessors() {
+    let mut skip = base().skip(1);
+
+    let _: &Iter<_> = skip.get_ref();
+    let _: &mut Iter<_> = skip.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut skip).get_pin_mut();
+
+    assert_eq!(skip.next().await, Some(2));
+
+    let mut inner = skip.into_inner();
+    assert_eq!(inner.next().await, Some(3));
+}
+
+#[tokio::test]
+async fn filter_accessors() {
+    let mut filter = base().filter(|&x| x % 2 == 1);
+
+    let _: &Iter<_> = filter.get_ref();
+    let _: &mut Iter<_> = filter.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut filter).get_pin_mut();
+
+    assert_eq!(filter.next().await, Some(1));
+
+    let mut inner = filter.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn filter_map_accessors() {
+    let mut filter_map = base().filter_map(|x| (x % 2 == 1).then_some(x * 10));
+
+    let _: &Iter<_> = filter_map.get_ref();
+    let _: &mut Iter<_> = filter_map.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut filter_map).get_pin_mut();
+
+    assert_eq!(filter_map.next().await, Some(10));
+
+    let mut inner = filter_map.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn map_while_accessors() {
+    let mut map_while = base().map_while(|x| (x < 3).then_some(x + 100));
+
+    let _: &Iter<_> = map_while.get_ref();
+    let _: &mut Iter<_> = map_while.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut map_while).get_pin_mut();
+
+    assert_eq!(map_while.next().await, Some(101));
+
+    let mut inner = map_while.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn take_while_accessors() {
+    let mut take_while = base().take_while(|&x| x < 3);
+
+    let _: &Iter<_> = take_while.get_ref();
+    let _: &mut Iter<_> = take_while.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut take_while).get_pin_mut();
+
+    assert_eq!(take_while.next().await, Some(1));
+
+    let mut inner = take_while.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn skip_while_accessors() {
+    let mut skip_while = base().skip_while(|&x| x < 2);
+
+    let _: &Iter<_> = skip_while.get_ref();
+    let _: &mut Iter<_> = skip_while.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut skip_while).get_pin_mut();
+
+    assert_eq!(skip_while.next().await, Some(2));
+
+    let mut inner = skip_while.into_inner();
+    assert_eq!(inner.next().await, Some(3));
+}
+
+#[tokio::test]
+async fn then_accessors() {
+    let mut then = base().then(|x| std::future::ready(x + 1));
+
+    let _: &Iter<_> = then.get_ref();
+    let _: &mut Iter<_> = then.get_mut();
+    let _: Pin<&mut Iter<_>> = Pin::new(&mut then).get_pin_mut();
+
+    assert_eq!(then.next().await, Some(2));
+
+    let mut inner = then.into_inner();
+    assert_eq!(inner.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn get_mut_can_mutate_inner() {
+    // Mutating through `get_mut` is observed by the combinator.
+    let mut map = base().map(|x| x * 2);
+
+    // Skip one item directly on the inner stream.
+    let inner = map.get_mut();
+    assert_eq!(inner.next().await, Some(1));
+
+    // The combinator now sees the stream from item 2 onwards.
+    assert_eq!(map.next().await, Some(4));
+}
@@ -129,6 +129,26 @@ async fn take_while_terminated_after_predicate_fails() {
    assert!(stream.is_terminated());
 }

+// ── map_while ─────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn map_while_not_terminated_before_closure_returns_none() {
+    let stream =
+        tokio_stream::iter(vec![1, 2, 3]).map_while(|x| if x < 10 { Some(x) } else { None });
+    assert!(!stream.is_terminated());
+}
+
+#[tokio::test]
+async fn map_while_terminated_after_closure_returns_none() {
+    let mut stream =
+        tokio_stream::iter(vec![1, 5, 2]).map_while(|x| if x < 3 { Some(x) } else { None });
+    assert_eq!(stream.next().await, Some(1));
+    assert!(!stream.is_terminated());
+    // closure returns `None` on 5 → done flag set
+    assert_eq!(stream.next().await, None);
+    assert!(stream.is_terminated());
+}
+
 // ── then ─────────────────────────────────────────────────────────────────────

 #[tokio::test]
@@ -1,7 +1,6 @@
-use tokio_stream as stream;
-use tokio_test::task;
-
 use std::iter;
+use tokio_stream::{self as stream, Stream};
+use tokio_test::{assert_pending, assert_ready, task};

 #[tokio::test]
 async fn coop() {
@@ -9,6 +8,7 @@ async fn coop() {

    for _ in 0..10_000 {
        if stream.poll_next().is_pending() {
+            tokio::task::yield_now().await;
            assert!(stream.is_woken());
            return;
        }
@@ -16,3 +16,37 @@ async fn coop() {

    panic!("did not yield");
 }
+
+#[tokio::test]
+async fn test_iter_coop_budget() {
+    let mut stream = task::spawn(stream::iter(iter::repeat(1)));
+
+    // Tokio's default budget is 128.
+    // Fallback yield_amt is 32.
+    let limit = if cfg!(feature = "rt") { 128 } else { 32 };
+
+    for i in 0..limit {
+        let res = stream.poll_next();
+        assert!(res.is_ready(), "Should be ready at index {i}");
+    }
+
+    // Next poll should be pending
+    assert_pending!(stream.poll_next());
+
+    tokio::task::yield_now().await;
+    assert!(stream.is_woken());
+}
+
+#[tokio::test]
+async fn test_iter_size_hint() {
+    let stream = stream::iter(vec![1, 2, 3]);
+    assert_eq!(stream.size_hint(), (3, Some(3)));
+}
+
+#[tokio::test]
+async fn test_iter_eof_behavior() {
+    let mut stream = task::spawn(stream::iter(vec![1]));
+
+    assert_ready!(stream.poll_next());
+    assert_ready!(stream.poll_next()); // EOF should be ready None
+}
@@ -0,0 +1,22 @@
+use tokio_stream::StreamExt;
+
+#[tokio::test]
+async fn map_while_yields_until_closure_returns_none() {
+    let mut stream =
+        tokio_stream::iter(1..=10).map_while(|x| if x < 4 { Some(x + 3) } else { None });
+    assert_eq!(stream.next().await, Some(4));
+    assert_eq!(stream.next().await, Some(5));
+    assert_eq!(stream.next().await, Some(6));
+    assert_eq!(stream.next().await, None);
+}
+
+#[tokio::test]
+async fn map_while_does_not_poll_after_closure_returns_none() {
+    // Once the closure returns `None`, the underlying stream must not be polled
+    // again, so the trailing `2` is never yielded.
+    let mut stream =
+        tokio_stream::iter(vec![1, 5, 2]).map_while(|x| if x < 3 { Some(x) } else { None });
+    assert_eq!(stream.next().await, Some(1));
+    assert_eq!(stream.next().await, None);
+    assert_eq!(stream.next().await, None);
+}
@@ -0,0 +1,217 @@
+use tokio_stream::{self as stream, Stream, StreamExt};
+use tokio_test::{assert_pending, assert_ready, task};
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[tokio::test]
+async fn size_hint_without_peek() {
+    let mut s = stream::iter(vec![1, 2, 3]).peekable();
+    assert_eq!(s.size_hint(), (3, Some(3)));
+    s.next().await;
+    assert_eq!(s.size_hint(), (2, Some(2)));
+    s.next().await;
+    assert_eq!(s.size_hint(), (1, Some(1)));
+    s.next().await;
+    assert_eq!(s.size_hint(), (0, Some(0)));
+}
+
+#[tokio::test]
+async fn size_hint_with_peek() {
+    let mut s = stream::iter(vec![1, 2, 3]).peekable();
+
+    // before peek: all items are in inner stream
+    assert_eq!(s.size_hint(), (3, Some(3)));
+
+    // after peek: one item moves into self.peek buffer — total must still be 3
+    let _ = s.peek().await;
+    assert_eq!(s.size_hint(), (3, Some(3)));
+
+    // consume the peeked item via next()
+    assert_eq!(s.next().await, Some(1));
+    assert_eq!(s.size_hint(), (2, Some(2)));
+
+    // peek again
+    let _ = s.peek().await;
+    assert_eq!(s.size_hint(), (2, Some(2)));
+
+    s.next().await;
+    assert_eq!(s.size_hint(), (1, Some(1)));
+
+    // peek the last item
+    let _ = s.peek().await;
+    assert_eq!(s.size_hint(), (1, Some(1)));
+
+    s.next().await;
+    assert_eq!(s.size_hint(), (0, Some(0)));
+}
+
+#[tokio::test]
+async fn size_hint_empty_stream() {
+    let mut s = stream::iter(Vec::<i32>::new()).peekable();
+    assert_eq!(s.size_hint(), (0, Some(0)));
+    assert_eq!(s.peek().await, None);
+    assert_eq!(s.size_hint(), (0, Some(0)));
+}
+
+#[tokio::test]
+async fn peek_returns_correct_item() {
+    let mut s = stream::iter(vec![10, 20, 30]).peekable();
+
+    assert_eq!(s.peek().await, Some(&10));
+    assert_eq!(s.peek().await, Some(&10)); // second peek returns same item
+    assert_eq!(s.next().await, Some(10)); // next() gives same item
+    assert_eq!(s.next().await, Some(20));
+    assert_eq!(s.next().await, Some(30));
+    assert_eq!(s.next().await, None);
+}
+
+#[tokio::test]
+async fn size_hint_overflow() {
+    // When inner stream reports usize::MAX upper bound and an item is peeked,
+    // checked_add must return None rather than wrapping.
+    struct MaxHint(bool); // bool = whether to return one item
+
+    impl Stream for MaxHint {
+        type Item = ();
+        fn poll_next(
+            mut self: std::pin::Pin<&mut Self>,
+            _cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Option<()>> {
+            if self.0 {
+                self.0 = false;
+                std::task::Poll::Ready(Some(()))
+            } else {
+                std::task::Poll::Ready(None)
+            }
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            (usize::MAX, Some(usize::MAX))
+        }
+    }
+
+    let mut s = MaxHint(true).peekable();
+
+    // before peek: delegates directly to inner
+    assert_eq!(s.size_hint(), (usize::MAX, Some(usize::MAX)));
+
+    // after peek: peek_len=1, checked_add(1) on usize::MAX must give None not panic
+    let _ = s.peek().await;
+    assert_eq!(s.size_hint(), (usize::MAX, None));
+}
+
+#[tokio::test]
+async fn size_hint_unbounded_upper() {
+    // A stream that reports unknown upper bound
+    struct Unbounded;
+
+    impl Stream for Unbounded {
+        type Item = u32;
+        fn poll_next(
+            self: std::pin::Pin<&mut Self>,
+            _cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Option<u32>> {
+            std::task::Poll::Ready(Some(42))
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            (1, None)
+        }
+    }
+
+    let mut s = Unbounded.peekable();
+    assert_eq!(s.size_hint(), (1, None));
+    let _ = s.peek().await;
+    assert_eq!(s.size_hint(), (2, None)); // still unbounded after peek
+}
+
+#[tokio::test]
+async fn peek_does_not_consume() {
+    let mut stream = stream::iter(vec![1, 2, 3]).peekable();
+
+    assert_eq!(stream.peek().await, Some(&1));
+    assert_eq!(stream.peek().await, Some(&1));
+    assert_eq!(stream.next().await, Some(1));
+    assert_eq!(stream.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn peek_mut_mutates_yielded_item() {
+    let mut stream = stream::iter(vec![1, 2, 3]).peekable();
+
+    if let Some(item) = stream.peek_mut().await {
+        *item += 10;
+    }
+
+    assert_eq!(stream.next().await, Some(11));
+    assert_eq!(stream.next().await, Some(2));
+}
+
+#[tokio::test]
+async fn peek_on_empty_stream() {
+    let mut stream = stream::iter(Vec::<i32>::new()).peekable();
+
+    assert_eq!(stream.peek().await, None);
+    assert_eq!(stream.peek_mut().await, None);
+    assert_eq!(stream.next().await, None);
+}
+
+#[test]
+fn poll_peek_does_not_advance_stream() {
+    let mut stream = task::spawn(stream::iter(vec![1, 2, 3]).peekable());
+
+    let first = stream.enter(|cx, s| s.poll_peek(cx).map(|opt| opt.copied()));
+    assert_eq!(assert_ready!(first), Some(1));
+
+    let second = stream.enter(|cx, s| s.poll_peek(cx).map(|opt| opt.copied()));
+    assert_eq!(assert_ready!(second), Some(1));
+
+    let next = stream.enter(|cx, s| s.poll_next(cx));
+    assert_eq!(assert_ready!(next), Some(1));
+
+    let next = stream.enter(|cx, s| s.poll_next(cx));
+    assert_eq!(assert_ready!(next), Some(2));
+}
+
+#[test]
+fn poll_peek_mutates_buffered_item() {
+    let mut stream = task::spawn(stream::iter(vec![1, 2, 3]).peekable());
+
+    stream.enter(|cx, s| {
+        if let Poll::Ready(Some(item)) = s.poll_peek(cx) {
+            *item += 100;
+        }
+    });
+
+    let next = stream.enter(|cx, s| s.poll_next(cx));
+    assert_eq!(assert_ready!(next), Some(101));
+}
+
+struct PendingOnce {
+    polled: bool,
+}
+
+impl Stream for PendingOnce {
+    type Item = i32;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<i32>> {
+        if self.polled {
+            Poll::Ready(Some(7))
+        } else {
+            self.polled = true;
+            cx.waker().wake_by_ref();
+            Poll::Pending
+        }
+    }
+}
+
+#[test]
+fn poll_peek_propagates_pending() {
+    let mut stream = task::spawn(PendingOnce { polled: false }.peekable());
+
+    assert_pending!(stream.enter(|cx, s| s.poll_peek(cx).map(|opt| opt.copied())));
+
+    let ready = stream.enter(|cx, s| s.poll_peek(cx).map(|opt| opt.copied()));
+    assert_eq!(assert_ready!(ready), Some(7));
+}
@@ -225,6 +225,30 @@ fn size_hint_without_upper() {
    assert_eq!(size_hint, (3, None));
 }

+#[test]
+fn size_hint_overflow() {
+    struct Monster;
+
+    impl Stream for Monster {
+        type Item = ();
+
+        fn poll_next(self: Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> Poll<Option<()>> {
+            panic!()
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            (usize::MAX, Some(usize::MAX))
+        }
+    }
+
+    let mut map = StreamMap::new();
+
+    map.insert("a", Monster);
+    map.insert("b", Monster);
+
+    assert_eq!(map.size_hint(), (usize::MAX, None));
+}
+
 #[test]
 fn new_capacity_zero() {
    let map = StreamMap::<&str, stream::Pending<()>>::new();
@@ -405,6 +429,27 @@ async fn poll_next_many_enough() {
    assert!(buffer.contains(&(1, 1)));
 }

+#[tokio::test]
+async fn poll_next_many_does_not_exceed_limit() {
+    let mut stream_map: StreamMap<usize, UsizeStream> = StreamMap::new();
+
+    stream_map.insert(0, Box::pin(iter([0usize].into_iter())) as UsizeStream);
+    stream_map.insert(1, Box::pin(iter([1usize].into_iter())) as UsizeStream);
+
+    let mut buffer = vec![];
+    let n = poll_fn(|cx| stream_map.poll_next_many(cx, &mut buffer, 1)).await;
+
+    assert_eq!(n, 1);
+    assert_eq!(buffer.len(), 1);
+
+    let n = poll_fn(|cx| stream_map.poll_next_many(cx, &mut buffer, 1)).await;
+
+    assert_eq!(n, 1);
+    assert_eq!(buffer.len(), 2);
+    assert!(buffer.contains(&(0, 0)));
+    assert!(buffer.contains(&(1, 1)));
+}
+
 #[tokio::test]
 async fn poll_next_many_correctly_loops_around() {
    for _ in 0..10 {
@@ -519,6 +564,27 @@ async fn next_many_enough() {
    assert!(buffer.contains(&(1, 1)));
 }

+#[tokio::test]
+async fn next_many_does_not_exceed_limit() {
+    let mut stream_map: StreamMap<usize, UsizeStream> = StreamMap::new();
+
+    stream_map.insert(0, Box::pin(iter([0usize].into_iter())) as UsizeStream);
+    stream_map.insert(1, Box::pin(iter([1usize].into_iter())) as UsizeStream);
+
+    let mut buffer = vec![];
+    let n = poll_fn(|cx| pin!(stream_map.next_many(&mut buffer, 1)).poll(cx)).await;
+
+    assert_eq!(n, 1);
+    assert_eq!(buffer.len(), 1);
+
+    let n = poll_fn(|cx| pin!(stream_map.next_many(&mut buffer, 1)).poll(cx)).await;
+
+    assert_eq!(n, 1);
+    assert_eq!(buffer.len(), 2);
+    assert!(buffer.contains(&(0, 0)));
+    assert!(buffer.contains(&(1, 1)));
+}
+
 #[tokio::test]
 async fn next_many_correctly_loops_around() {
    for _ in 0..10 {
@@ -198,10 +198,7 @@ macro_rules! assert_ready_eq {
 /// ```
 #[macro_export]
 macro_rules! assert_ok {
-    ($e:expr) => {
-        assert_ok!($e,)
-    };
-    ($e:expr,) => {{
+    ($e:expr $(,)?) => {{
        use std::result::Result::*;
        match $e {
            Ok(v) => v,
@@ -241,10 +238,7 @@ macro_rules! assert_ok {
 /// ```
 #[macro_export]
 macro_rules! assert_err {
-    ($e:expr) => {
-        assert_err!($e,);
-    };
-    ($e:expr,) => {{
+    ($e:expr $(,)?) => {{
        use std::result::Result::*;
        match $e {
            Ok(v) => panic!("assertion failed: Ok({:?})", v),
@@ -70,6 +70,9 @@ const IDLE: usize = 0;
 const WAKE: usize = 1;
 const SLEEP: usize = 2;

+/// Default maximum number of poll iterations in [`Spawn::poll_until_idle`].
+const POLL_UNTIL_IDLE_MAX_ITERATIONS: usize = 150;
+
 impl<T> Spawn<T> {
    /// Consumes `self` returning the inner value
    pub fn into_inner(self) -> T
@@ -123,6 +126,43 @@ impl<T: Future> Spawn<T> {
        let fut = self.future.as_mut();
        self.task.enter(|cx| fut.poll(cx))
    }
+
+    /// Polls the future until it is idle.
+    ///
+    /// A future is considered idle when it either completes, or returns
+    /// [`Poll::Pending`] without a pending wake notification.
+    ///
+    /// Unlike [`poll`](Self::poll), this method keeps polling while the future
+    /// returns [`Poll::Pending`] but has received a wake notification, advancing
+    /// the future as far as possible without waiting for external events.
+    ///
+    /// Polling is bounded to avoid infinite loops when a future wakes without
+    /// making progress.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the iteration limit is exceeded.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use tokio_test::task;
+    ///
+    /// let mut task = task::spawn(async { 42 });
+    ///
+    /// assert!(task.poll_until_idle().is_ready());
+    /// ```
+    pub fn poll_until_idle(&mut self) -> Poll<T::Output> {
+        for _ in 0..POLL_UNTIL_IDLE_MAX_ITERATIONS {
+            let result = self.poll();
+            if result.is_ready() || !self.is_woken() {
+                return result;
+            }
+        }
+        panic!(
+            "poll_until_idle exceeded {POLL_UNTIL_IDLE_MAX_ITERATIONS} iterations; future may be waking without making progress"
+        );
+    }
 }

 impl<T: Stream> Spawn<T> {
@@ -1,3 +1,4 @@
+use std::future::{pending, Future};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio_stream::Stream;
@@ -23,3 +24,81 @@ fn test_spawn_stream_size_hint() {
    let spawn = task::spawn(SizedStream);
    assert_eq!(spawn.size_hint(), (100, Some(200)));
 }
+
+#[test]
+fn poll_until_idle_ready() {
+    let mut task = task::spawn(async { 42 });
+    assert_eq!(task.poll_until_idle(), Poll::Ready(42));
+}
+
+#[test]
+fn poll_until_idle_pending_not_woken() {
+    let mut task = task::spawn(pending::<()>());
+    assert!(task.poll_until_idle().is_pending());
+}
+
+struct WakeThenReady {
+    step: u8,
+}
+
+impl Future for WakeThenReady {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        match self.step {
+            0 => {
+                self.step = 1;
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }
+            _ => Poll::Ready(()),
+        }
+    }
+}
+
+#[test]
+fn poll_until_idle_advances_on_wake() {
+    let mut task = task::spawn(WakeThenReady { step: 0 });
+    assert!(task.poll_until_idle().is_ready());
+}
+
+struct WakeNTimes {
+    remaining: u8,
+}
+
+impl Future for WakeNTimes {
+    type Output = u8;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u8> {
+        if self.remaining == 0 {
+            return Poll::Ready(0);
+        }
+        self.remaining -= 1;
+        cx.waker().wake_by_ref();
+        Poll::Pending
+    }
+}
+
+#[test]
+fn poll_until_idle_multiple_wakes() {
+    let mut task = task::spawn(WakeNTimes { remaining: 3 });
+    assert_eq!(task.poll_until_idle(), Poll::Ready(0));
+}
+
+struct WakeForever;
+
+impl Future for WakeForever {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        cx.waker().wake_by_ref();
+        Poll::Pending
+    }
+}
+
+#[test]
+#[should_panic(expected = "poll_until_idle exceeded 150 iterations")]
+fn poll_until_idle_panics_on_infinite_wake() {
+    let mut task = task::spawn(WakeForever);
+    let _ = task.poll_until_idle();
+}
@@ -1,3 +1,39 @@
+# 0.7.19 (July 21st, 2026)
+
+### Added
+
+- io: add `write_all_vectored` ([#7768], [#8159])
+- sync: add `DropGuard::token` for cancellation tokens ([#8226])
+- sync: implement `PartialEq` and `Eq` for `CancellationToken` ([#8110])
+- task: add `AbortOnDrop` ([#7855])
+- task: add `JoinMap::try_join_next` ([#8099])
+
+### Changed
+
+- codec: use `libc::memchr` for `LinesCodec` delimiter scan ([#8141])
+
+### Fixed
+
+- codec: fix `is_readable` when buffer is not empty ([#7912])
+- task: avoid replacing the `JoinQueue` waker in `try_join_next` ([#8279])
+- time: wake `DelayQueue` after resetting to expired ([#8274])
+
+### Documented
+
+- codec: document `UdpFramed` decoder errors ([#8248])
+
+[#7768]: https://github.com/tokio-rs/tokio/pull/7768
+[#7855]: https://github.com/tokio-rs/tokio/pull/7855
+[#7912]: https://github.com/tokio-rs/tokio/pull/7912
+[#8099]: https://github.com/tokio-rs/tokio/pull/8099
+[#8110]: https://github.com/tokio-rs/tokio/pull/8110
+[#8141]: https://github.com/tokio-rs/tokio/pull/8141
+[#8159]: https://github.com/tokio-rs/tokio/pull/8159
+[#8226]: https://github.com/tokio-rs/tokio/pull/8226
+[#8248]: https://github.com/tokio-rs/tokio/pull/8248
+[#8274]: https://github.com/tokio-rs/tokio/pull/8274
+[#8279]: https://github.com/tokio-rs/tokio/pull/8279
+
 # 0.7.18 (January 4th, 2026)

 ### Added
@@ -4,7 +4,7 @@ name = "tokio-util"
 # - Remove path dependencies (if any)
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.18"
+version = "0.7.19"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -25,7 +25,7 @@ full = ["codec", "compat", "io-util", "time", "net", "rt", "join-map"]

 net = ["tokio/net"]
 compat = ["futures-io"]
-codec = []
+codec = ["libc"]
 time = ["tokio/time", "slab"]
 io = []
 io-util = ["io", "tokio/rt", "tokio/io-util"]
@@ -46,6 +46,9 @@ slab = { version = "0.4.4", optional = true } # Backs `DelayQueue`
 tracing = { version = "0.1.29", default-features = false, features = ["std"], optional = true }
 hashbrown = { version = "0.15.0", default-features = false, optional = true }

+[target.'cfg(unix)'.dependencies]
+libc = { version = "0.2.168", optional = true } # Backs the LinesCodec delimiter scan via libc::memchr
+
 [dev-dependencies]
 tokio = { version = "1.0.0", features = ["full"] }
 tokio-test = "0.4.0"
@@ -47,7 +47,7 @@ pub trait Decoder {
    /// implementation of `decode_eof` to yield an `io::Error` when the decoder
    /// fails to consume all available data.
    ///
-    /// Note that implementors of this trait can simply indicate `type Error =
+    /// Note that implementers of this trait can simply indicate `type Error =
    /// io::Error` to use I/O errors as this type.
    ///
    /// [`FramedRead`]: crate::codec::FramedRead
@@ -493,8 +493,12 @@ impl LengthDelimitedCodec {
    /// words, if a frame is currently in process of being decoded with a frame
    /// size greater than `val` but less than the max frame length in effect
    /// before calling this function, then the frame will be allowed.
+    ///
+    /// If `val` is larger than what the length field can represent, it is
+    /// clipped to the maximum representable value.
    pub fn set_max_frame_length(&mut self, val: usize) {
        self.builder.max_frame_length(val);
+        self.builder.adjust_max_frame_len();
    }

    fn decode_head(&mut self, src: &mut BytesMut) -> io::Result<Option<usize>> {
@@ -599,10 +603,10 @@ impl Decoder for LengthDelimitedCodec {
    }
 }

-impl Encoder<Bytes> for LengthDelimitedCodec {
+impl Encoder<&[u8]> for LengthDelimitedCodec {
    type Error = io::Error;

-    fn encode(&mut self, data: Bytes, dst: &mut BytesMut) -> Result<(), io::Error> {
+    fn encode(&mut self, data: &[u8], dst: &mut BytesMut) -> Result<(), io::Error> {
        let n = data.len();

        if n > self.builder.max_frame_len {
@@ -627,8 +631,8 @@ impl Encoder<Bytes> for LengthDelimitedCodec {
        })?;

        // Reserve capacity in the destination buffer to fit the frame and
-        // length field (plus adjustment).
-        dst.reserve(self.builder.length_field_len + n);
+        // length field.
+        dst.reserve(self.builder.length_field_len + data.len());

        if self.builder.length_field_is_big_endian {
            dst.put_uint(n as u64, self.builder.length_field_len);
@@ -637,12 +641,20 @@ impl Encoder<Bytes> for LengthDelimitedCodec {
        }

        // Write the frame to the buffer
-        dst.extend_from_slice(&data[..]);
+        dst.extend_from_slice(data);

        Ok(())
    }
 }

+impl Encoder<Bytes> for LengthDelimitedCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, data: Bytes, dst: &mut BytesMut) -> Result<(), io::Error> {
+        Encoder::<&[u8]>::encode(self, data.as_ref(), dst)
+    }
+}
+
 impl Default for LengthDelimitedCodec {
    fn default() -> Self {
        Self::new()
@@ -675,18 +687,21 @@ impl Builder {
    /// # Examples
    ///
    /// ```
-    /// # use tokio::io::AsyncRead;
    /// use tokio_util::codec::LengthDelimitedCodec;
+    /// use tokio_stream::StreamExt;
    ///
-    /// # fn bind_read<T: AsyncRead>(io: T) {
-    /// LengthDelimitedCodec::builder()
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() {
+    /// let io: &[u8] = b"\x00\x0bhello world";
+    /// let mut reader = LengthDelimitedCodec::builder()
    ///     .length_field_offset(0)
    ///     .length_field_type::<u16>()
    ///     .length_adjustment(0)
-    ///     .num_skip(0)
    ///     .new_read(io);
+    ///
+    /// let frame = reader.next().await.unwrap().unwrap();
+    /// assert_eq!(&frame[..], b"hello world");
    /// # }
-    /// # pub fn main() {}
    /// ```
    pub fn new() -> Builder {
        Builder {
@@ -949,14 +964,20 @@ impl Builder {
    /// # Examples
    ///
    /// ```
-    /// use tokio_util::codec::LengthDelimitedCodec;
+    /// use bytes::{Bytes, BytesMut};
+    /// use tokio_util::codec::{Decoder, Encoder, LengthDelimitedCodec};
+    ///
    /// # pub fn main() {
-    /// LengthDelimitedCodec::builder()
+    /// let mut codec = LengthDelimitedCodec::builder()
    ///     .length_field_offset(0)
    ///     .length_field_type::<u16>()
    ///     .length_adjustment(0)
-    ///     .num_skip(0)
    ///     .new_codec();
+    ///
+    /// let mut buf = BytesMut::new();
+    /// codec.encode(Bytes::from_static(b"hello world"), &mut buf).unwrap();
+    /// let frame = codec.decode(&mut buf).unwrap().unwrap();
+    /// assert_eq!(&frame[..], b"hello world");
    /// # }
    /// ```
    pub fn new_codec(&self) -> LengthDelimitedCodec {
@@ -975,18 +996,21 @@ impl Builder {
    /// # Examples
    ///
    /// ```
-    /// # use tokio::io::AsyncRead;
    /// use tokio_util::codec::LengthDelimitedCodec;
+    /// use tokio_stream::StreamExt;
    ///
-    /// # fn bind_read<T: AsyncRead>(io: T) {
-    /// LengthDelimitedCodec::builder()
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() {
+    /// let io: &[u8] = b"\x00\x0bhello world";
+    /// let mut reader = LengthDelimitedCodec::builder()
    ///     .length_field_offset(0)
    ///     .length_field_type::<u16>()
    ///     .length_adjustment(0)
-    ///     .num_skip(0)
    ///     .new_read(io);
+    ///
+    /// let frame = reader.next().await.unwrap().unwrap();
+    /// assert_eq!(&frame[..], b"hello world");
    /// # }
-    /// # pub fn main() {}
    /// ```
    pub fn new_read<T>(&self, upstream: T) -> FramedRead<T, LengthDelimitedCodec>
    where
@@ -1040,7 +1064,7 @@ impl Builder {

    fn num_head_bytes(&self) -> usize {
        let num = self.length_field_offset + self.length_field_len;
-        cmp::max(num, self.num_skip.unwrap_or(0))
+        cmp::max(num, self.num_skip.unwrap_or_default())
    }

    fn get_num_skip(&self) -> usize {
@@ -1049,16 +1073,25 @@ impl Builder {
    }

    fn adjust_max_frame_len(&mut self) {
-        // Calculate the maximum number that can be represented using `length_field_len` bytes.
-        let max_number = match 1u64.checked_shl((8 * self.length_field_len) as u32) {
+        let max_allowed_len = self.max_allowed_frame_len();
+
+        if self.max_frame_len > max_allowed_len {
+            self.max_frame_len = max_allowed_len;
+        }
+    }
+
+    fn max_allowed_frame_len(&self) -> usize {
+        let max_allowed_len = self
+            .max_length_field_value()
+            .saturating_add_signed(self.length_adjustment as i64);
+
+        usize::try_from(max_allowed_len).unwrap_or(usize::MAX)
+    }
+
+    fn max_length_field_value(&self) -> u64 {
+        match 1u64.checked_shl((8 * self.length_field_len) as u32) {
            Some(shl) => shl - 1,
            None => u64::MAX,
-        };
-
-        let max_allowed_len = max_number.saturating_add_signed(self.length_adjustment as i64);
-
-        if self.max_frame_len as u64 > max_allowed_len {
-            self.max_frame_len = usize::try_from(max_allowed_len).unwrap_or(usize::MAX);
        }
    }
 }
@@ -115,9 +115,7 @@ impl Decoder for LinesCodec {
            // there's no max_length set, we'll read to the end of the buffer.
            let read_to = cmp::min(self.max_length.saturating_add(1), buf.len());

-            let newline_offset = buf[self.next_index..read_to]
-                .iter()
-                .position(|b| *b == b'\n');
+            let newline_offset = crate::util::memchr::memchr(b'\n', &buf[self.next_index..read_to]);

            match (self.is_discarding, newline_offset) {
                (true, Some(offset)) => {
@@ -12,7 +12,7 @@
 //! The following example demonstrates how to use a codec such as [`LinesCodec`] to
 //! write framed data. [`FramedWrite`] can be used to achieve this. Data sent to
 //! [`FramedWrite`] are first framed according to a specific codec, and then sent to
-//! an implementor of [`AsyncWrite`].
+//! an implementer of [`AsyncWrite`].
 //!
 //! ```
 //! use futures::sink::SinkExt;
@@ -43,7 +43,7 @@
 //! # Example decoding using `LinesCodec`
 //! The following example demonstrates how to use a codec such as [`LinesCodec`] to
 //! read a stream of framed data. [`FramedRead`] can be used to achieve this. [`FramedRead`]
-//! will keep reading from an [`AsyncRead`] implementor until a whole frame, according to a codec,
+//! will keep reading from an [`AsyncRead`] implementer until a whole frame, according to a codec,
 //! can be parsed.
 //!
 //!```
@@ -291,6 +291,10 @@ impl AsyncWrite for Sender {
            return Poll::Ready(Err(IoError::new(IoErrorKind::BrokenPipe, CLOSED_ERROR_MSG)));
        }

+        if bufs.iter().all(|buf| buf.is_empty()) {
+            return Poll::Ready(Ok(0));
+        }
+
        let free = inner
            .backpressure_boundary
            .checked_sub(inner.buf.len())
@@ -158,6 +158,7 @@ pub struct StreamReader<S, B> {
    inner: S,
    // This field is not pinned.
    chunk: Option<B>,
+    eof: bool,
 }

 impl<S, B, E> StreamReader<S, B>
@@ -179,6 +180,7 @@ where
        Self {
            inner: stream,
            chunk: None,
+            eof: false,
        }
    }

@@ -277,6 +279,8 @@ where
                // This unwrap is very sad, but it can't be avoided.
                let buf = self.project().chunk.as_ref().unwrap().chunk();
                return Poll::Ready(Ok(buf));
+            } else if *self.as_mut().project().eof {
+                return Poll::Ready(Ok(&[]));
            } else {
                match self.as_mut().project().inner.poll_next(cx) {
                    Poll::Ready(Some(Ok(chunk))) => {
@@ -284,7 +288,10 @@ where
                        *self.as_mut().project().chunk = Some(chunk);
                    }
                    Poll::Ready(Some(Err(err))) => return Poll::Ready(Err(err.into())),
-                    Poll::Ready(None) => return Poll::Ready(Ok(&[])),
+                    Poll::Ready(None) => {
+                        *self.as_mut().project().eof = true;
+                        return Poll::Ready(Ok(&[]));
+                    }
                    Poll::Pending => return Poll::Pending,
                }
            }
@@ -311,6 +318,7 @@ impl<S: Unpin, B> Unpin for StreamReader<S, B> {}
 struct StreamReaderProject<'a, S, B> {
    inner: Pin<&'a mut S>,
    chunk: &'a mut Option<B>,
+    eof: &'a mut bool,
 }

 impl<S, B> StreamReader<S, B> {
@@ -322,6 +330,7 @@ impl<S, B> StreamReader<S, B> {
        StreamReaderProject {
            inner: unsafe { Pin::new_unchecked(&mut me.inner) },
            chunk: &mut me.chunk,
+            eof: &mut me.eof,
        }
    }
 }
@@ -148,7 +148,7 @@ fn advance_slices<'a>(bufs: &mut &mut [IoSlice<'a>], n: usize) {
    *bufs = &mut std::mem::take(bufs)[remove..];
    if let Some(first) = bufs.first_mut() {
        let buf = &first[left..];
-        // necessary due to limitating in the borrow checker,
+        // Necessary due to a limitation in the borrow checker,
        // when tokio MSRV reaches 1.81.0 this entire function
        // can be replaced with `IoSlice::advance_slices`
        //
@@ -10,6 +10,13 @@ pub struct DropGuard {
 }

 impl DropGuard {
+    /// Returns a reference to the cancellation token wrapped by this guard.
+    pub fn token(&self) -> &CancellationToken {
+        self.inner
+            .as_ref()
+            .expect("`inner` can only be None in a destructor")
+    }
+
    /// Returns stored cancellation token and removes this drop guard instance
    /// (i.e. it will no longer cancel token). Other guards for this token
    /// are not affected.
@@ -13,6 +13,13 @@ pub struct DropGuardRef<'a> {
 }

 impl<'a> DropGuardRef<'a> {
+    /// Returns a reference to the cancellation token wrapped by this guard.
+    pub fn token(&self) -> &CancellationToken {
+        self.inner
+            .as_ref()
+            .expect("`inner` can only be None in a destructor")
+    }
+
    /// Returns stored cancellation token and removes this drop guard instance
    /// (i.e. it will no longer cancel token). Other guards for this token
    /// are not affected.
@@ -188,6 +188,10 @@ impl<T> JoinQueue<T> {
    /// Note that on success the handle will panic on subsequent polls
    /// since it becomes consumed.
    fn try_poll_handle(jh: &mut AbortOnDropHandle<T>) -> Option<Result<T, JoinError>> {
+        if !jh.is_finished() {
+            return None;
+        }
+
        let waker = futures_util::task::noop_waker();
        let mut cx = Context::from_waker(&waker);

@@ -4,19 +4,18 @@ use std::fmt::{Debug, Formatter};
 use std::future::Future;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
-use tokio::runtime::Builder;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 use tokio::sync::oneshot;
-use tokio::task::{spawn_local, JoinHandle, LocalSet};
+use tokio::task::{spawn_local, JoinHandle};

 /// A cloneable handle to a local pool, used for spawning `!Send` tasks.
 ///
-/// Internally the local pool uses a [`tokio::task::LocalSet`] for each worker thread
+/// Internally the local pool uses a [`tokio::runtime::LocalRuntime`] for each worker thread
 /// in the pool. Consequently you can also use [`tokio::task::spawn_local`] (which will
 /// execute on the same thread) inside the Future you supply to the various spawn methods
 /// of `LocalPoolHandle`.
 ///
-/// [`tokio::task::LocalSet`]: tokio::task::LocalSet
+/// [`tokio::runtime::LocalRuntime`]: tokio::runtime::LocalRuntime
 /// [`tokio::task::spawn_local`]: tokio::task::spawn_local
 ///
 /// # Examples
@@ -238,10 +237,10 @@ impl LocalPool {
            let _abort_guard = AbortGuard(abort_handle);

            // Inside the future we can't run spawn_local yet because we're not
-            // in the context of a LocalSet. We need to send create_task to the
-            // LocalSet task for spawning.
+            // in the context of a LocalRuntime. We need to send create_task to the
+            // LocalRuntime task for spawning.
            let spawn_task = Box::new(move || {
-                // Once we're in the LocalSet context we can call spawn_local
+                // Once we're in the LocalRuntime context we can call spawn_local
                let join_handle =
                    spawn_local(
                        async move { Abortable::new(create_task(), abort_registration).await },
@@ -255,7 +254,7 @@ impl LocalPool {
                }
            });

-            // Send the callback to the LocalSet task
+            // Send the callback to the LocalRuntime 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}");
@@ -379,15 +378,17 @@ impl LocalWorkerHandle {
    /// Create a new worker for executing pinned tasks
    fn new_worker() -> LocalWorkerHandle {
        let (sender, receiver) = unbounded_channel();
-        let runtime = Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("Failed to start a pinned worker thread runtime");
-        let runtime_handle = runtime.handle().clone();
+        let (handle_sender, handle_receiver) = std::sync::mpsc::channel();
+
        let task_count = Arc::new(AtomicUsize::new(0));
        let task_count_clone = Arc::clone(&task_count);

-        std::thread::spawn(|| Self::run(runtime, receiver, task_count_clone));
+        std::thread::spawn(|| Self::run(handle_sender, receiver, task_count_clone));
+
+        let runtime_handle = handle_receiver
+            .recv()
+            .expect("Failed to recv local runtime init result")
+            .expect("Failed to start local runtime");

        LocalWorkerHandle {
            runtime_handle,
@@ -397,28 +398,38 @@ impl LocalWorkerHandle {
    }

    fn run(
-        runtime: tokio::runtime::Runtime,
+        handle_sender: std::sync::mpsc::Sender<std::io::Result<tokio::runtime::Handle>>,
        mut task_receiver: UnboundedReceiver<PinnedFutureSpawner>,
        task_count: Arc<AtomicUsize>,
    ) {
-        let local_set = LocalSet::new();
-        local_set.block_on(&runtime, async {
+        let runtime = match tokio::runtime::LocalRuntime::new() {
+            Ok(runtime) => runtime,
+            Err(err) => {
+                let _ = handle_sender.send(Err(err));
+                return;
+            }
+        };
+
+        let runtime_handle = runtime.handle().clone();
+
+        handle_sender
+            .send(Ok(runtime_handle))
+            .expect("Failed to send local runtime handle");
+        drop(handle_sender);
+
+        runtime.block_on(async {
            while let Some(spawn_task) = task_receiver.recv().await {
                // Calls spawn_local(future)
                (spawn_task)();
            }
        });

-        // If there are any tasks on the runtime associated with a LocalSet task
-        // that has already completed, but whose output has not yet been
-        // reported, let that task complete.
+        // If there are any tasks on the runtime that has already completed,
+        // but whose output has not yet been reported, let that task complete.
        //
        // Since the task_count is decremented when the runtime task exits,
        // reading that counter lets us know if any such tasks completed during
        // the call to `block_on`.
-        //
-        // Tasks on the LocalSet can't complete during this loop since they're
-        // stored on the LocalSet and we aren't accessing it.
        let mut previous_task_count = task_count.load(Ordering::SeqCst);
        loop {
            // This call will also run tasks spawned on the runtime.
@@ -431,15 +442,10 @@ impl LocalWorkerHandle {
            }
        }

-        // It's now no longer possible for a task on the runtime to be
-        // associated with a LocalSet task that has completed. Drop both the
-        // LocalSet and runtime to let tasks on the runtime be cancelled if and
-        // only if they are still on the LocalSet.
-        //
-        // Drop the LocalSet task first so that anyone awaiting the runtime
-        // JoinHandle will see the cancelled error after the LocalSet task
-        // destructor has completed.
-        drop(local_set);
+        // It's now no longer possible for a task on the local runtime
+        // associated with task that has completed. Drop both
+        // local runtime to let tasks on the runtime be cancelled if and
+        // only if they are still on the runtime.
        drop(runtime);
    }
 }
@@ -580,12 +580,7 @@ impl<T> DelayQueue<T> {
    /// current task for wakeup if the value is not yet available, and returning
    /// `None` if the queue is exhausted.
    pub fn poll_expired(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Expired<T>>> {
-        if !self
-            .waker
-            .as_ref()
-            .map(|w| w.will_wake(cx.waker()))
-            .unwrap_or(false)
-        {
+        if !self.waker.as_ref().is_some_and(|w| w.will_wake(cx.waker())) {
            self.waker = Some(cx.waker().clone());
        }

@@ -864,11 +859,19 @@ impl<T> DelayQueue<T> {
        self.slab[*key].expired = false;

        self.insert_idx(when, *key);
+        let inserted_expired = self.slab[*key].expired;

        let next_deadline = self.next_deadline();
-        if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) {
-            // This should awaken us if necessary (ie, if already expired)
-            delay.as_mut().reset(deadline);
+        match (next_deadline, &mut self.delay) {
+            (None, _) => self.delay = None,
+            (Some(deadline), Some(delay)) => delay.as_mut().reset(deadline),
+            (Some(deadline), None) => self.delay = Some(Box::pin(sleep_until(deadline))),
+        }
+
+        if inserted_expired {
+            if let Some(waker) = self.waker.take() {
+                waker.wake();
+            }
        }
    }

@@ -1028,10 +1031,18 @@ impl<T> DelayQueue<T> {
    /// # }
    /// ```
    pub fn clear(&mut self) {
+        let had_entries = !self.slab.is_empty();
+
        self.slab.clear();
        self.expired = Stack::default();
        self.wheel = Wheel::new();
        self.delay = None;
+
+        if had_entries {
+            if let Some(waker) = self.waker.take() {
+                waker.wake();
+            }
+        }
    }

    /// Returns the number of elements the queue can hold without reallocating.
@@ -111,8 +111,7 @@ where
        debug_assert!({
            self.levels[level]
                .next_expiration(self.elapsed)
-                .map(|e| e.deadline >= self.elapsed)
-                .unwrap_or(true)
+                .map_or(true, |e| e.deadline >= self.elapsed)
        });

        Ok(())
@@ -22,6 +22,12 @@ use std::{
 /// handle encoding and decoding of messages frames. Note that the incoming and
 /// outgoing frame types may be distinct.
 ///
+/// A single datagram may decode into multiple frames. `UdpFramed` will keep
+/// calling [`Decoder::decode_eof`] with the current datagram until it returns
+/// `Ok(None)`. If a decoder wants to discard a malformed datagram and continue
+/// receiving later datagrams, it should consume or clear the remaining bytes
+/// from the buffer before returning `Err`.
+///
 /// This function returns a *single* object that is both [`Stream`] and [`Sink`];
 /// grouping this into a single object is often useful for layering things which
 /// require both read and write access to the underlying object.
@@ -0,0 +1,102 @@
+//! Search for a byte in a byte array using libc.
+//!
+//! When nothing pulls in libc, then just use a trivial implementation. Note
+//! that we only depend on libc on unix.
+
+#[cfg(not(all(unix, feature = "libc")))]
+fn memchr_inner(needle: u8, haystack: &[u8]) -> Option<usize> {
+    haystack.iter().position(|val| needle == *val)
+}
+
+#[cfg(all(unix, feature = "libc"))]
+fn memchr_inner(needle: u8, haystack: &[u8]) -> Option<usize> {
+    let start = haystack.as_ptr();
+
+    // SAFETY: `start` is valid for `haystack.len()` bytes.
+    let ptr = (unsafe { libc::memchr(start.cast(), needle as _, haystack.len()) })
+        .cast::<u8>()
+        .cast_const();
+
+    if ptr.is_null() {
+        None
+    } else {
+        // SAFETY: `ptr` will always be in bounds, since libc guarantees that the ptr will either
+        //          be to an element inside the array or the ptr will be null
+        //          since the ptr is in bounds the offset must also always be non null
+        //          and there can't be more than isize::MAX elements inside an array
+        //          as rust guarantees that the maximum number of bytes a allocation
+        //          may occupy is isize::MAX
+        unsafe {
+            // TODO(MSRV 1.87): When bumping MSRV, switch to `ptr.byte_offset_from_unsigned(start)`.
+            Some(usize::try_from(ptr.offset_from(start)).unwrap_unchecked())
+        }
+    }
+}
+
+pub(crate) fn memchr(needle: u8, haystack: &[u8]) -> Option<usize> {
+    let index = memchr_inner(needle, haystack)?;
+
+    // SAFETY: `memchr_inner` returns Some(index) and in that case index must point to an element in haystack
+    //         or `memchr_inner` None which is guarded by the `?` operator above
+    //         therefore the index must **always** point to an element in the array
+    //         and so this indexing operation is safe
+    // TODO(MSRV 1.81): When bumping MSRV, switch to `std::hint::assert_unchecked(haystack.get(..=index).is_some());`
+    unsafe {
+        if haystack.get(..=index).is_none() {
+            std::hint::unreachable_unchecked()
+        }
+    }
+
+    Some(index)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::memchr;
+
+    #[test]
+    fn memchr_test() {
+        let haystack = b"123abc456\0\xffabc\n";
+
+        assert_eq!(memchr(b'1', haystack), Some(0));
+        assert_eq!(memchr(b'2', haystack), Some(1));
+        assert_eq!(memchr(b'3', haystack), Some(2));
+        assert_eq!(memchr(b'4', haystack), Some(6));
+        assert_eq!(memchr(b'5', haystack), Some(7));
+        assert_eq!(memchr(b'6', haystack), Some(8));
+        assert_eq!(memchr(b'7', haystack), None);
+        assert_eq!(memchr(b'a', haystack), Some(3));
+        assert_eq!(memchr(b'b', haystack), Some(4));
+        assert_eq!(memchr(b'c', haystack), Some(5));
+        assert_eq!(memchr(b'd', haystack), None);
+        assert_eq!(memchr(b'A', haystack), None);
+        assert_eq!(memchr(0, haystack), Some(9));
+        assert_eq!(memchr(0xff, haystack), Some(10));
+        assert_eq!(memchr(0xfe, haystack), None);
+        assert_eq!(memchr(1, haystack), None);
+        assert_eq!(memchr(b'\n', haystack), Some(14));
+        assert_eq!(memchr(b'\r', haystack), None);
+    }
+
+    #[test]
+    fn memchr_all() {
+        let mut arr = Vec::new();
+        for b in 0..=255 {
+            arr.push(b);
+        }
+        for b in 0..=255 {
+            assert_eq!(memchr(b, &arr), Some(b as usize));
+        }
+        arr.reverse();
+        for b in 0..=255 {
+            assert_eq!(memchr(b, &arr), Some(255 - b as usize));
+        }
+    }
+
+    #[test]
+    fn memchr_empty() {
+        for b in 0..=255 {
+            assert_eq!(memchr(b, b""), None);
+        }
+    }
+}
@@ -1,4 +1,6 @@
 mod maybe_dangling;
+#[cfg(feature = "codec")]
+pub(crate) mod memchr;
 #[cfg(any(feature = "io", feature = "codec"))]
 mod poll_buf;

@@ -354,3 +354,16 @@ async fn poll_write_vectored_3() {
    let n = assert_ready!(tx.poll_write_vectored(&mut noop_context(), io_slices)).unwrap();
    assert_eq!(n, 0);
 }
+
+/// The `Sender::poll_write_vectored` should return `Poll::Ready(Ok(0))`
+/// if all the input buffers have zero length, even when the channel is full.
+#[tokio::test]
+async fn poll_write_vectored_4() {
+    let (mut tx, _rx) = simplex::new(1);
+    tx.write_all(&[1]).await.unwrap();
+
+    let io_slices = &[IoSlice::new(&[]), IoSlice::new(&[])];
+    tokio::pin!(tx);
+    let n = assert_ready!(tx.poll_write_vectored(&mut noop_context(), io_slices)).unwrap();
+    assert_eq!(n, 0);
+}
@@ -33,3 +33,22 @@ async fn test_stream_reader() -> std::io::Result<()> {

    Ok(())
 }
+
+#[tokio::test]
+async fn test_stream_reader_does_not_poll_after_eof() -> std::io::Result<()> {
+    // the first poll of this stream will return `Poll::Ready(None)`,
+    // and the second poll will panic
+    let stream = futures::stream::unfold((), |_| async { None::<(std::io::Result<Bytes>, ())> });
+    let read = StreamReader::new(stream);
+    tokio::pin!(read);
+    let mut buf = [0; 1];
+
+    // the first poll hits the inner stream,
+    // and the inner stream returns `Poll::Ready(None)`.
+    assert_eq!(read.read(&mut buf).await?, 0);
+    // the second poll doesn't hit the inner stream,
+    // so this `.read()` doesn't panic.
+    assert_eq!(read.read(&mut buf).await?, 0);
+
+    Ok(())
+}
@@ -70,6 +70,26 @@ macro_rules! assert_done {
    }};
 }

+fn poll_ready_bytes<T>(
+    sink: Pin<&mut T>,
+    cx: &mut Context<'_>,
+) -> Poll<Result<(), <T as Sink<Bytes>>::Error>>
+where
+    T: Sink<Bytes>,
+{
+    Sink::<Bytes>::poll_ready(sink, cx)
+}
+
+fn poll_flush_bytes<T>(
+    sink: Pin<&mut T>,
+    cx: &mut Context<'_>,
+) -> Poll<Result<(), <T as Sink<Bytes>>::Error>>
+where
+    T: Sink<Bytes>,
+{
+    Sink::<Bytes>::poll_flush(sink, cx)
+}
+
 #[test]
 fn read_empty_io_yields_nothing() {
    let io = Box::pin(FramedRead::new(mock!(), LengthDelimitedCodec::new()));
@@ -423,9 +443,9 @@ fn write_single_frame_length_adjusted() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));
-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));
        assert!(io.get_ref().calls.is_empty());
    });
 }
@@ -436,7 +456,7 @@ fn write_nothing_yields_nothing() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io, cx));
    });
 }

@@ -453,9 +473,31 @@ fn write_single_frame_one_packet() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));
-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));
+        assert!(io.get_ref().calls.is_empty());
+    });
+}
+
+#[test]
+fn write_single_frame_from_slice() {
+    let io = FramedWrite::new(
+        mock! {
+            data(b"\x00\x00\x00\x09"),
+            data(b"abcdefghi"),
+            flush(),
+        },
+        LengthDelimitedCodec::new(),
+    );
+    pin_mut!(io);
+
+    task::spawn(()).enter(|cx, _| {
+        let data: &[u8] = b"abcdefghi";
+
+        assert_ready_ok!(Sink::<&[u8]>::poll_ready(io.as_mut(), cx));
+        assert_ok!(io.as_mut().start_send(data));
+        assert_ready_ok!(Sink::<&[u8]>::poll_flush(io.as_mut(), cx));
        assert!(io.get_ref().calls.is_empty());
    });
 }
@@ -477,16 +519,16 @@ fn write_single_multi_frame_one_packet() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));

-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("123")));

-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("hello world")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));
        assert!(io.get_ref().calls.is_empty());
    });
 }
@@ -510,20 +552,20 @@ fn write_single_multi_frame_multi_packet() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));

-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("123")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));

-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("hello world")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));
        assert!(io.get_ref().calls.is_empty());
    });
 }
@@ -544,12 +586,12 @@ fn write_single_frame_would_block() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));

-        assert_pending!(io.as_mut().poll_flush(cx));
-        assert_pending!(io.as_mut().poll_flush(cx));
-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_pending!(poll_flush_bytes(io.as_mut(), cx));
+        assert_pending!(poll_flush_bytes(io.as_mut(), cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));

        assert!(io.get_ref().calls.is_empty());
    });
@@ -567,10 +609,10 @@ fn write_single_frame_little_endian() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));
        assert!(io.get_ref().calls.is_empty());
    });
 }
@@ -587,10 +629,10 @@ fn write_single_frame_with_short_length_field() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdefghi")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));

        assert!(io.get_ref().calls.is_empty());
    });
@@ -604,7 +646,7 @@ fn write_max_frame_len() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_err!(io.as_mut().start_send(Bytes::from("abcdef")));

        assert!(io.get_ref().calls.is_empty());
@@ -621,10 +663,10 @@ fn write_update_max_frame_len_at_rest() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdef")));

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));

        io.encoder_mut().set_max_frame_length(5);

@@ -646,14 +688,14 @@ fn write_update_max_frame_len_in_flight() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdef")));

-        assert_pending!(io.as_mut().poll_flush(cx));
+        assert_pending!(poll_flush_bytes(io.as_mut(), cx));

        io.encoder_mut().set_max_frame_length(5);

-        assert_ready_ok!(io.as_mut().poll_flush(cx));
+        assert_ready_ok!(poll_flush_bytes(io.as_mut(), cx));

        assert_err!(io.as_mut().start_send(Bytes::from("abcdef")));
        assert!(io.get_ref().calls.is_empty());
@@ -666,10 +708,10 @@ fn write_zero() {
    pin_mut!(io);

    task::spawn(()).enter(|cx, _| {
-        assert_ready_ok!(io.as_mut().poll_ready(cx));
+        assert_ready_ok!(poll_ready_bytes(io.as_mut(), cx));
        assert_ok!(io.as_mut().start_send(Bytes::from("abcdef")));

-        assert_ready_err!(io.as_mut().poll_flush(cx));
+        assert_ready_err!(poll_flush_bytes(io.as_mut(), cx));

        assert!(io.get_ref().calls.is_empty());
    });
@@ -699,6 +741,33 @@ fn frame_does_not_fit() {
    assert_eq!(codec.max_frame_length(), 255);
 }

+#[test]
+fn runtime_max_frame_len_respects_length_field() {
+    for (adjustment, max_frame_len) in [(-1, 254), (0, 255), (1, 256)] {
+        let mut codec = LengthDelimitedCodec::builder()
+            .length_field_length(1)
+            .length_adjustment(adjustment)
+            .new_codec();
+
+        codec.set_max_frame_length(1_000);
+        assert_eq!(codec.max_frame_length(), max_frame_len);
+
+        let mut dst = BytesMut::new();
+        codec
+            .encode(Bytes::from(vec![0; max_frame_len]), &mut dst)
+            .unwrap();
+        assert_eq!(dst[0], u8::MAX);
+
+        let mut dst = BytesMut::from(&b"prefix"[..]);
+        let original = dst.clone();
+        let result = codec.encode(Bytes::from(vec![0; max_frame_len + 1]), &mut dst);
+
+        assert!(result.is_err());
+        assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
+        assert_eq!(dst, original);
+    }
+}
+
 #[test]
 fn neg_adjusted_frame_does_not_fit() {
    let codec = LengthDelimitedCodec::builder()
@@ -581,7 +581,6 @@ mod spawn_local {
        map.spawn_local((), async {});
    }

-    #[cfg(tokio_unstable)]
    mod local_runtime {
        use super::*;

@@ -693,7 +692,6 @@ mod spawn_local {
 mod spawn_local_on {
    use super::*;

-    #[cfg(tokio_unstable)]
    mod local_runtime {
        use super::*;

@@ -1,5 +1,8 @@
 #![warn(rust_2018_idioms)]

+use std::task::Context;
+
+use futures_test::task::new_count_waker;
 use tokio::sync::oneshot;
 use tokio::task::yield_now;
 use tokio::time::Duration;
@@ -276,6 +279,60 @@ async fn test_join_queue_try_join_next() {
    check_try_join_next_is_noop(&mut queue);
 }

+#[tokio::test]
+async fn test_join_queue_try_join_next_does_not_replace_waker() {
+    let (send, recv) = oneshot::channel();
+    let mut queue = JoinQueue::new();
+    queue.spawn(async move {
+        recv.await.unwrap();
+        42
+    });
+
+    let (waker, wake_count) = new_count_waker();
+    let mut cx = Context::from_waker(&waker);
+
+    assert_pending!(queue.poll_join_next(&mut cx));
+    assert_eq!(wake_count, 0);
+    assert!(queue.try_join_next().is_none());
+
+    send.send(()).unwrap();
+    yield_now().await;
+
+    assert_eq!(wake_count, 1);
+    assert_eq!(
+        assert_ready!(queue.poll_join_next(&mut cx))
+            .unwrap()
+            .unwrap(),
+        42
+    );
+}
+
+#[tokio::test]
+async fn test_join_queue_try_join_next_with_id_does_not_replace_waker() {
+    let (send, recv) = oneshot::channel();
+    let mut queue = JoinQueue::new();
+    queue.spawn(async move {
+        recv.await.unwrap();
+        42
+    });
+
+    let (waker, wake_count) = new_count_waker();
+    let mut cx = Context::from_waker(&waker);
+
+    assert_pending!(queue.poll_join_next_with_id(&mut cx));
+    assert_eq!(wake_count, 0);
+    assert!(queue.try_join_next_with_id().is_none());
+
+    send.send(()).unwrap();
+    yield_now().await;
+
+    assert_eq!(wake_count, 1);
+    let (_, output) = assert_ready!(queue.poll_join_next_with_id(&mut cx))
+        .unwrap()
+        .unwrap();
+    assert_eq!(output, 42);
+}
+
 #[tokio::test]
 async fn test_join_queue_try_join_next_disabled_coop() {
    // This number is large enough to trigger coop. Without using `tokio::task::coop::unconstrained`
@@ -1,7 +1,6 @@
 #![warn(rust_2018_idioms)]

 use futures::future::pending;
-#[cfg(tokio_unstable)]
 use std::rc::Rc;
 use tokio::sync::mpsc;
 use tokio::task::LocalSet;
@@ -182,7 +181,6 @@ fn notify_many() {
    }
 }

-#[cfg(tokio_unstable)]
 mod spawn {
    use super::*;

@@ -209,7 +207,6 @@ mod spawn {
    }
 }

-#[cfg(tokio_unstable)]
 mod spawn_local {
    use super::*;

@@ -278,7 +275,6 @@ mod spawn_local {
 mod spawn_local_on {
    use super::*;

-    #[cfg(tokio_unstable)]
    mod local_runtime {
        use super::*;

@@ -202,6 +202,27 @@ async fn reset_entry() {
    assert!(entry.is_none())
 }

+#[tokio::test]
+async fn reset_to_past_wakes_pending_queue() {
+    time::pause();
+
+    let mut queue = task::spawn(DelayQueue::new());
+    let key = queue.insert("foo", ms(10_000));
+
+    assert_pending!(poll!(queue));
+    assert!(!queue.is_woken());
+
+    queue.reset_at(&key, Instant::now() - ms(100));
+
+    assert!(queue.is_woken());
+
+    let entry = assert_ready_some!(poll!(queue));
+    assert_eq!(*entry.get_ref(), "foo");
+
+    let entry = assert_ready!(poll!(queue));
+    assert!(entry.is_none());
+}
+
 // Reproduces tokio-rs/tokio#849.
 #[tokio::test]
 async fn reset_much_later() {
@@ -894,6 +915,20 @@ async fn wake_after_remove_last() {
    assert!(assert_ready!(poll!(queue)).is_none());
 }

+#[tokio::test(start_paused = true)]
+async fn wake_after_clear() {
+    let mut queue = task::spawn(DelayQueue::new());
+    queue.insert("foo", ms(1000));
+
+    assert_pending!(poll!(queue));
+    assert!(!queue.is_woken());
+
+    queue.clear();
+
+    assert!(queue.is_woken());
+    assert!(assert_ready!(poll!(queue)).is_none());
+}
+
 fn ms(n: u64) -> Duration {
    Duration::from_millis(n)
 }
@@ -1,3 +1,119 @@
+# 1.53.1 (July 20th, 2026)
+
+### Fixed
+
+- signal: restore MSRV by removing `OnceLock::wait` from the Windows handler ([#8300])
+
+### Fixed (unstable)
+
+- time: fix alt timer cancellation and insertion race ([#8252])
+
+### Documented
+
+- runtime: remove dead link definition in Runtime::block_on ([#8301])
+
+[#8252]: https://github.com/tokio-rs/tokio/pull/8252
+[#8300]: https://github.com/tokio-rs/tokio/pull/8300
+[#8301]: https://github.com/tokio-rs/tokio/pull/8301
+
+# 1.53.0 (July 17th, 2026)
+
+### Added
+
+- fs: implement `From<OwnedFd>` and `From<OwnedHandle>` for `File` ([#8266])
+- metrics: add task schedule latency metric ([#7986])
+- net: add `SocketAddr` methods to Unix sockets ([#8144])
+
+### Changed
+
+- io: add `#[inline]` to IO trait impls for in-memory types ([#8242])
+- net: implement UCred::pid on FreeBSD ([#8086])
+- net: support Nuttx target os ([#8259])
+- signal: refactor global variables on Windows ([#8231])
+- sync: `mpsc::{Receiver,UnboundedReceiver}` now drops waker on drop, even if there are still senders ([#8095])
+- taskdump: support taskdumps on s390x ([#8192])
+- time: add `#[track_caller]` to `timeout_at()` ([#8077])
+- time: consolidate mutex locks on spurious poll ([#8124])
+- time: defer waker clone on spurious poll ([#8107])
+- time: move lazy-registration state into `Sleep` ([#8132])
+- tracing: remove unnecessary span clone ([#8126])
+
+### Fixed
+
+- io: do not treat zero-length reads as EOF in `Chain` ([#8251])
+- net: use getpeereid for QNX peer credentials ([#8270])
+- runtime: avoid illegal state in `FastRand` ([#8078])
+- sync: wake mpsc receiver when a queued `reserve[_many]` returns permits ([#8260])
+- taskdump: skip double wake on `Trace::capture`/`Trace::trace_with` ([#8043])
+- time: avoid stack overflow in runtime constructor ([#8093])
+- time (alt timer): ensure timers stay in the same runtime after `.reset()` ([#8169])
+
+### IO uring (unstable)
+
+- fs: use io-uring for `fs::try_exists` ([#8080])
+- fs: use io-uring for renaming files ([#7800])
+- rt: flush io-uring CQE in case of CQE overflow ([#8277])
+
+### Documented
+
+- docs: clarify cancel safety wording ([#8181])
+- fs: clarify `create_dir_all` succeeds if path exists ([#8149])
+- io: add warning about stdout reordering with multiple handles ([#8276])
+- net: document pipe `try_read*`/`try_write*` readiness behavior ([#8032])
+- runtime: document interaction with fork() ([#8202])
+- sync: clarify broadcast lagging semantics ([#8239])
+- sync: document memory ordering guarantees for Semaphore ([#8119])
+- task: explain why `yield_now` defers its waker ([#8254])
+- time: add panic docs to `timeout_at()` ([#8077])
+- time: fix reversed poll order in timeout doc ([#8214])
+
+[#7800]: https://github.com/tokio-rs/tokio/pull/7800
+[#7986]: https://github.com/tokio-rs/tokio/pull/7986
+[#8032]: https://github.com/tokio-rs/tokio/pull/8032
+[#8043]: https://github.com/tokio-rs/tokio/pull/8043
+[#8077]: https://github.com/tokio-rs/tokio/pull/8077
+[#8078]: https://github.com/tokio-rs/tokio/pull/8078
+[#8080]: https://github.com/tokio-rs/tokio/pull/8080
+[#8086]: https://github.com/tokio-rs/tokio/pull/8086
+[#8093]: https://github.com/tokio-rs/tokio/pull/8093
+[#8095]: https://github.com/tokio-rs/tokio/pull/8095
+[#8107]: https://github.com/tokio-rs/tokio/pull/8107
+[#8119]: https://github.com/tokio-rs/tokio/pull/8119
+[#8124]: https://github.com/tokio-rs/tokio/pull/8124
+[#8126]: https://github.com/tokio-rs/tokio/pull/8126
+[#8132]: https://github.com/tokio-rs/tokio/pull/8132
+[#8144]: https://github.com/tokio-rs/tokio/pull/8144
+[#8149]: https://github.com/tokio-rs/tokio/pull/8149
+[#8169]: https://github.com/tokio-rs/tokio/pull/8169
+[#8181]: https://github.com/tokio-rs/tokio/pull/8181
+[#8192]: https://github.com/tokio-rs/tokio/pull/8192
+[#8193]: https://github.com/tokio-rs/tokio/pull/8193
+[#8202]: https://github.com/tokio-rs/tokio/pull/8202
+[#8214]: https://github.com/tokio-rs/tokio/pull/8214
+[#8231]: https://github.com/tokio-rs/tokio/pull/8231
+[#8239]: https://github.com/tokio-rs/tokio/pull/8239
+[#8242]: https://github.com/tokio-rs/tokio/pull/8242
+[#8251]: https://github.com/tokio-rs/tokio/pull/8251
+[#8254]: https://github.com/tokio-rs/tokio/pull/8254
+[#8259]: https://github.com/tokio-rs/tokio/pull/8259
+[#8260]: https://github.com/tokio-rs/tokio/pull/8260
+[#8266]: https://github.com/tokio-rs/tokio/pull/8266
+[#8270]: https://github.com/tokio-rs/tokio/pull/8270
+[#8276]: https://github.com/tokio-rs/tokio/pull/8276
+[#8277]: https://github.com/tokio-rs/tokio/pull/8277
+
+# 1.52.4 (July 16th, 2026)
+
+### Fixed
+
+- runtime: don't skip the driver when `before_park` schedules work ([#8222])
+
+### Fixed (unstable)
+
+- taskdump: remove crate disambiguators from output ([#8264])
+
+[#8264]: https://github.com/tokio-rs/tokio/pull/8264
+
 # 1.52.3 (May 8th, 2026)

 ### Fixed
@@ -64,6 +180,14 @@ This release reverts the LIFO slot stealing change introduced in 1.51.0
 [#8035]: https://github.com/tokio-rs/tokio/pull/8035
 [#8040]: https://github.com/tokio-rs/tokio/pull/8040

+# 1.51.4 (July 16th, 2026)
+
+### Fixed
+
+- runtime: don't skip the driver when `before_park` schedules work ([#8222])
+
+[#8222]: https://github.com/tokio-rs/tokio/pull/8222
+
 # 1.51.3 (May 8th, 2026)

 ### Fixed
@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.52.3"
+version = "1.53.1"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -88,6 +88,8 @@ time = []
 io-uring = ["dep:io-uring", "libc", "mio/os-poll", "mio/os-ext", "dep:slab"]
 # Unstable feature. Requires `--cfg tokio_unstable` to enable.
 taskdump = ["dep:backtrace"]
+# Unstable feature. Requires `--cfg tokio_unstable` to enable.
+schedule-latency = []

 [dependencies]
 tokio-macros = { version = "~2.7.0", optional = true }
@@ -1,7 +1,3 @@
-*[TokioConf 2026 program and tickets are now available!](https://tokioconf.com)*
-
---
-
 # Tokio

 A runtime for writing reliable, asynchronous, and slim applications with
@@ -60,7 +56,7 @@ Make sure you enable the full features of the tokio crate on Cargo.toml:

 ```toml
 [dependencies]
-tokio = { version = "1.52.3", features = ["full"] }
+tokio = { version = "1.53.1", features = ["full"] }
 ```
 Then, on your main.rs:

@@ -52,6 +52,7 @@ impl Registration {

    /// Clears resource level readiness represented by the specified `ReadyEvent`
    async fn clear_readiness(&self, ready_event: ReadyEvent);
+}
 ```

 A new registration is created for a `T: mio::Evented` and a `interest`. This
@@ -418,7 +418,14 @@ impl File {

        let (op, buf) = match inner.state {
            State::Idle(_) => unreachable!(),
-            State::Busy(ref mut rx) => rx.await?,
+            State::Busy(ref mut rx) => {
+                let res = rx.await;
+                if res.is_err() {
+                    // Restore a valid Idle state before returning the error.
+                    inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                }
+                res?
+            }
        };

        inner.state = State::Idle(Some(buf));
@@ -621,7 +628,12 @@ impl AsyncRead for File {
                    inner.state = State::Busy(Inner::poll_read_inner(std, buf, max_buf_size)?);
                }
                State::Busy(ref mut rx) => {
-                    let (op, mut buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, mut buf) = res?;

                    match op {
                        Operation::Read(Ok(_)) => {
@@ -701,7 +713,12 @@ impl AsyncSeek for File {
            match inner.state {
                State::Idle(_) => return Poll::Ready(Ok(inner.pos)),
                State::Busy(ref mut rx) => {
-                    let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, buf) = res?;
                    inner.state = State::Idle(Some(buf));

                    match op {
@@ -752,7 +769,7 @@ impl AsyncWrite for File {
                    let n = buf.copy_from(src, me.max_buf_size);
                    let std = me.std.clone();

-                    let blocking_task_join_handle = spawn_mandatory_blocking(move || {
+                    let res = spawn_mandatory_blocking(move || {
                        let res = if let Some(seek) = seek {
                            (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std))
                        } else {
@@ -761,16 +778,25 @@ impl AsyncWrite for File {

                        (Operation::Write(res), buf)
                    })
-                    .ok_or_else(|| {
-                        io::Error::new(io::ErrorKind::Other, "background task failed")
-                    })?;
+                    .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "background task failed"));
+
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let blocking_task_join_handle = res?;

                    inner.state = State::Busy(blocking_task_join_handle);

                    return Poll::Ready(Ok(n));
                }
                State::Busy(ref mut rx) => {
-                    let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, buf) = res?;
                    inner.state = State::Idle(Some(buf));

                    match op {
@@ -823,7 +849,7 @@ impl AsyncWrite for File {
                    let n = buf.copy_from_bufs(bufs, me.max_buf_size);
                    let std = me.std.clone();

-                    let blocking_task_join_handle = spawn_mandatory_blocking(move || {
+                    let res = spawn_mandatory_blocking(move || {
                        let res = if let Some(seek) = seek {
                            (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std))
                        } else {
@@ -832,16 +858,25 @@ impl AsyncWrite for File {

                        (Operation::Write(res), buf)
                    })
-                    .ok_or_else(|| {
-                        io::Error::new(io::ErrorKind::Other, "background task failed")
-                    })?;
+                    .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "background task failed"));
+
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let blocking_task_join_handle = res?;

                    inner.state = State::Busy(blocking_task_join_handle);

                    return Poll::Ready(Ok(n));
                }
                State::Busy(ref mut rx) => {
-                    let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, buf) = res?;
                    inner.state = State::Idle(Some(buf));

                    match op {
@@ -897,6 +932,13 @@ impl fmt::Debug for File {
    }
 }

+#[cfg(unix)]
+impl From<std::os::fd::OwnedFd> for File {
+    fn from(fd: std::os::fd::OwnedFd) -> Self {
+        Self::from_std(StdFile::from(fd))
+    }
+}
+
 #[cfg(unix)]
 impl std::os::unix::io::AsRawFd for File {
    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
@@ -923,7 +965,13 @@ impl std::os::unix::io::FromRawFd for File {
 }

 cfg_windows! {
-    use crate::os::windows::io::{AsRawHandle, FromRawHandle, RawHandle, AsHandle, BorrowedHandle};
+    use crate::os::windows::io::{AsRawHandle, FromRawHandle, RawHandle, AsHandle, BorrowedHandle, OwnedHandle};
+
+    impl From<OwnedHandle> for File {
+        fn from(handle: OwnedHandle) -> Self {
+            Self::from_std(StdFile::from(handle))
+        }
+    }

    impl AsRawHandle for File {
        fn as_raw_handle(&self) -> RawHandle {
@@ -1040,7 +1088,7 @@ impl Inner {
        if driver_handle
            .check_and_init(io_uring::opcode::Read::CODE)
            .await
-            .unwrap_or(false)
+            .unwrap_or_default()
        {
            let fd: crate::io::uring::utils::ArcFd = std;
            Self::uring_read(fd, buf, max_buf_size).await
@@ -1095,7 +1143,14 @@ impl Inner {

        let (op, buf) = match self.state {
            State::Idle(_) => return Poll::Ready(Ok(())),
-            State::Busy(ref mut rx) => ready!(Pin::new(rx).poll(cx))?,
+            State::Busy(ref mut rx) => {
+                let res = ready!(Pin::new(rx).poll(cx));
+                if res.is_err() {
+                    // Restore a valid Idle state before returning the error.
+                    self.state = State::Idle(Some(Buf::with_capacity(0)));
+                }
+                res?
+            }
        };

        // The buffer is not used here
@@ -38,6 +38,10 @@ mock! {
        pub fn try_clone(&self) -> io::Result<Self>;
    }
    #[cfg(windows)]
+    impl From<std::os::windows::io::OwnedHandle> for File {
+        fn from(handle: std::os::windows::io::OwnedHandle) -> Self;
+    }
+    #[cfg(windows)]
    impl std::os::windows::io::AsRawHandle for File {
        fn as_raw_handle(&self) -> std::os::windows::io::RawHandle;
    }
@@ -106,6 +110,15 @@ impl From<MockFile> for OwnedFd {
    }
 }

+#[cfg(all(test, unix))]
+impl From<OwnedFd> for MockFile {
+    #[inline]
+    fn from(file: OwnedFd) -> MockFile {
+        use std::os::fd::IntoRawFd;
+        unsafe { MockFile::from_raw_fd(IntoRawFd::into_raw_fd(file)) }
+    }
+}
+
 tokio_thread_local! {
    static QUEUE: RefCell<VecDeque<Box<dyn FnOnce() + Send>>> = RefCell::new(VecDeque::new())
 }
@@ -16,9 +16,17 @@
 //! such as hangs during runtime shutdown. For special files, you should use a
 //! dedicated type such as [`tokio::net::unix::pipe`] or [`AsyncFd`] instead.
 //!
-//! Currently, Tokio will always use [`spawn_blocking`] on all platforms, but it
-//! may be changed to use asynchronous file system APIs such as io_uring in the
-//! future.
+//! Tokio currently uses [`spawn_blocking`] on all platforms, but also uses
+//! io_uring for file operations on Linux when compiled with `tokio_unstable`.
+//! Operations that cannot be cancelled with [`spawn_blocking`] may possibly
+//! be cancelled with io_uring.
+//!
+//! # Cancellation
+//!
+//! Cancelling a future from this module will stop waiting for the result, but
+//! the underlying blocking operation will continue to run on the thread pool.
+//! For example, cancelling a [`write()`] future after it has been
+//! polled will still result in the data being written to disk.
 //!
 //! # Usage
 //!
@@ -518,6 +518,10 @@ impl OpenOptions {
    /// [`Other`]: std::io::ErrorKind::Other
    /// [`PermissionDenied`]: std::io::ErrorKind::PermissionDenied
    pub async fn open(&self, path: impl AsRef<Path>) -> io::Result<File> {
+        self.open_inner(path.as_ref()).await
+    }
+
+    async fn open_inner(&self, path: &Path) -> io::Result<File> {
        match &self.inner {
            Kind::Std(opts) => Self::std_open(opts, path).await,
            #[cfg(all(
@@ -535,7 +539,7 @@ impl OpenOptions {
                    .check_and_init(io_uring::opcode::OpenAt::CODE)
                    .await?
                {
-                    Op::open(path.as_ref(), opts)?.await
+                    Op::open(path, opts)?.await
                } else {
                    let opts = opts.clone().into();
                    Self::std_open(&opts, path).await
@@ -544,12 +548,11 @@ impl OpenOptions {
        }
    }

-    async fn std_open(opts: &StdOpenOptions, path: impl AsRef<Path>) -> io::Result<File> {
-        let path = path.as_ref().to_owned();
+    async fn std_open(opts: &StdOpenOptions, path: &Path) -> io::Result<File> {
+        let path = path.to_owned();
        let opts = opts.clone();

-        let std = asyncify(move || opts.open(path)).await?;
-        Ok(File::from_std(std))
+        Ok(asyncify(move || opts.open(path)).await?.into())
    }

    #[cfg(windows)]
@@ -54,14 +54,23 @@ use std::{io, path::Path};
 /// }
 /// ```
 pub async fn read(path: impl AsRef<Path>) -> io::Result<Vec<u8>> {
-    let path = path.as_ref().to_owned();
+    let path = path.as_ref();

    #[cfg(all(
        tokio_unstable,
        feature = "io-uring",
        feature = "rt",
        feature = "fs",
-        target_os = "linux"
+        // libc::statx is only supported on these platforms
+        // FIXME: Add musl target env when our minimum supported
+        // rust version is 1.93. To clarify, statx support is
+        // introduced to musl in 1.25 as mentioned officially here:
+        // https://musl.libc.org/releases.html.
+        // However, rustup target_env building for *-linux-musl
+        // uses 1.25 musl on all *-linux-musl platforms starting
+        // in 1.93 stable rust version.
+        // https://blog.rust-lang.org/2025/12/05/Updating-musl-1.2.5/
+        any(target_env = "gnu", target_os = "android")
    ))]
    {
        use crate::fs::read_uring;
@@ -72,9 +81,14 @@ pub async fn read(path: impl AsRef<Path>) -> io::Result<Vec<u8>> {
            .check_and_init(io_uring::opcode::Read::CODE)
            .await?
        {
-            return read_uring(&path).await;
+            return read_uring(path).await;
        }
    }

+    read_spawn_blocking(path).await
+}
+
+async fn read_spawn_blocking(path: &Path) -> io::Result<Vec<u8>> {
+    let path = path.to_owned();
    asyncify(move || std::fs::read(path)).await
 }
@@ -74,7 +74,7 @@ impl ReadDir {
    ///
    /// # Cancel safety
    ///
-    /// This method is cancellation safe.
+    /// This method is cancel safe.
    pub async fn next_entry(&mut self) -> io::Result<Option<DirEntry>> {
        use std::future::poll_fn;
        poll_fn(|cx| self.poll_next_entry(cx)).await
@@ -19,9 +19,26 @@ const MAX_READ_SIZE: usize = 64 * 1024 * 1024;
 pub(crate) async fn read_uring(path: &Path) -> io::Result<Vec<u8>> {
    let file = OpenOptions::new().read(true).open(path).await?;

-    // TODO: use io uring in the future to obtain metadata
+    #[cfg(not(any(target_env = "gnu", target_os = "android")))]
    let size_hint: Option<usize> = file.metadata().await.map(|m| m.len() as usize).ok();

+    #[cfg(
+        // libc::statx is only supported on these platforms
+        // FIXME: Add musl target env when our minimum supported
+        // rust version is 1.93. To clarify, statx support is
+        // introduced to musl in 1.25 as mentioned officially here:
+        // https://musl.libc.org/releases.html.
+        // However, rustup target_env building for *-linux-musl
+        // uses 1.25 musl on all *-linux-musl platforms starting
+        // in 1.93 stable rust version.
+        // https://blog.rust-lang.org/2025/12/05/Updating-musl-1.2.5/
+        any(target_env = "gnu", target_os = "android")
+    )]
+    let size_hint = Op::file_metadata(&file)?
+        .await
+        .map(|m| m.len() as usize)
+        .ok();
+
    let fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
@@ -10,8 +10,37 @@ use std::path::Path;
 ///
 /// This is an async version of [`std::fs::rename`].
 pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
-    let from = from.as_ref().to_owned();
-    let to = to.as_ref().to_owned();
+    let from = from.as_ref();
+    let to = to.as_ref();

+    #[cfg(all(
+        tokio_unstable,
+        feature = "io-uring",
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux",
+    ))]
+    {
+        use crate::io::uring::rename::Rename;
+        use crate::runtime::driver::op::Op;
+
+        let handle = crate::runtime::Handle::current();
+        let driver_handle = handle.inner.driver().io();
+
+        type RenameOp = Op<Rename>;
+
+        if driver_handle
+            .check_and_init(io_uring::opcode::RenameAt::CODE)
+            .await?
+        {
+            return RenameOp::rename(from, to)?.await;
+        }
+    }
+
+    rename_blocking(from, to).await
+}
+
+async fn rename_blocking(from: &Path, to: &Path) -> io::Result<()> {
+    let [from, to] = [from, to].map(Path::to_owned);
    asyncify(move || std::fs::rename(from, to)).await
 }
@@ -23,6 +23,67 @@ use std::path::Path;
 /// # }
 /// ```
 pub async fn try_exists(path: impl AsRef<Path>) -> io::Result<bool> {
-    let path = path.as_ref().to_owned();
+    let path = path.as_ref();
+
+    #[cfg(all(
+        tokio_unstable,
+        feature = "io-uring",
+        feature = "rt",
+        feature = "fs",
+        // libc::statx is only supported on these platforms
+        // FIXME: Add musl target env when our minimum supported
+        // rust version is 1.93. To clarify, statx support is
+        // introduced to musl in 1.25 as mentioned officially here:
+        // https://musl.libc.org/releases.html.
+        // However, rustup target_env building for *-linux-musl
+        // uses 1.25 musl on all *-linux-musl platforms starting
+        // in 1.93 stable rust version.
+        // https://blog.rust-lang.org/2025/12/05/Updating-musl-1.2.5/
+        any(target_env = "gnu", target_os = "android")
+    ))]
+    {
+        let handle = crate::runtime::Handle::current();
+        let driver_handle = handle.inner.driver().io();
+        if driver_handle
+            .check_and_init(io_uring::opcode::Statx::CODE)
+            .await?
+        {
+            return try_exists_uring(path).await;
+        }
+    }
+
+    try_exists_spawn_blocking(path).await
+}
+
+cfg_io_uring! {
+    #[inline]
+    #[cfg(
+        // libc::statx is only supported on these platforms
+        // FIXME: Add musl target env when our minimum supported
+        // rust version is 1.93. To clarify, statx support is
+        // introduced to musl in 1.25 as mentioned officially here:
+        // https://musl.libc.org/releases.html.
+        // However, rustup target_env building for *-linux-musl
+        // uses 1.25 musl on all *-linux-musl platforms starting
+        // in 1.93 stable rust version.
+        // https://blog.rust-lang.org/2025/12/05/Updating-musl-1.2.5/
+        any(target_env = "gnu", target_os = "android")
+    )]
+    async fn try_exists_uring(path: &Path) -> io::Result<bool> {
+        use crate::runtime::driver::op::Op;
+
+        match Op::metadata(path)?.await {
+            Ok(_) => Ok(true),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
+            Err(error) => Err(error),
+        }
+    }
+}
+
+async fn try_exists_spawn_blocking(path: &Path) -> io::Result<bool> {
+    let path = path.to_owned();
+    // FIXME: When MSRV is 1.81, change this to
+    // std::fs::exists() to be consistent with
+    // all other tokio::fs operations
    asyncify(move || path.try_exists()).await
 }
@@ -97,20 +97,24 @@ where
 }

 impl AsyncBufRead for &[u8] {
+    #[inline]
    fn poll_fill_buf(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
        Poll::Ready(Ok(*self))
    }

+    #[inline]
    fn consume(mut self: Pin<&mut Self>, amt: usize) {
        *self = &self[amt..];
    }
 }

 impl<T: AsRef<[u8]> + Unpin> AsyncBufRead for io::Cursor<T> {
+    #[inline]
    fn poll_fill_buf(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
        Poll::Ready(io::BufRead::fill_buf(self.get_mut()))
    }

+    #[inline]
    fn consume(self: Pin<&mut Self>, amt: usize) {
        io::BufRead::consume(self.get_mut(), amt);
    }
@@ -94,6 +94,7 @@ where
 }

 impl AsyncRead for &[u8] {
+    #[inline]
    fn poll_read(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
@@ -108,6 +109,7 @@ impl AsyncRead for &[u8] {
 }

 impl<T: AsRef<[u8]> + Unpin> AsyncRead for io::Cursor<T> {
+    #[inline]
    fn poll_read(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
@@ -87,9 +87,11 @@ where
 }

 impl<T: AsRef<[u8]> + Unpin> AsyncSeek for io::Cursor<T> {
+    #[inline]
    fn start_seek(mut self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
        io::Seek::seek(&mut *self, pos).map(drop)
    }
+    #[inline]
    fn poll_complete(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<u64>> {
        Poll::Ready(Ok(self.get_mut().position()))
    }
@@ -251,6 +251,7 @@ where
 }

 impl AsyncWrite for Vec<u8> {
+    #[inline]
    fn poll_write(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
@@ -260,6 +261,7 @@ impl AsyncWrite for Vec<u8> {
        Poll::Ready(Ok(buf.len()))
    }

+    #[inline]
    fn poll_write_vectored(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -268,20 +270,24 @@ impl AsyncWrite for Vec<u8> {
        Poll::Ready(io::Write::write_vectored(&mut *self, bufs))
    }

+    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

+    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(()))
    }

+    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(()))
    }
 }

 impl AsyncWrite for io::Cursor<&mut [u8]> {
+    #[inline]
    fn poll_write(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -290,6 +296,7 @@ impl AsyncWrite for io::Cursor<&mut [u8]> {
        Poll::Ready(io::Write::write(&mut *self, buf))
    }

+    #[inline]
    fn poll_write_vectored(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -298,20 +305,24 @@ impl AsyncWrite for io::Cursor<&mut [u8]> {
        Poll::Ready(io::Write::write_vectored(&mut *self, bufs))
    }

+    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

+    #[inline]
    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(io::Write::flush(&mut *self))
    }

+    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        self.poll_flush(cx)
    }
 }

 impl AsyncWrite for io::Cursor<&mut Vec<u8>> {
+    #[inline]
    fn poll_write(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -320,6 +331,7 @@ impl AsyncWrite for io::Cursor<&mut Vec<u8>> {
        Poll::Ready(io::Write::write(&mut *self, buf))
    }

+    #[inline]
    fn poll_write_vectored(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -328,20 +340,24 @@ impl AsyncWrite for io::Cursor<&mut Vec<u8>> {
        Poll::Ready(io::Write::write_vectored(&mut *self, bufs))
    }

+    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

+    #[inline]
    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(io::Write::flush(&mut *self))
    }

+    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        self.poll_flush(cx)
    }
 }

 impl AsyncWrite for io::Cursor<Vec<u8>> {
+    #[inline]
    fn poll_write(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -350,6 +366,7 @@ impl AsyncWrite for io::Cursor<Vec<u8>> {
        Poll::Ready(io::Write::write(&mut *self, buf))
    }

+    #[inline]
    fn poll_write_vectored(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -358,20 +375,24 @@ impl AsyncWrite for io::Cursor<Vec<u8>> {
        Poll::Ready(io::Write::write_vectored(&mut *self, bufs))
    }

+    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

+    #[inline]
    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(io::Write::flush(&mut *self))
    }

+    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        self.poll_flush(cx)
    }
 }

 impl AsyncWrite for io::Cursor<Box<[u8]>> {
+    #[inline]
    fn poll_write(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -380,6 +401,7 @@ impl AsyncWrite for io::Cursor<Box<[u8]>> {
        Poll::Ready(io::Write::write(&mut *self, buf))
    }

+    #[inline]
    fn poll_write_vectored(
        mut self: Pin<&mut Self>,
        _: &mut Context<'_>,
@@ -388,14 +410,17 @@ impl AsyncWrite for io::Cursor<Box<[u8]>> {
        Poll::Ready(io::Write::write_vectored(&mut *self, bufs))
    }

+    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

+    #[inline]
    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(io::Write::flush(&mut *self))
    }

+    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        self.poll_flush(cx)
    }
@@ -15,8 +15,8 @@ use std::task::{ready, Context, Poll};

 /// Like [`mio::event::Source`], but for POSIX AIO only.
 ///
-/// Tokio's consumer must pass an implementor of this trait to create a
-/// [`Aio`] object.  Implementors must implement at least one of [`AioSource::register`] and
+/// Tokio's consumer must pass an implementer of this trait to create a
+/// [`Aio`] object.  Implementers must implement at least one of [`AioSource::register`] and
 /// [`AioSource::register_borrowed`].
 pub trait AioSource {
    /// Registers this AIO event source with Tokio's reactor.
@@ -27,7 +27,7 @@ pub trait AioSource {
    /// source may end up notifying the wrong file.
    #[deprecated(since = "1.52.0", note = "use register_borrowed instead")]
    fn register(&mut self, _kq: RawFd, _token: usize) {
-        // This default implementation exists so new AioSource implementors that implement the
+        // This default implementation exists so new AioSource implementers that implement the
        // register_borrowed method can compile without the need to implement register.
        unimplemented!("Use AioSource::register_borrowed instead")
    }
@@ -35,7 +35,7 @@ pub trait AioSource {
    /// Registers this AIO event source with Tokio's reactor.
    fn register_borrowed(&mut self, kq: BorrowedFd<'_>, token: usize) {
        // This default implementation serves to provide backwards compatibility with AioSource
-        // implementors written before 1.52.0 that only implemented the unsafe `register` method.
+        // implementers written before 1.52.0 that only implemented the unsafe `register` method.
        #[allow(deprecated)]
        self.register(kq.as_raw_fd(), token)
    }
@@ -83,7 +83,7 @@ where
                .rev()
                .take(MAX_BYTES_PER_CHAR)
                .position(|byte| *byte < 0b1000_0000 || *byte >= 0b1100_0000)
-                .unwrap_or(0)
+                .unwrap_or_default()
                + 1;
            buf = &buf[..buf.len() - trailing_incomplete_char_size];
        }
@@ -15,6 +15,45 @@ cfg_io_std! {
    /// to occur as a single write, so multiple threads writing data with
    /// [`write_all`] may result in interleaved output.
    ///
+    /// # Warning
+    ///
+    /// Each call to [`stdout()`] creates a **new** handle with its own
+    /// internal state. Writes through different handles are not
+    /// coordinated, so creating a new handle in a loop can cause output
+    /// to appear out of order:
+    ///
+    /// ```no_run
+    /// # use tokio::io::{self, AsyncWriteExt};
+    /// # #[tokio::main]
+    /// # async fn main() -> std::io::Result<()> {
+    /// // WRONG: creates a new handle each iteration
+    /// for i in 0..10 {
+    ///     let mut out = io::stdout();
+    ///     out.write_all(b"data").await?;
+    ///     out.write_all(b"\n").await?;
+    ///     // out is dropped here; its last write may still be
+    ///     // running when the next iteration starts
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// To preserve order, create one handle outside the loop and
+    /// reuse it:
+    ///
+    /// ```no_run
+    /// # use tokio::io::{self, AsyncWriteExt};
+    /// # #[tokio::main]
+    /// # async fn main() -> std::io::Result<()> {
+    /// let mut out = io::stdout();
+    /// for i in 0..10 {
+    ///     out.write_all(b"data").await?;
+    ///     out.write_all(b"\n").await?;
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
    /// Created by the [`stdout`] function.
    ///
    /// [`stdout`]: stdout()
@@ -1,4 +1,6 @@
 pub(crate) mod open;
 pub(crate) mod read;
+pub(crate) mod rename;
+pub(crate) mod statx;
 pub(crate) mod utils;
 pub(crate) mod write;
--- a/Show More
+++ b/Show More