chore: prepare for Tokio v1.26.0 release (#5521 )

# 1.26.0 (March 1st, 2023) ### Fixed - macros: fix empty `join!` and `try_join!` ([#5504]) - sync: don't leak tracing spans in mutex guards ([#5469]) - sync: drop wakers after unlocking the mutex in Notify ([#5471]) - sync: drop wakers outside lock in semaphore ([#5475]) ### Added - fs: add `fs::try_exists` ([#4299]) - net: add types for named unix pipes ([#5351]) - sync: add `MappedOwnedMutexGuard` ([#5474]) ### Changed - chore: update windows-sys to 0.45 ([#5386]) - net: use Message Read Mode for named pipes ([#5350]) - sync: mark lock guards with `#[clippy::has_significant_drop]` ([#5422]) - sync: reduce contention in watch channel ([#5464]) - time: remove cache padding in timer entries ([#5468]) - time: Improve `Instant::now()` perf with test-util ([#5513]) ### Internal Changes - io: use `poll_fn` in `copy_bidirectional` ([#5486]) - net: refactor named pipe builders to not use bitfields ([#5477]) - rt: remove Arc from Clock ([#5434]) - sync: make `notify_waiters` calls atomic ([#5458]) - time: don't store deadline twice in sleep entries ([#5410]) ### Unstable - metrics: add a new metric for budget exhaustion yields ([#5517]) ### Documented - io: improve AsyncFd example ([#5481]) - runtime: document the nature of the main future ([#5494]) - runtime: remove extra period in docs ([#5511]) - signal: updated Documentation for Signals ([#5459]) - sync: add doc aliases for `blocking_*` methods ([#5448]) - sync: fix docs for Send/Sync bounds in broadcast ([#5480]) - sync: document drop behavior for channels ([#5497]) - task: clarify what happens to spawned work during runtime shutdown ([#5394]) - task: clarify `process::Command` docs ([#5413]) - task: fix wording with 'unsend' ([#5452]) - time: document immediate completion guarantee for timeouts ([#5509]) - tokio: document supported platforms ([#5483]) [#4299]: https://github.com/tokio-rs/tokio/pull/4299 [#5350]: https://github.com/tokio-rs/tokio/pull/5350 [#5351]: https://github.com/tokio-rs/tokio/pull/5351 [#5386]: https://github.com/tokio-rs/tokio/pull/5386 [#5394]: https://github.com/tokio-rs/tokio/pull/5394 [#5410]: https://github.com/tokio-rs/tokio/pull/5410 [#5413]: https://github.com/tokio-rs/tokio/pull/5413 [#5422]: https://github.com/tokio-rs/tokio/pull/5422 [#5434]: https://github.com/tokio-rs/tokio/pull/5434 [#5448]: https://github.com/tokio-rs/tokio/pull/5448 [#5452]: https://github.com/tokio-rs/tokio/pull/5452 [#5458]: https://github.com/tokio-rs/tokio/pull/5458 [#5459]: https://github.com/tokio-rs/tokio/pull/5459 [#5464]: https://github.com/tokio-rs/tokio/pull/5464 [#5468]: https://github.com/tokio-rs/tokio/pull/5468 [#5469]: https://github.com/tokio-rs/tokio/pull/5469 [#5471]: https://github.com/tokio-rs/tokio/pull/5471 [#5474]: https://github.com/tokio-rs/tokio/pull/5474 [#5475]: https://github.com/tokio-rs/tokio/pull/5475 [#5477]: https://github.com/tokio-rs/tokio/pull/5477 [#5480]: https://github.com/tokio-rs/tokio/pull/5480 [#5481]: https://github.com/tokio-rs/tokio/pull/5481 [#5483]: https://github.com/tokio-rs/tokio/pull/5483 [#5486]: https://github.com/tokio-rs/tokio/pull/5486 [#5494]: https://github.com/tokio-rs/tokio/pull/5494 [#5497]: https://github.com/tokio-rs/tokio/pull/5497 [#5504]: https://github.com/tokio-rs/tokio/pull/5504 [#5509]: https://github.com/tokio-rs/tokio/pull/5509 [#5511]: https://github.com/tokio-rs/tokio/pull/5511 [#5513]: https://github.com/tokio-rs/tokio/pull/5513 [#5517]: https://github.com/tokio-rs/tokio/pull/5517
metrics: add a new metric for budget exhaustion yields (#5517 )
2026-09-09 00:00:08 +02:00 · 2023-03-01 22:09:48 +00:00 · 2023-03-01 21:25:35 +01:00 · 2023-02-27 10:21:42 -08:00 · 2023-02-27 15:41:17 +01:00 · 2023-02-27 14:17:31 +00:00
456 changed files with 32216 additions and 10314 deletions
@@ -2,8 +2,6 @@

 [advisories]
 ignore = [
-    # https://github.com/tokio-rs/tokio/issues/4177
-    "RUSTSEC-2020-0159",
    # We depend on nix 0.22 only via mio-aio, a dev-dependency.
    # https://github.com/tokio-rs/tokio/pull/4255#issuecomment-974786349
    "RUSTSEC-2021-0119",
@@ -0,0 +1,2 @@
+# [build]
+# rustflags = ["--cfg", "tokio_unstable"]
@@ -0,0 +1,25 @@
+version: 2.1        
+jobs:
+  test-arm:
+    machine:
+      image: ubuntu-2004:202101-01
+    resource_class: arm.medium
+    environment:
+      # Change to pin rust version
+      RUST_STABLE: stable
+    steps:
+      - checkout
+      - run:
+          name: Install Rust
+          command: |
+            curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs -o rustup.sh
+            chmod +x rustup.sh
+            ./rustup.sh -y --default-toolchain $RUST_STABLE
+            source "$HOME"/.cargo/env
+      # Only run Tokio tests
+      - run: cargo test --all-features -p tokio
+
+workflows:
+  ci:
+    jobs:
+      - test-arm
@@ -1,6 +1,10 @@
+only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
+auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
 freebsd_instance:
-  image: freebsd-12-2-release-amd64
+  image_family: freebsd-12-4
 env:
+  RUST_STABLE: stable
+  RUST_NIGHTLY: nightly-2022-10-25
  RUSTFLAGS: -D warnings

 # Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the
@@ -12,7 +16,7 @@ task:
  setup_script:
    - pkg install -y bash curl
    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain stable
+    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
    - . $HOME/.cargo/env
    - |
      echo "~~~~ rustc --version ~~~~"
@@ -24,12 +28,12 @@ task:
 task:
  name: FreeBSD docs
  env:
-    RUSTFLAGS: --cfg docsrs
-    RUSTDOCFLAGS: --cfg docsrs -Dwarnings
+    RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
+    RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
  setup_script:
    - pkg install -y bash curl
    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain nightly-2021-11-23
+    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_NIGHTLY
    - . $HOME/.cargo/env
    - |
      echo "~~~~ rustc --version ~~~~"
@@ -43,7 +47,7 @@ task:
  setup_script:
    - pkg install -y bash curl
    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain stable
+    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
    - . $HOME/.cargo/env
    - rustup target add i686-unknown-freebsd
    - |
@@ -1 +1 @@
-msrv = "1.46"
+msrv = "1.49"
@@ -0,0 +1,4 @@
+contact_links:
+  - name: Question
+    url: https://github.com/tokio-rs/tokio/discussions
+    about: Questions about Tokio should be posted as a GitHub discussion.
@@ -1,16 +0,0 @@
---
-name: Question
-about: Please use the discussions tab for questions
-title: ''
-labels: ''
-assignees: ''
-
---
-
-Please post your question as a discussion here:
-https://github.com/tokio-rs/tokio/discussions
-
-
-You may also be able to find help here:
-https://discord.gg/tokio
-https://users.rust-lang.org/
@@ -9,14 +9,22 @@ on:
  schedule:
    - cron: '0 2 * * *' # run at 2 AM UTC

+permissions:
+  contents: read
+
 jobs:
  security-audit:
+    permissions:
+      checks: write  # for rustsec/audit-check to create check
+      contents: read  # for actions/checkout to fetch code
+      issues: write  # for rustsec/audit-check to create issues
    runs-on: ubuntu-latest
    if: "!contains(github.event.head_commit.message, 'ci skip')"
    steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3

      - name: Audit Check
-        uses: actions-rs/audit-check@v1
+        # https://github.com/rustsec/audit-check/issues/2
+        uses: rustsec/audit-check@master
        with:
          token: ${{ secrets.GITHUB_TOKEN }}
@@ -9,8 +9,26 @@ name: CI
 env:
  RUSTFLAGS: -Dwarnings
  RUST_BACKTRACE: 1
-  nightly: nightly-2021-11-23
-  minrust: 1.46
+  # Change to specific Rust release to pin
+  rust_stable: stable
+  rust_nightly: nightly-2022-11-03
+  rust_clippy: 1.65.0
+  # When updating this, also update:
+  # - README.md
+  # - tokio/README.md
+  # - CONTRIBUTING.md
+  # - tokio/Cargo.toml
+  # - tokio-util/Cargo.toml
+  # - tokio-test/Cargo.toml
+  # - tokio-stream/Cargo.toml
+  rust_min: 1.49.0
+
+defaults:
+  run:
+    shell: bash
+
+permissions:
+  contents: read

 jobs:
  # Depends on all action sthat are required for a "successful" CI run.
@@ -19,18 +37,28 @@ jobs:
    runs-on: ubuntu-latest
    needs:
      - test
-      - test-unstable
      - test-parking_lot
+      - valgrind
+      - test-unstable
      - miri
-      - cross
+      - asan
+      - semver
+      - cross-check
+      - cross-test
+      - no-atomic-u64
      - features
      - minrust
+      - minimal-versions
      - fmt
      - clippy
      - docs
-      - valgrind
      - loom-compile
      - check-readme
+      - test-hyper
+      - x86_64-fortanix-unknown-sgx
+      - wasm32-unknown-unknown
+      - wasm32-wasi
+      - check-external-types
    steps:
      - run: exit 0

@@ -44,12 +72,16 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
      - name: Install Rust
        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
      - name: Install cargo-hack
-        run: cargo install cargo-hack
+        uses: taiki-e/install-action@cargo-hack

      # Run `tokio` with `full` features. This excludes testing utilities which
      # can alter the runtime behavior of Tokio.
@@ -89,10 +121,12 @@ jobs:
    name: compile tests with parking lot send guards
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
      - name: Enable parking_lot send_guard feature
        # Inserts the line "plsend = ["parking_lot/send_guard"]" right after [features]
        run: sed -i '/\[features\]/a plsend = ["parking_lot/send_guard"]' tokio/Cargo.toml
@@ -103,15 +137,15 @@ jobs:
    name: valgrind
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2

      - name: Install Valgrind
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y valgrind
+        uses: taiki-e/install-action@valgrind

      # Compile tests
      - name: cargo build test-mem
@@ -141,99 +175,184 @@ jobs:
          - ubuntu-latest
          - macos-latest
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
-
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
      # Run `tokio` with "unstable" cfg flag.
      - name: test tokio full --cfg unstable
        run: cargo test --all-features
        working-directory: tokio
        env:
          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+          # in order to run doctests for unstable features, we must also pass
+          # the unstable cfg to RustDoc
+          RUSTDOCFLAGS: --cfg tokio_unstable

  miri:
    name: miri
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
        with:
-          toolchain: ${{ env.nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: Install Miri
-        run: |
-          set -e
-          rustup component add miri
-          cargo miri setup
-          rm -rf tokio/tests
-
+          toolchain: ${{ env.rust_nightly }}
+          components: miri
+      - uses: Swatinem/rust-cache@v2
      - name: miri
-        run: cargo miri test --features rt,rt-multi-thread,sync task
-        working-directory: tokio
-  san:
-    name: san
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ env.nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: asan
-        run: cargo test --all-features --target x86_64-unknown-linux-gnu --lib -- --test-threads 1
+        # Many of tests in tokio/tests and doctests use #[tokio::test] or
+        # #[tokio::main] that calls epoll_create1 that Miri does not support.
+        run: cargo miri test --features full --lib --no-fail-fast
        working-directory: tokio
        env:
-          RUSTFLAGS: -Z sanitizer=address
-          ASAN_OPTIONS: detect_leaks=0
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields

-  cross:
-    name: cross
+  asan:
+    name: asan
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install llvm
+        # Required to resolve symbols in sanitizer output
+        run: sudo apt-get install -y llvm
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_nightly }}
+      - uses: Swatinem/rust-cache@v2
+      - name: asan
+        run: cargo test --workspace --all-features --target x86_64-unknown-linux-gnu --tests -- --test-threads 1
+        env:
+          RUSTFLAGS: -Z sanitizer=address
+          # Ignore `trybuild` errors as they are irrelevant and flaky on nightly
+          TRYBUILD: overwrite
+
+  semver:
+    name: semver
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+      - name: Install cargo-semver-checks
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-semver-checks
+      - name: Check semver compatibility
+        run: |
+          cargo semver-checks check-release --release-type minor
+
+  cross-check:
+    name: cross-check
    runs-on: ubuntu-latest
    strategy:
      matrix:
        target:
-          - i686-unknown-linux-gnu
          - powerpc-unknown-linux-gnu
          - powerpc64-unknown-linux-gnu
          - mips-unknown-linux-gnu
          - arm-linux-androideabi
+          - mipsel-unknown-linux-musl
    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
        with:
-          toolchain: stable
+          toolchain: ${{ env.rust_stable }}
          target: ${{ matrix.target }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - uses: actions-rs/cargo@v1
+      - name: Install cross
+        uses: taiki-e/install-action@cross
+      - run: cross check --workspace --all-features --target ${{ matrix.target }}
+        env:
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+
+  cross-test:
+    name: cross-test
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+          - target: i686-unknown-linux-gnu
+          - target: arm-unknown-linux-gnueabihf
+          - target: armv7-unknown-linux-gnueabihf
+          - target: aarch64-unknown-linux-gnu
+
+          # Run a platform without AtomicU64 and no const Mutex::new
+          - target: arm-unknown-linux-gnueabihf
+            rustflags: --cfg tokio_no_const_mutex_new
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust stable
+        uses: dtolnay/rust-toolchain@master
        with:
-          use-cross: true
-          command: check
-          args: --workspace --target ${{ matrix.target }}
+          toolchain: ${{ env.rust_stable }}
+          target: ${{ matrix.target }}
+      - name: Install cross
+        uses: taiki-e/install-action@cross
+      # First run with all features (including parking_lot)
+      - run: cross test -p tokio --all-features --target ${{ matrix.target }} --tests
+        env:
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings --cfg tokio_no_ipv6 ${{ matrix.rustflags }}
+      # Now run without parking_lot
+      - name: Remove `parking_lot` from `full` feature
+        run: sed -i '0,/parking_lot/{/parking_lot/d;}' tokio/Cargo.toml
+      # The `tokio_no_parking_lot` cfg is here to ensure the `sed` above does not silently break.
+      - run: cross test -p tokio --features full,test-util --target ${{ matrix.target }} --tests
+        env:
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings --cfg tokio_no_ipv6 --cfg tokio_no_parking_lot ${{ matrix.rustflags }}
+
+  # See https://github.com/tokio-rs/tokio/issues/5187
+  no-atomic-u64:
+    name: Test i686-unknown-linux-gnu without AtomicU64
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_nightly }}
+          components: rust-src
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
+      # Install linker and libraries for i686-unknown-linux-gnu
+      - uses: taiki-e/setup-cross-toolchain-action@v1
+        with:
+          target: i686-unknown-linux-gnu
+      - run: cargo test -Zbuild-std --target target-specs/i686-unknown-linux-gnu.json -p tokio --all-features
+        env:
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings --cfg tokio_no_atomic_u64
+      # https://github.com/tokio-rs/tokio/pull/5356
+      # https://github.com/tokio-rs/tokio/issues/5373
+      - run: cargo hack build -p tokio --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
+        env:
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings --cfg tokio_no_atomic_u64 --cfg tokio_no_const_mutex_new
+      - run: cargo hack build -p tokio --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
+        env:
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings --cfg tokio_no_atomic_u64

  features:
    name: features
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
        with:
-          toolchain: ${{ env.nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+          toolchain: ${{ env.rust_nightly }}
+          target: ${{ matrix.target }}
+      - uses: Swatinem/rust-cache@v2
      - name: Install cargo-hack
-        run: cargo install cargo-hack
-
-      - name: check --each-feature
-        run: cargo hack check --all --each-feature -Z avoid-dev-deps
-
+        uses: taiki-e/install-action@cargo-hack
+      - name: check --feature-powerset
+        run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
      # Try with unstable feature flags
-      - name: check --each-feature --unstable
-        run: cargo hack check --all --each-feature -Z avoid-dev-deps
+      - name: check --feature-powerset --unstable
+        run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
        env:
          RUSTFLAGS: --cfg tokio_unstable -Dwarnings

@@ -241,27 +360,42 @@ jobs:
    name: minrust
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_min }}
+        uses: dtolnay/rust-toolchain@master
        with:
-          toolchain: ${{ env.minrust }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: "test --workspace --all-features"
+          toolchain: ${{ env.rust_min }}
+      - uses: Swatinem/rust-cache@v2
+      # First compile just the main tokio crate with minrust and newest version
+      # of all dependencies, then pin once_cell and compile the rest of the
+      # crates with the pinned once_cell version.
+      #
+      # This is necessary because tokio-util transitively depends on once_cell,
+      # which is not compatible with the current minrust after the 1.15.0
+      # release.
+      - name: "check -p tokio --all-features"
+        run: cargo check -p tokio --all-features
+        env:
+          RUSTFLAGS: "" # remove -Dwarnings
+      - name: "pin once_cell version"
+        run: cargo update -p once_cell --precise 1.14.0
+      - name: "check --workspace --all-features"
        run: cargo check --workspace --all-features
+        env:
+          RUSTFLAGS: "" # remove -Dwarnings

  minimal-versions:
    name: minimal-versions
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
        with:
-          toolchain: ${{ env.nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+          toolchain: ${{ env.rust_nightly }}
+      - uses: Swatinem/rust-cache@v2
      - name: Install cargo-hack
-        run: cargo install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
      - name: "check --all-features -Z minimal-versions"
        run: |
          # Remove dev-dependencies from Cargo.toml to prevent the next `cargo update`
@@ -285,11 +419,13 @@ jobs:
    name: fmt
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
-
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+          components: rustfmt
+      - uses: Swatinem/rust-cache@v2
      # Check fmt
      - name: "rustfmt --check"
        # Workaround for rust-lang/cargo#7732
@@ -303,13 +439,13 @@ jobs:
    name: clippy
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update 1.57 && rustup default 1.57
-      - uses: Swatinem/rust-cache@v1
-      - name: Install clippy
-        run: rustup component add clippy
-
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_clippy }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_clippy }}
+          components: clippy
+      - uses: Swatinem/rust-cache@v2
      # Run clippy
      - name: "clippy --all"
        run: cargo clippy --all --tests --all-features
@@ -318,26 +454,28 @@ jobs:
    name: docs
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
        with:
-          toolchain: ${{ env.nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+          toolchain: ${{ env.rust_nightly }}
+      - uses: Swatinem/rust-cache@v2
      - name: "doc --lib --all-features"
        run: cargo doc --lib --no-deps --all-features --document-private-items
        env:
-          RUSTFLAGS: --cfg docsrs
-          RUSTDOCFLAGS: --cfg docsrs -Dwarnings
+          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
+          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings

  loom-compile:
    name: build loom tests
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
      - name: build --cfg loom
        run: cargo test --no-run --lib --features full
        working-directory: tokio
@@ -348,10 +486,146 @@ jobs:
    name: Check README
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
      - name: Verify that both READMEs are identical
        run: diff README.md tokio/README.md

      - name: Verify that Tokio version is up to date in README
        working-directory: tokio
        run: grep -q "$(sed '/^version = /!d' Cargo.toml | head -n1)" README.md
+
+  test-hyper:
+    name: Test hyper
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os:
+          - windows-latest
+          - ubuntu-latest
+          - macos-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
+      - name: Test hyper
+        run: |
+          set -x
+          git clone https://github.com/hyperium/hyper.git
+          cd hyper
+          # checkout the latest release because HEAD maybe contains breakage.
+          tag=$(git describe --abbrev=0 --tags)
+          git checkout "${tag}"
+          echo '[workspace]' >>Cargo.toml
+          echo '[patch.crates-io]' >>Cargo.toml
+          echo 'tokio = { path = "../tokio" }' >>Cargo.toml
+          echo 'tokio-util = { path = "../tokio-util" }' >>Cargo.toml
+          echo 'tokio-stream = { path = "../tokio-stream" }' >>Cargo.toml
+          echo 'tokio-test = { path = "../tokio-test" }' >>Cargo.toml
+          git diff
+          cargo test --features full
+
+  x86_64-fortanix-unknown-sgx:
+    name: build tokio for x86_64-fortanix-unknown-sgx
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_nightly }}
+          target: x86_64-fortanix-unknown-sgx
+      - uses: Swatinem/rust-cache@v2
+      # NOTE: Currently the only test we can run is to build tokio with rt and sync features.
+      - name: build tokio
+        run: cargo build --target x86_64-fortanix-unknown-sgx --features rt,sync
+        working-directory: tokio
+
+  wasm32-unknown-unknown:
+    name: test tokio for wasm32-unknown-unknown
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
+      - name: Install wasm-pack
+        run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
+      - name: test tokio
+        run: wasm-pack test --node -- --features "macros sync"
+        working-directory: tokio
+
+  wasm32-wasi:
+    name: wasm32-wasi
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
+
+      # Install dependencies
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
+
+      - name: Install wasm32-wasi target
+        run: rustup target add wasm32-wasi
+
+      - name: Install wasmtime
+        uses: taiki-e/install-action@wasmtime
+
+      - name: Install cargo-wasi
+        run: cargo install cargo-wasi
+
+      - name: WASI test tokio full
+        run: cargo test -p tokio --target wasm32-wasi --features full
+        env:
+          CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+
+      - name: WASI test tokio-util full
+        run: cargo test -p tokio-util --target wasm32-wasi --features full
+        env:
+          CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+
+      - name: WASI test tokio-stream
+        run: cargo test -p tokio-stream --target wasm32-wasi --features time,net,io-util,sync
+        env:
+          CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+
+      - name: test tests-integration --features wasi-rt
+        # TODO: this should become: `cargo hack wasi test --each-feature`
+        run: cargo wasi test --test rt_yield --features wasi-rt
+        working-directory: tests-integration
+
+  check-external-types:
+    name: check-external-types
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os:
+          - windows-latest
+          - ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust nightly-2022-11-16
+        uses: dtolnay/rust-toolchain@master
+        with:
+          # `check-external-types` requires a specific Rust nightly version. See
+          # the README for details: https://github.com/awslabs/cargo-check-external-types
+          toolchain: nightly-2022-11-16
+      - uses: Swatinem/rust-cache@v2
+      - name: check-external-types
+        run: |
+          set -x
+          cargo install cargo-check-external-types --locked --version 0.1.6
+          cargo check-external-types --all-features --config external-types.toml
+        working-directory: tokio
@@ -4,9 +4,16 @@ on:

 # See .github/labeler.yml file

+permissions:
+  contents: read
+
 jobs:
  triage:
+    permissions:
+      contents: read  # for actions/labeler to determine modified files
+      pull-requests: write  # for actions/labeler to add labels to PRs
    runs-on: ubuntu-latest
+    if: github.repository_owner == 'tokio-rs'
    steps:
    - uses: actions/labeler@v3
      with:
@@ -10,12 +10,17 @@ name: Loom
 env:
  RUSTFLAGS: -Dwarnings
  RUST_BACKTRACE: 1
+  # Change to specific Rust release to pin
+  rust_stable: stable
+
+permissions:
+  contents: read

 jobs:
  loom:
    name: loom
    # base_ref is null when it's not a pull request
-    if: contains(github.event.pull_request.labels.*.name, 'R-loom') || (github.base_ref == null)
+    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom') || (github.base_ref == null))
    runs-on: ubuntu-latest
    strategy:
      matrix:
@@ -27,14 +32,17 @@ jobs:
          - loom_pool::group_d
          - time::driver
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
      - name: loom ${{ matrix.scope }}
        run: cargo test --lib --release --features full -- --nocapture $SCOPE
        working-directory: tokio
        env:
          RUSTFLAGS: --cfg loom --cfg tokio_unstable -Dwarnings
          LOOM_MAX_PREEMPTIONS: 2
+          LOOM_MAX_BRANCHES: 10000
          SCOPE: ${{ matrix.scope }}
@@ -8,25 +8,21 @@ on:
    paths:
      - '**/Cargo.toml'

+permissions:
+  contents: read
+
 jobs:
  security-audit:
    runs-on: ubuntu-latest
    if: "!contains(github.event.head_commit.message, 'ci skip')"
    steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3

      - name: Install cargo-audit
-        uses: actions-rs/cargo@v1
-        with:
-          command: install
-          args: cargo-audit
+        run: cargo install cargo-audit

      - name: Generate lockfile
-        uses: actions-rs/cargo@v1
-        with:
-          command: generate-lockfile
+        run: cargo generate-lockfile

      - name: Audit dependencies
-        uses: actions-rs/cargo@v1
-        with:
-          command: audit
+        run: cargo audit
@@ -5,8 +5,17 @@ on:
    branches:
      - master

+env:
+  RUSTFLAGS: -Dwarnings
+  RUST_BACKTRACE: 1
+  # Change to specific Rust release to pin
+  rust_stable: stable
+
+permissions:
+  contents: read
+
 jobs:
-  stess-test:
+  stress-test:
    name: Stress Test
    runs-on: ubuntu-latest
    strategy:
@@ -14,14 +23,14 @@ jobs:
        stress-test:
          - simple_echo_tcp
    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust
-        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
      - name: Install Valgrind
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y valgrind
+        uses: taiki-e/install-action@valgrind

      # Compiles each of the stress test examples.
      - name: Compile stress test examples
@@ -29,4 +38,4 @@ jobs:

      # Runs each of the examples using Valgrind. Detects leaks and displays them.
      - name: Run valgrind
-        run: valgrind  --leak-check=full --show-leak-kinds=all ./target/release/examples/${{ matrix.stress-test }}
+        run: valgrind --error-exitcode=1 --leak-check=full --show-leak-kinds=all ./target/release/examples/${{ matrix.stress-test }}
@@ -1,2 +1,4 @@
 target
 Cargo.lock
+
+.cargo/config.toml
@@ -131,12 +131,30 @@ cargo check --all-features
 cargo test --all-features
 ```

+Clippy must be run using the MSRV, so Tokio can avoid having to `#[allow]` new
+lints whose fixes would be incompatible with the current MSRV:
+
+<!--
+When updating this, also update:
+- .github/workflows/ci.yml
+- README.md
+- tokio/README.md
+- tokio/Cargo.toml
+- tokio-util/Cargo.toml
+- tokio-test/Cargo.toml
+- tokio-stream/Cargo.toml
+-->
+
+```
+cargo +1.49.0 clippy --all --tests --all-features
+```
+
 When building documentation normally, the markers that list the features
 required for various parts of Tokio are missing. To build the documentation
 correctly, use this command:

 ```
-RUSTDOCFLAGS="--cfg docsrs" cargo +nightly doc --all-features
+RUSTDOCFLAGS="--cfg docsrs" RUSTFLAGS="--cfg docsrs" cargo +nightly doc --all-features
 ```

 To build documentation including Tokio's unstable features, it is necessary to
@@ -144,15 +162,9 @@ pass `--cfg tokio_unstable` to both RustDoc *and* rustc. To build the
 documentation for unstable features, use this command:

 ```
-RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg tokio_unstable" cargo +nightly doc --all-features
+RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg docsrs --cfg tokio_unstable" cargo +nightly doc --all-features
 ```

-There is currently a [bug in cargo] that means documentation cannot be built
-from the root of the workspace. If you `cd` into the `tokio` subdirectory the
-command shown above will work.
-
-[bug in cargo]: https://github.com/rust-lang/cargo/issues/9274
-
 The `cargo fmt` command does not work on the Tokio codebase. You can use the
 command below instead:

@@ -173,6 +185,12 @@ LOOM_MAX_PREEMPTIONS=1 RUSTFLAGS="--cfg loom" \
    cargo test --lib --release --features full -- --test-threads=1 --nocapture
 ```

+You can run miri tests with
+```
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" \
+    cargo +nightly miri test --features full --lib
+```
+
 ### Tests

 If the change being proposed alters code (as opposed to only documentation for
@@ -191,6 +209,31 @@ utilities available to use in tests, no matter the crate being tested.
 The best strategy for writing a new integration test is to look at existing
 integration tests in the crate and follow the style.

+#### Fuzz tests
+
+Some of our crates include a set of fuzz tests, this will be marked by a
+directory `fuzz`. It is a good idea to run fuzz tests after each change.
+To get started with fuzz testing you'll need to install 
+[cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz).
+
+`cargo install cargo-fuzz`
+
+To list the available fuzzing harnesses you can run;
+
+```bash
+$ cd tokio
+$ cargo fuzz list
+fuzz_linked_list
+````
+
+Running a fuzz test is as simple as;
+
+`cargo fuzz run fuzz_linked_list`
+
+**NOTE**: Keep in mind that by default when running a fuzz test the fuzz
+harness will run forever and will only exit if you `ctrl-c` or it finds
+a bug.
+
 #### Documentation tests

 Ideally, every API has at least one [documentation test] that demonstrates how to
@@ -538,7 +581,7 @@ Tokio ≥1.0.0 comes with LTS guarantees:

 The goal of these guarantees is to provide stability to the ecosystem.

-## Mininum Supported Rust Version (MSRV)
+## Minimum Supported Rust Version (MSRV)

 * All Tokio ≥1.0.0 releases will support at least a 6-month old Rust
   compiler release.
@@ -0,0 +1,5 @@
+[build.env]
+passthrough = [
+    "RUSTFLAGS",
+    "RUST_BACKTRACE",
+]
@@ -1,4 +1,4 @@
-Copyright (c) 2021 Tokio Contributors
+Copyright (c) 2023 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:

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

@@ -161,11 +161,33 @@ several other libraries, including:
 [`mio`]: https://github.com/tokio-rs/mio
 [`bytes`]: https://github.com/tokio-rs/bytes

+## Changelog
+
+The Tokio repository contains multiple crates. Each crate has its own changelog.
+
+ * `tokio` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio/CHANGELOG.md)
+ * `tokio-util` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-util/CHANGELOG.md)
+ * `tokio-stream` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-stream/CHANGELOG.md)
+ * `tokio-macros` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-macros/CHANGELOG.md)
+ * `tokio-test` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-test/CHANGELOG.md)
+
 ## Supported Rust Versions

-Tokio is built against the latest stable release. The minimum supported version
-is 1.46.  The current Tokio version is not guaranteed to build on Rust versions
-earlier than the minimum supported version.
+<!--
+When updating this, also update:
+- .github/workflows/ci.yml
+- CONTRIBUTING.md
+- README.md
+- tokio/README.md
+- tokio/Cargo.toml
+- tokio-util/Cargo.toml
+- tokio-test/Cargo.toml
+- tokio-stream/Cargo.toml
+-->
+
+Tokio will keep a rolling MSRV (minimum supported rust version) policy of **at
+least** 6 months. When increasing the MSRV, the new Rust version must have been
+released at least six months ago. The current MSRV is 1.49.0.

 ## Release schedule

@@ -180,18 +202,18 @@ warrants a patch release with a fix for the bug, it will be backported and
 released as a new patch release for each LTS minor version. Our current LTS
 releases are:

- * `1.8.x` - LTS release until February 2022.
- * `1.14.x` - LTS release until June 2022.
+ * `1.18.x` - LTS release until June 2023
+ * `1.20.x` - LTS release until September 2023.

-Each LTS release will continue to receive backported fixes for at least half a
-year. If you wish to use a fixed minor release in your project, we recommend
-that you use an LTS release.
+Each LTS release will continue to receive backported fixes for at least a year.
+If you wish to use a fixed minor release in your project, we recommend that you
+use an LTS release.

 To use a fixed minor version, you can specify the version with a tilde. For
-example, to specify that you wish to use the newest `1.8.x` patch release, you
+example, to specify that you wish to use the newest `1.18.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.8", features = [...] }
+tokio = { version = "~1.18", features = [...] }
 ```

 ## License
@@ -1,13 +1,13 @@
 ## Report a security issue

-The Tokio project team welcomes security reports and is committed to providing prompt attention to security issues. Security issues should be reported privately via [[email protected]](mailto:[email protected]). Security issues should not be reported via the public Github Issue tracker.
+The Tokio project team welcomes security reports and is committed to providing prompt attention to security issues. Security issues should be reported privately via [[email protected]](mailto:[email protected]). Security issues should not be reported via the public GitHub Issue tracker.

 ## Vulnerability coordination

-Remediation of security vulnerabilities is prioritized by the project team. The project team coordinates remediation with third-party project stakeholders via [Github Security Advisories](https://help.github.com/en/github/managing-security-vulnerabilities/about-github-security-advisories). Third-party stakeholders may include the reporter of the issue, affected direct or indirect users of Tokio, and maintainers of upstream dependencies if applicable.
+Remediation of security vulnerabilities is prioritized by the project team. The project team coordinates remediation with third-party project stakeholders via [GitHub Security Advisories](https://help.github.com/en/github/managing-security-vulnerabilities/about-github-security-advisories). Third-party stakeholders may include the reporter of the issue, affected direct or indirect users of Tokio, and maintainers of upstream dependencies if applicable.

-Downstream project maintainers and Tokio users can request participation in coordination of applicable security issues by sending your contact email address, Github username(s) and any other salient information to [[email protected]](mailto:[email protected]). Participation in security issue coordination processes is at the discretion of the Tokio team.
+Downstream project maintainers and Tokio users can request participation in coordination of applicable security issues by sending your contact email address, GitHub username(s) and any other salient information to [[email protected]](mailto:[email protected]). Participation in security issue coordination processes is at the discretion of the Tokio team.

 ## Security advisories

-The project team is committed to transparency in the security issue disclosure process. The Tokio team announces security issues via [project Github Release notes](https://github.com/tokio-rs/tokio/releases) and the [RustSec advisory database](https://github.com/RustSec/advisory-db) (i.e. `cargo-audit`).
+The project team is committed to transparency in the security issue disclosure process. The Tokio team announces security issues via [project GitHub Release notes](https://github.com/tokio-rs/tokio/releases) and the [RustSec advisory database](https://github.com/RustSec/advisory-db) (i.e. `cargo-audit`).
@@ -4,9 +4,14 @@ version = "0.0.0"
 publish = false
 edition = "2018"

+[features]
+test-util = ["tokio/test-util"]
+
 [dependencies]
 tokio = { version = "1.5.0", path = "../tokio", features = ["full"] }
 bencher = "0.1.5"
+rand = "0.8"
+rand_chacha = "0.3"

 [dev-dependencies]
 tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
@@ -25,6 +30,16 @@ name = "sync_mpsc"
 path = "sync_mpsc.rs"
 harness = false

+[[bench]]
+name = "sync_mpsc_oneshot"
+path = "sync_mpsc_oneshot.rs"
+harness = false
+
+[[bench]]
+name = "sync_watch"
+path = "sync_watch.rs"
+harness = false
+
 [[bench]]
 name = "rt_multi_threaded"
 path = "rt_multi_threaded.rs"
@@ -50,3 +65,13 @@ harness = false
 name = "fs"
 path = "fs.rs"
 harness = false
+
+[[bench]]
+name = "copy"
+path = "copy.rs"
+harness = false
+
+[[bench]]
+name = "time_now"
+path = "time_now.rs"
+harness = false
@@ -0,0 +1,238 @@
+use bencher::{benchmark_group, benchmark_main, Bencher};
+
+use rand::{Rng, SeedableRng};
+use rand_chacha::ChaCha20Rng;
+
+use tokio::io::{copy, repeat, AsyncRead, AsyncReadExt, AsyncWrite};
+use tokio::time::{interval, Interval, MissedTickBehavior};
+
+use std::task::Poll;
+use std::time::Duration;
+
+const KILO: usize = 1024;
+
+// Tunable parameters if you want to change this benchmark. If reader and writer
+// are matched in kilobytes per second, then this only exposes buffering to the
+// benchmark.
+const RNG_SEED: u64 = 0;
+// How much data to copy in a single benchmark run
+const SOURCE_SIZE: u64 = 256 * KILO as u64;
+// Read side provides CHUNK_SIZE every READ_SERVICE_PERIOD. If it's not called
+// frequently, it'll burst to catch up (representing OS buffers draining)
+const CHUNK_SIZE: usize = 2 * KILO;
+const READ_SERVICE_PERIOD: Duration = Duration::from_millis(1);
+// Write side buffers up to WRITE_BUFFER, and flushes to disk every
+// WRITE_SERVICE_PERIOD.
+const WRITE_BUFFER: usize = 40 * KILO;
+const WRITE_SERVICE_PERIOD: Duration = Duration::from_millis(20);
+// How likely you are to have to wait for previously written data to be flushed
+// because another writer claimed the buffer space
+const PROBABILITY_FLUSH_WAIT: f64 = 0.1;
+
+/// A slow writer that aims to simulate HDD behaviour under heavy load.
+///
+/// There is a limited buffer, which is fully drained on the next write after
+/// a time limit is reached. Flush waits for the time limit to be reached
+/// and then drains the buffer.
+///
+/// At random, the HDD will stall writers while it flushes out all buffers. If
+/// this happens to you, you will be unable to write until the next time the
+/// buffer is drained.
+struct SlowHddWriter {
+    service_intervals: Interval,
+    blocking_rng: ChaCha20Rng,
+    buffer_size: usize,
+    buffer_used: usize,
+}
+
+impl SlowHddWriter {
+    fn new(service_interval: Duration, buffer_size: usize) -> Self {
+        let blocking_rng = ChaCha20Rng::seed_from_u64(RNG_SEED);
+        let mut service_intervals = interval(service_interval);
+        service_intervals.set_missed_tick_behavior(MissedTickBehavior::Delay);
+        Self {
+            service_intervals,
+            blocking_rng,
+            buffer_size,
+            buffer_used: 0,
+        }
+    }
+
+    fn service_write(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        // If we hit a service interval, the buffer can be cleared
+        let res = self.service_intervals.poll_tick(cx).map(|_| Ok(()));
+        if let Poll::Ready(_) = res {
+            self.buffer_used = 0;
+        }
+        res
+    }
+
+    fn write_bytes(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        writeable: usize,
+    ) -> std::task::Poll<Result<usize, std::io::Error>> {
+        let service_res = self.as_mut().service_write(cx);
+
+        if service_res.is_pending() && self.blocking_rng.gen_bool(PROBABILITY_FLUSH_WAIT) {
+            return Poll::Pending;
+        }
+        let available = self.buffer_size - self.buffer_used;
+
+        if available == 0 {
+            assert!(service_res.is_pending());
+            Poll::Pending
+        } else {
+            let written = available.min(writeable);
+            self.buffer_used += written;
+            Poll::Ready(Ok(written))
+        }
+    }
+}
+
+impl Unpin for SlowHddWriter {}
+
+impl AsyncWrite for SlowHddWriter {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<Result<usize, std::io::Error>> {
+        self.write_bytes(cx, buf.len())
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        self.service_write(cx)
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        self.service_write(cx)
+    }
+
+    fn poll_write_vectored(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> std::task::Poll<Result<usize, std::io::Error>> {
+        let writeable = bufs.into_iter().fold(0, |acc, buf| acc + buf.len());
+        self.write_bytes(cx, writeable)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        true
+    }
+}
+
+/// A reader that limits the maximum chunk it'll give you back
+///
+/// Simulates something reading from a slow link - you get one chunk per call,
+/// and you are offered chunks on a schedule
+struct ChunkReader {
+    data: Vec<u8>,
+    service_intervals: Interval,
+}
+
+impl ChunkReader {
+    fn new(chunk_size: usize, service_interval: Duration) -> Self {
+        let mut service_intervals = interval(service_interval);
+        service_intervals.set_missed_tick_behavior(MissedTickBehavior::Burst);
+        let data: Vec<u8> = std::iter::repeat(0).take(chunk_size).collect();
+        Self {
+            data,
+            service_intervals,
+        }
+    }
+}
+
+impl AsyncRead for ChunkReader {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        if self.service_intervals.poll_tick(cx).is_pending() {
+            return Poll::Pending;
+        }
+        buf.put_slice(&self.data[..buf.remaining().min(self.data.len())]);
+        Poll::Ready(Ok(()))
+    }
+}
+
+fn rt() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .unwrap()
+}
+
+fn copy_mem_to_mem(b: &mut Bencher) {
+    let rt = rt();
+
+    b.iter(|| {
+        let task = || async {
+            let mut source = repeat(0).take(SOURCE_SIZE);
+            let mut dest = Vec::new();
+            copy(&mut source, &mut dest).await.unwrap();
+        };
+
+        rt.block_on(task());
+    })
+}
+
+fn copy_mem_to_slow_hdd(b: &mut Bencher) {
+    let rt = rt();
+
+    b.iter(|| {
+        let task = || async {
+            let mut source = repeat(0).take(SOURCE_SIZE);
+            let mut dest = SlowHddWriter::new(WRITE_SERVICE_PERIOD, WRITE_BUFFER);
+            copy(&mut source, &mut dest).await.unwrap();
+        };
+
+        rt.block_on(task());
+    })
+}
+
+fn copy_chunk_to_mem(b: &mut Bencher) {
+    let rt = rt();
+    b.iter(|| {
+        let task = || async {
+            let mut source = ChunkReader::new(CHUNK_SIZE, READ_SERVICE_PERIOD).take(SOURCE_SIZE);
+            let mut dest = Vec::new();
+            copy(&mut source, &mut dest).await.unwrap();
+        };
+
+        rt.block_on(task());
+    })
+}
+
+fn copy_chunk_to_slow_hdd(b: &mut Bencher) {
+    let rt = rt();
+    b.iter(|| {
+        let task = || async {
+            let mut source = ChunkReader::new(CHUNK_SIZE, READ_SERVICE_PERIOD).take(SOURCE_SIZE);
+            let mut dest = SlowHddWriter::new(WRITE_SERVICE_PERIOD, WRITE_BUFFER);
+            copy(&mut source, &mut dest).await.unwrap();
+        };
+
+        rt.block_on(task());
+    })
+}
+
+benchmark_group!(
+    copy_bench,
+    copy_mem_to_mem,
+    copy_mem_to_slow_hdd,
+    copy_chunk_to_mem,
+    copy_chunk_to_slow_hdd,
+);
+benchmark_main!(copy_bench);
@@ -0,0 +1,53 @@
+use bencher::{benchmark_group, benchmark_main, Bencher};
+use tokio::{
+    runtime::Runtime,
+    sync::{mpsc, oneshot},
+};
+
+fn request_reply_current_thread(b: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+
+    request_reply(b, rt);
+}
+
+fn request_reply_multi_threaded(b: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .build()
+        .unwrap();
+
+    request_reply(b, rt);
+}
+
+fn request_reply(b: &mut Bencher, rt: Runtime) {
+    let tx = rt.block_on(async move {
+        let (tx, mut rx) = mpsc::channel::<oneshot::Sender<()>>(10);
+        tokio::spawn(async move {
+            while let Some(reply) = rx.recv().await {
+                reply.send(()).unwrap();
+            }
+        });
+        tx
+    });
+
+    b.iter(|| {
+        let task_tx = tx.clone();
+        rt.block_on(async move {
+            for _ in 0..1_000 {
+                let (o_tx, o_rx) = oneshot::channel();
+                task_tx.send(o_tx).await.unwrap();
+                let _ = o_rx.await;
+            }
+        })
+    });
+}
+
+benchmark_group!(
+    sync_mpsc_oneshot_group,
+    request_reply_current_thread,
+    request_reply_multi_threaded,
+);
+
+benchmark_main!(sync_mpsc_oneshot_group);
@@ -14,7 +14,7 @@ fn read_uncontended(b: &mut Bencher) {
        rt.block_on(async move {
            for _ in 0..6 {
                let read = lock.read().await;
-                black_box(read);
+                let _read = black_box(read);
            }
        })
    });
@@ -28,7 +28,7 @@ fn read_concurrent_uncontended_multi(b: &mut Bencher) {

    async fn task(lock: Arc<RwLock<()>>) {
        let read = lock.read().await;
-        black_box(read);
+        let _read = black_box(read);
    }

    let lock = Arc::new(RwLock::new(()));
@@ -55,7 +55,7 @@ fn read_concurrent_uncontended(b: &mut Bencher) {

    async fn task(lock: Arc<RwLock<()>>) {
        let read = lock.read().await;
-        black_box(read);
+        let _read = black_box(read);
    }

    let lock = Arc::new(RwLock::new(()));
@@ -82,7 +82,7 @@ fn read_concurrent_contended_multi(b: &mut Bencher) {

    async fn task(lock: Arc<RwLock<()>>) {
        let read = lock.read().await;
-        black_box(read);
+        let _read = black_box(read);
    }

    let lock = Arc::new(RwLock::new(()));
@@ -110,7 +110,7 @@ fn read_concurrent_contended(b: &mut Bencher) {

    async fn task(lock: Arc<RwLock<()>>) {
        let read = lock.read().await;
-        black_box(read);
+        let _read = black_box(read);
    }

    let lock = Arc::new(RwLock::new(()));
@@ -0,0 +1,64 @@
+use bencher::{black_box, Bencher};
+use rand::prelude::*;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use tokio::sync::{watch, Notify};
+
+fn rt() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(6)
+        .build()
+        .unwrap()
+}
+
+fn do_work(rng: &mut impl RngCore) -> u32 {
+    use std::fmt::Write;
+    let mut message = String::new();
+    for i in 1..=10 {
+        let _ = write!(&mut message, " {i}={}", rng.gen::<f64>());
+    }
+    message
+        .as_bytes()
+        .iter()
+        .map(|&c| c as u32)
+        .fold(0, u32::wrapping_add)
+}
+
+fn contention_resubscribe(b: &mut Bencher) {
+    const NTASK: u64 = 1000;
+
+    let rt = rt();
+    let (snd, rcv) = watch::channel(0i32);
+    let wg = Arc::new((AtomicU64::new(0), Notify::new()));
+    for n in 0..NTASK {
+        let mut rcv = rcv.clone();
+        let wg = wg.clone();
+        let mut rng = rand::rngs::StdRng::seed_from_u64(n);
+        rt.spawn(async move {
+            while rcv.changed().await.is_ok() {
+                let _ = *rcv.borrow(); // contend on rwlock
+                let r = do_work(&mut rng);
+                let _ = black_box(r);
+                if wg.0.fetch_sub(1, Ordering::Release) == 1 {
+                    wg.1.notify_one();
+                }
+            }
+        });
+    }
+
+    b.iter(|| {
+        rt.block_on(async {
+            for _ in 0..100 {
+                assert_eq!(wg.0.fetch_add(NTASK, Ordering::Relaxed), 0);
+                let _ = snd.send(black_box(42));
+                while wg.0.load(Ordering::Acquire) > 0 {
+                    wg.1.notified().await;
+                }
+            }
+        });
+    });
+}
+
+bencher::benchmark_group!(contention, contention_resubscribe);
+
+bencher::benchmark_main!(contention);
@@ -0,0 +1,25 @@
+//! Benchmark spawning a task onto the basic and threaded Tokio executors.
+//! This essentially measure the time to enqueue a task in the local and remote
+//! case.
+
+#[macro_use]
+extern crate bencher;
+
+use bencher::{black_box, Bencher};
+
+fn time_now_current_thread(bench: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    bench.iter(|| {
+        rt.block_on(async {
+            black_box(tokio::time::Instant::now());
+        })
+    })
+}
+
+bencher::benchmark_group!(time_now, time_now_current_thread,);
+
+bencher::benchmark_main!(time_now);
@@ -12,7 +12,7 @@ tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
 tokio-stream = { version = "0.1", path = "../tokio-stream" }

 tracing = "0.1"
-tracing-subscriber = { version = "0.2.7", default-features = false, features = ["fmt", "ansi", "env-filter", "chrono", "tracing-log"] }
+tracing-subscriber = { version = "0.3.1", default-features = false, features = ["fmt", "ansi", "env-filter", "tracing-log"] }
 bytes = "1.0.0"
 futures = { version = "0.3.0", features = ["thread-pool"]}
 http = "0.2"
@@ -24,8 +24,8 @@ httpdate = "1.0"
 once_cell = "1.5.2"
 rand = "0.8.3"

-[target.'cfg(windows)'.dev-dependencies.winapi]
-version = "0.3.8"
+[target.'cfg(windows)'.dev-dependencies.windows-sys]
+version = "0.45"

 [[example]]
 name = "chat"
@@ -75,7 +75,6 @@ path = "tinyhttp.rs"
 name = "custom-executor"
 path = "custom-executor.rs"

-
 [[example]]
 name = "custom-executor-tokio-context"
 path = "custom-executor-tokio-context.rs"
@@ -1,9 +1,7 @@
-//! Hello world server.
-//!
 //! A simple client that opens a TCP stream, writes "hello world\n", and closes
 //! the connection.
 //!
-//! You can test this out by running:
+//! To start a server that this client can talk to on port 6142, you can use this command:
 //!
 //!     ncat -l 6142
 //!
@@ -26,7 +24,7 @@ pub async fn main() -> Result<(), Box<dyn Error>> {
    let mut stream = TcpStream::connect("127.0.0.1:6142").await?;
    println!("created stream");

-    let result = stream.write(b"hello world\n").await;
+    let result = stream.write_all(b"hello world\n").await;
    println!("wrote to stream; success={:?}", result.is_ok());

    Ok(())
@@ -6,7 +6,7 @@ async fn windows_main() -> io::Result<()> {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
    use tokio::time;
-    use winapi::shared::winerror;
+    use windows_sys::Win32::Foundation::ERROR_PIPE_BUSY;

    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-multi-client";
    const N: usize = 10;
@@ -59,7 +59,7 @@ async fn windows_main() -> io::Result<()> {
            let mut client = loop {
                match ClientOptions::new().open(PIPE_NAME) {
                    Ok(client) => break client,
-                    Err(e) if e.raw_os_error() == Some(winerror::ERROR_PIPE_BUSY as i32) => (),
+                    Err(e) if e.raw_os_error() == Some(ERROR_PIPE_BUSY as i32) => (),
                    Err(e) => return Err(e),
                }

@@ -0,0 +1,16 @@
+[build]
+  command = """
+    rustup install nightly --profile minimal && cargo doc --no-deps --all-features
+    """
+  publish = "target/doc"
+
+[build.environment]
+  RUSTDOCFLAGS="""
+    --cfg docsrs \
+    --cfg tokio_unstable \
+    """
+  RUSTFLAGS="--cfg tokio_unstable --cfg docsrs"
+
+[[redirects]]
+  from = "/"
+  to = "/tokio"
@@ -0,0 +1,40 @@
+{
+  "arch": "x86",
+  "cpu": "pentium4",
+  "crt-static-respected": true,
+  "data-layout": "e-m:e-p:32:32-p270:32:32-p271:32:32-p272:64:64-f64:32:64-f80:32-n8:16:32-S128",
+  "dynamic-linking": true,
+  "env": "gnu",
+  "has-rpath": true,
+  "has-thread-local": true,
+  "llvm-target": "i686-unknown-linux-gnu",
+  "max-atomic-width": 32,
+  "os": "linux",
+  "position-independent-executables": true,
+  "pre-link-args": {
+    "gcc": [
+      "-m32"
+    ]
+  },
+  "relro-level": "full",
+  "stack-probes": {
+    "kind": "inline-or-call",
+    "min-llvm-version-for-inline": [
+      16,
+      0,
+      0
+    ]
+  },
+  "supported-sanitizers": [
+    "address"
+  ],
+  "supported-split-debuginfo": [
+    "packed",
+    "unpacked",
+    "off"
+  ],
+  "target-family": [
+    "unix"
+  ],
+  "target-pointer-width": "32"
+}
@@ -1,2 +1,10 @@
 Tests the various combination of feature flags. This is broken out to a separate
 crate to work around limitations with cargo features.
+
+To run all of the tests in this directory, run the following commands:
+```
+cargo test --features full
+cargo test --features rt
+```
+If one of the tests fail, you can pass `TRYBUILD=overwrite` to the `cargo test`
+command that failed to have it regenerate the test output.
@@ -1,4 +1,4 @@
-error: function is never used: `f`
+error: function `f` is never used
 --> $DIR/macros_dead_code.rs:6:10
  |
 6 | async fn f() {}
@@ -1,3 +1,5 @@
+#![deny(duplicate_macro_attributes)]
+
 use tests_build::tokio;

 #[tokio::main]
@@ -33,6 +35,15 @@ async fn test_worker_threads_not_int() {}
 #[tokio::test(flavor = "current_thread", worker_threads = 4)]
 async fn test_worker_threads_and_current_thread() {}

+#[tokio::test(crate = 456)]
+async fn test_crate_not_ident_int() {}
+
+#[tokio::test(crate = "456")]
+async fn test_crate_not_ident_invalid() {}
+
+#[tokio::test(crate = "abc::edf")]
+async fn test_crate_not_ident_path() {}
+
 #[tokio::test]
 #[test]
 async fn test_has_second_test_attr() {}
@@ -1,71 +1,101 @@
 error: the `async` keyword is missing from the function declaration
- --> $DIR/macros_invalid_input.rs:4:1
+ --> $DIR/macros_invalid_input.rs:6:1
  |
-4 | fn main_is_not_async() {}
+6 | fn main_is_not_async() {}
  | ^^

-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`
- --> $DIR/macros_invalid_input.rs:6:15
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
+ --> $DIR/macros_invalid_input.rs:8:15
  |
-6 | #[tokio::main(foo)]
+8 | #[tokio::main(foo)]
  |               ^^^

 error: Must have specified ident
- --> $DIR/macros_invalid_input.rs:9:15
-  |
-9 | #[tokio::main(threadpool::bar)]
-  |               ^^^^^^^^^^^^^^^
+  --> $DIR/macros_invalid_input.rs:11:15
+   |
+11 | #[tokio::main(threadpool::bar)]
+   |               ^^^^^^^^^^^^^^^

 error: the `async` keyword is missing from the function declaration
-  --> $DIR/macros_invalid_input.rs:13:1
+  --> $DIR/macros_invalid_input.rs:15:1
   |
-13 | fn test_is_not_async() {}
+15 | fn test_is_not_async() {}
   | ^^

-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`
-  --> $DIR/macros_invalid_input.rs:15:15
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
+  --> $DIR/macros_invalid_input.rs:17:15
   |
-15 | #[tokio::test(foo)]
+17 | #[tokio::test(foo)]
   |               ^^^

-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`
-  --> $DIR/macros_invalid_input.rs:18:15
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
+  --> $DIR/macros_invalid_input.rs:20:15
   |
-18 | #[tokio::test(foo = 123)]
+20 | #[tokio::test(foo = 123)]
   |               ^^^^^^^^^

 error: Failed to parse value of `flavor` as string.
-  --> $DIR/macros_invalid_input.rs:21:24
+  --> $DIR/macros_invalid_input.rs:23:24
   |
-21 | #[tokio::test(flavor = 123)]
+23 | #[tokio::test(flavor = 123)]
   |                        ^^^

 error: No such runtime flavor `foo`. The runtime flavors are `current_thread` and `multi_thread`.
-  --> $DIR/macros_invalid_input.rs:24:24
+  --> $DIR/macros_invalid_input.rs:26:24
   |
-24 | #[tokio::test(flavor = "foo")]
+26 | #[tokio::test(flavor = "foo")]
   |                        ^^^^^

 error: The `start_paused` option requires the `current_thread` runtime flavor. Use `#[tokio::test(flavor = "current_thread")]`
-  --> $DIR/macros_invalid_input.rs:27:55
+  --> $DIR/macros_invalid_input.rs:29:55
   |
-27 | #[tokio::test(flavor = "multi_thread", start_paused = false)]
+29 | #[tokio::test(flavor = "multi_thread", start_paused = false)]
   |                                                       ^^^^^

 error: Failed to parse value of `worker_threads` as integer.
-  --> $DIR/macros_invalid_input.rs:30:57
+  --> $DIR/macros_invalid_input.rs:32:57
   |
-30 | #[tokio::test(flavor = "multi_thread", worker_threads = "foo")]
+32 | #[tokio::test(flavor = "multi_thread", worker_threads = "foo")]
   |                                                         ^^^^^

 error: The `worker_threads` option requires the `multi_thread` runtime flavor. Use `#[tokio::test(flavor = "multi_thread")]`
-  --> $DIR/macros_invalid_input.rs:33:59
+  --> $DIR/macros_invalid_input.rs:35:59
   |
-33 | #[tokio::test(flavor = "current_thread", worker_threads = 4)]
+35 | #[tokio::test(flavor = "current_thread", worker_threads = 4)]
   |                                                           ^

-error: second test attribute is supplied
-  --> $DIR/macros_invalid_input.rs:37:1
+error: Failed to parse value of `crate` as ident.
+  --> $DIR/macros_invalid_input.rs:38:23
   |
-37 | #[test]
+38 | #[tokio::test(crate = 456)]
+   |                       ^^^
+
+error: Failed to parse value of `crate` as ident: "456"
+  --> $DIR/macros_invalid_input.rs:41:23
+   |
+41 | #[tokio::test(crate = "456")]
+   |                       ^^^^^
+
+error: Failed to parse value of `crate` as ident: "abc::edf"
+  --> $DIR/macros_invalid_input.rs:44:23
+   |
+44 | #[tokio::test(crate = "abc::edf")]
+   |                       ^^^^^^^^^^
+
+error: second test attribute is supplied
+  --> $DIR/macros_invalid_input.rs:48:1
+   |
+48 | #[test]
   | ^^^^^^^
+
+error: duplicated attribute
+  --> $DIR/macros_invalid_input.rs:48:1
+   |
+48 | #[test]
+   | ^^^^^^^
+   |
+note: the lint level is defined here
+  --> $DIR/macros_invalid_input.rs:1:9
+   |
+1  | #![deny(duplicate_macro_attributes)]
+   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -23,4 +23,13 @@ async fn extra_semicolon() -> Result<(), ()> {
    Ok(());
 }

+// https://github.com/tokio-rs/tokio/issues/4635
+#[allow(redundant_semicolons)]
+#[rustfmt::skip]
+#[tokio::main]
+async fn issue_4635() {
+    return 1;
+    ;
+}
+
 fn main() {}
@@ -1,25 +1,19 @@
 error[E0308]: mismatched types
 --> $DIR/macros_type_mismatch.rs:5:5
  |
+4 | async fn missing_semicolon_or_return_type() {
+  |                                             - help: a return type might be missing here: `-> _`
 5 |     Ok(())
  |     ^^^^^^ expected `()`, found enum `Result`
  |
  = note: expected unit type `()`
                  found enum `Result<(), _>`
-help: consider using a semicolon here
-  |
-5 |     Ok(());
-  |           +
-help: try adding a return type
-  |
-4 | async fn missing_semicolon_or_return_type() -> Result<(), _> {
-  |                                             ++++++++++++++++

 error[E0308]: mismatched types
  --> $DIR/macros_type_mismatch.rs:10:5
   |
 9  | async fn missing_return_type() {
-   |                                - help: try adding a return type: `-> Result<(), _>`
+   |                                - help: a return type might be missing here: `-> _`
 10 |     return Ok(());
   |     ^^^^^^^^^^^^^^ expected `()`, found enum `Result`
   |
@@ -37,9 +31,17 @@ error[E0308]: mismatched types
   |
   = note:   expected enum `Result<(), ()>`
           found unit type `()`
-help: try using a variant of the expected enum
+help: try adding an expression at the end of the block
   |
-23 |     Ok(Ok(());)
+23 ~     Ok(());;
+24 +     Ok(())
   |
-23 |     Err(Ok(());)
+
+error[E0308]: mismatched types
+  --> $DIR/macros_type_mismatch.rs:32:5
   |
+30 | async fn issue_4635() {
+   |                       - help: try adding a return type: `-> i32`
+31 |     return 1;
+32 |     ;
+   |     ^ expected `()`, found integer
@@ -16,11 +16,29 @@ required-features = ["rt-net"]
 name = "test-process-signal"
 required-features = ["rt-process-signal"]

+[[test]]
+name = "macros_main"
+
+[[test]]
+name = "macros_pin"
+
+[[test]]
+name = "macros_select"
+
+[[test]]
+name = "rt_yield"
+required-features = ["rt", "macros", "sync"]
+
 [features]
 # For mem check
 rt-net = ["tokio/rt", "tokio/rt-multi-thread", "tokio/net"]
 # For test-process-signal
-rt-process-signal = ["rt", "tokio/process", "tokio/signal"]
+rt-process-signal = ["rt-net", "tokio/process", "tokio/signal"]
+# For testing wasi + rt/macros/sync features
+#
+# This is an explicit feature so we can use `cargo hack` testing single features
+# instead of all possible permutations.
+wasi-rt = ["rt", "macros", "sync"]

 full = [
    "macros",
@@ -40,3 +58,4 @@ tokio = { path = "../tokio" }
 tokio-test = { path = "../tokio-test", optional = true }
 doc-comment = "0.3.1"
 futures = { version = "0.3.0", features = ["async-await"] }
+bytes = "1.0.0"
@@ -1,4 +1,8 @@
-#![cfg(all(feature = "macros", feature = "rt"))]
+#![cfg(all(
+    feature = "macros",
+    feature = "rt-multi-thread",
+    not(target_os = "wasi")
+))]

 #[tokio::main]
 async fn basic_main() -> usize {
@@ -4,6 +4,7 @@ use futures::channel::oneshot;
 use futures::executor::block_on;
 use std::thread;

+#[cfg_attr(target_os = "wasi", ignore = "WASI: std::thread::spawn not supported")]
 #[test]
 fn join_with_select() {
    block_on(async {
@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(feature = "full")]
+#![cfg(all(feature = "full", not(target_os = "wasi")))]

 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::join;
@@ -190,3 +190,54 @@ async fn pipe_from_one_command_to_another() {
    assert!(second_status.expect("second status").success());
    assert!(third_status.expect("third status").success());
 }
+
+#[tokio::test]
+async fn vectored_writes() {
+    use bytes::{Buf, Bytes};
+    use std::{io::IoSlice, pin::Pin};
+    use tokio::io::AsyncWrite;
+
+    let mut cat = cat().spawn().unwrap();
+    let mut stdin = cat.stdin.take().unwrap();
+    let are_writes_vectored = stdin.is_write_vectored();
+    let mut stdout = cat.stdout.take().unwrap();
+
+    let write = async {
+        let mut input = Bytes::from_static(b"hello\n").chain(Bytes::from_static(b"world!\n"));
+        let mut writes_completed = 0;
+
+        futures::future::poll_fn(|cx| loop {
+            let mut slices = [IoSlice::new(&[]); 2];
+            let vectored = input.chunks_vectored(&mut slices);
+            if vectored == 0 {
+                return std::task::Poll::Ready(std::io::Result::Ok(()));
+            }
+            let n = futures::ready!(Pin::new(&mut stdin).poll_write_vectored(cx, &slices))?;
+            writes_completed += 1;
+            input.advance(n);
+        })
+        .await?;
+
+        drop(stdin);
+
+        std::io::Result::Ok(writes_completed)
+    };
+
+    let read = async {
+        let mut buffer = Vec::with_capacity(6 + 7);
+        stdout.read_to_end(&mut buffer).await?;
+        std::io::Result::Ok(buffer)
+    };
+
+    let (write, read, status) = future::join3(write, read, cat.wait()).await;
+
+    assert!(status.unwrap().success());
+
+    let writes_completed = write.unwrap();
+    // on unix our small payload should always fit in whatever default sized pipe with a single
+    // syscall. if multiple are used, then the forwarding does not work, or we are on a platform
+    // for which the `std` does not support vectored writes.
+    assert_eq!(writes_completed == 1, are_writes_vectored);
+
+    assert_eq!(&read.unwrap(), b"hello\nworld!\n");
+}
@@ -0,0 +1,41 @@
+use tokio::sync::oneshot;
+use tokio::task;
+
+async fn spawn_send() {
+    let (tx, rx) = oneshot::channel();
+
+    let task = tokio::spawn(async {
+        for _ in 0..10 {
+            task::yield_now().await;
+        }
+
+        tx.send("done").unwrap();
+    });
+
+    assert_eq!("done", rx.await.unwrap());
+    task.await.unwrap();
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn entry_point() {
+    spawn_send().await;
+}
+
+#[tokio::test]
+async fn test_macro() {
+    spawn_send().await;
+}
+
+#[test]
+fn main_macro() {
+    entry_point();
+}
+
+#[test]
+fn manual_rt() {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+
+    rt.block_on(async { spawn_send().await });
+}
@@ -1,6 +1,32 @@
+# 1.8.2 (November 30th, 2022)
+
+- fix a regression introduced in 1.8.1 ([#5244])
+
+[#5244]: https://github.com/tokio-rs/tokio/pull/5244
+
+# 1.8.1 (November 29th, 2022)
+
+(yanked)
+
+- macros: Pin Futures in `#[tokio::test]` to stack ([#5205])
+- macros: Reduce usage of last statement spans in proc-macros ([#5092])
+- macros: Improve the documentation for `#[tokio::test]` ([#4761])
+
+[#5205]: https://github.com/tokio-rs/tokio/pull/5205
+[#5092]: https://github.com/tokio-rs/tokio/pull/5092
+[#4761]: https://github.com/tokio-rs/tokio/pull/4761
+
+# 1.8.0 (June 4th, 2022)
+
+- macros: always emit return statement ([#4636])
+- macros: support setting a custom crate name for `#[tokio::main]` and `#[tokio::test]` ([#4613])
+
+[#4613]: https://github.com/tokio-rs/tokio/pull/4613
+[#4636]: https://github.com/tokio-rs/tokio/pull/4636
+
 # 1.7.0 (December 15th, 2021)

- macros: address remainging clippy::semicolon_if_nothing_returned warning ([#4252])
+- macros: address remaining `clippy::semicolon_if_nothing_returned` warning ([#4252])

 [#4252]: https://github.com/tokio-rs/tokio/pull/4252

@@ -3,10 +3,10 @@ name = "tokio-macros"
 # When releasing to crates.io:
 # - Remove path dependencies
 # - Update CHANGELOG.md.
-# - Create "tokio-macros-1.0.x" git tag.
-version = "1.7.0"
+# - Create "tokio-macros-1.x.y" git tag.
+version = "1.8.2"
 edition = "2018"
-rust-version = "1.46"
+rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -1,4 +1,4 @@
-Copyright (c) 2021 Tokio Contributors
+Copyright (c) 2023 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -1,5 +1,5 @@
 use proc_macro::TokenStream;
-use proc_macro2::Span;
+use proc_macro2::{Ident, Span};
 use quote::{quote, quote_spanned, ToTokens};
 use syn::parse::Parser;

@@ -29,6 +29,7 @@ struct FinalConfig {
    flavor: RuntimeFlavor,
    worker_threads: Option<usize>,
    start_paused: Option<bool>,
+    crate_name: Option<String>,
 }

 /// Config used in case of the attribute not being able to build a valid config
@@ -36,6 +37,7 @@ const DEFAULT_ERROR_CONFIG: FinalConfig = FinalConfig {
    flavor: RuntimeFlavor::CurrentThread,
    worker_threads: None,
    start_paused: None,
+    crate_name: None,
 };

 struct Configuration {
@@ -45,6 +47,7 @@ struct Configuration {
    worker_threads: Option<(usize, Span)>,
    start_paused: Option<(bool, Span)>,
    is_test: bool,
+    crate_name: Option<String>,
 }

 impl Configuration {
@@ -59,6 +62,7 @@ impl Configuration {
            worker_threads: None,
            start_paused: None,
            is_test,
+            crate_name: None,
        }
    }

@@ -104,6 +108,15 @@ impl Configuration {
        Ok(())
    }

+    fn set_crate_name(&mut self, name: syn::Lit, span: Span) -> Result<(), syn::Error> {
+        if self.crate_name.is_some() {
+            return Err(syn::Error::new(span, "`crate` set multiple times."));
+        }
+        let name_ident = parse_ident(name, span, "crate")?;
+        self.crate_name = Some(name_ident.to_string());
+        Ok(())
+    }
+
    fn macro_name(&self) -> &'static str {
        if self.is_test {
            "tokio::test"
@@ -151,6 +164,7 @@ impl Configuration {
        };

        Ok(FinalConfig {
+            crate_name: self.crate_name.clone(),
            flavor,
            worker_threads,
            start_paused,
@@ -185,6 +199,27 @@ fn parse_string(int: syn::Lit, span: Span, field: &str) -> Result<String, syn::E
    }
 }

+fn parse_ident(lit: syn::Lit, span: Span, field: &str) -> Result<Ident, syn::Error> {
+    match lit {
+        syn::Lit::Str(s) => {
+            let err = syn::Error::new(
+                span,
+                format!(
+                    "Failed to parse value of `{}` as ident: \"{}\"",
+                    field,
+                    s.value()
+                ),
+            );
+            let path = s.parse::<syn::Path>().map_err(|_| err.clone())?;
+            path.get_ident().cloned().ok_or(err)
+        }
+        _ => Err(syn::Error::new(
+            span,
+            format!("Failed to parse value of `{}` as ident.", field),
+        )),
+    }
+}
+
 fn parse_bool(bool: syn::Lit, span: Span, field: &str) -> Result<bool, syn::Error> {
    match bool {
        syn::Lit::Bool(b) => Ok(b.value),
@@ -243,9 +278,15 @@ fn build_config(
                        let msg = "Attribute `core_threads` is renamed to `worker_threads`";
                        return Err(syn::Error::new_spanned(namevalue, msg));
                    }
+                    "crate" => {
+                        config.set_crate_name(
+                            namevalue.lit.clone(),
+                            syn::spanned::Spanned::span(&namevalue.lit),
+                        )?;
+                    }
                    name => {
                        let msg = format!(
-                            "Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`",
+                            "Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`",
                            name,
                        );
                        return Err(syn::Error::new_spanned(namevalue, msg));
@@ -275,7 +316,7 @@ fn build_config(
                        format!("The `{}` attribute requires an argument.", name)
                    }
                    name => {
-                        format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`", name)
+                        format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`", name)
                    }
                };
                return Err(syn::Error::new_spanned(path, msg));
@@ -313,12 +354,16 @@ fn parse_knobs(mut input: syn::ItemFn, is_test: bool, config: FinalConfig) -> To
        (start, end)
    };

+    let crate_name = config.crate_name.as_deref().unwrap_or("tokio");
+
+    let crate_ident = Ident::new(crate_name, last_stmt_start_span);
+
    let mut rt = match config.flavor {
        RuntimeFlavor::CurrentThread => quote_spanned! {last_stmt_start_span=>
-            tokio::runtime::Builder::new_current_thread()
+            #crate_ident::runtime::Builder::new_current_thread()
        },
        RuntimeFlavor::Threaded => quote_spanned! {last_stmt_start_span=>
-            tokio::runtime::Builder::new_multi_thread()
+            #crate_ident::runtime::Builder::new_multi_thread()
        },
    };
    if let Some(v) = config.worker_threads {
@@ -338,29 +383,50 @@ fn parse_knobs(mut input: syn::ItemFn, is_test: bool, config: FinalConfig) -> To

    let body = &input.block;
    let brace_token = input.block.brace_token;
-    let (tail_return, tail_semicolon) = match body.stmts.last() {
-        Some(syn::Stmt::Semi(syn::Expr::Return(_), _)) => (quote! { return }, quote! { ; }),
-        Some(syn::Stmt::Semi(..)) | Some(syn::Stmt::Local(..)) | None => {
-            match &input.sig.output {
-                syn::ReturnType::Type(_, ty) if matches!(&**ty, syn::Type::Tuple(ty) if ty.elems.is_empty()) =>
-                {
-                    (quote! {}, quote! { ; }) // unit
-                }
-                syn::ReturnType::Default => (quote! {}, quote! { ; }), // unit
-                syn::ReturnType::Type(..) => (quote! {}, quote! {}),   // ! or another
-            }
-        }
-        _ => (quote! {}, quote! {}),
-    };
-    input.block = syn::parse2(quote_spanned! {last_stmt_end_span=>
+    let body_ident = quote! { body };
+    let block_expr = quote_spanned! {last_stmt_end_span=>
+        #[allow(clippy::expect_used, clippy::diverging_sub_expression)]
        {
-            let body = async #body;
-            #[allow(clippy::expect_used)]
-            #tail_return #rt
+            return #rt
                .enable_all()
                .build()
                .expect("Failed building the Runtime")
-                .block_on(body)#tail_semicolon
+                .block_on(#body_ident);
+        }
+    };
+
+    // For test functions pin the body to the stack and use `Pin<&mut dyn
+    // Future>` to reduce the amount of `Runtime::block_on` (and related
+    // functions) copies we generate during compilation due to the generic
+    // parameter `F` (the future to block on). This could have an impact on
+    // performance, but because it's only for testing it's unlikely to be very
+    // large.
+    //
+    // We don't do this for the main function as it should only be used once so
+    // there will be no benefit.
+    let body = if is_test {
+        let output_type = match &input.sig.output {
+            // For functions with no return value syn doesn't print anything,
+            // but that doesn't work as `Output` for our boxed `Future`, so
+            // default to `()` (the same type as the function output).
+            syn::ReturnType::Default => quote! { () },
+            syn::ReturnType::Type(_, ret_type) => quote! { #ret_type },
+        };
+        quote! {
+            let body = async #body;
+            #crate_ident::pin!(body);
+            let body: ::std::pin::Pin<&mut dyn ::std::future::Future<Output = #output_type>> = body;
+        }
+    } else {
+        quote! {
+            let body = async #body;
+        }
+    };
+
+    input.block = syn::parse2(quote! {
+        {
+            #body
+            #block_expr
        }
    })
    .expect("Parsing failure");
@@ -414,7 +480,7 @@ pub(crate) fn test(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
    };
    let config = if let Some(attr) = input.attrs.iter().find(|attr| attr.path.is_ident("test")) {
        let msg = "second test attribute is supplied";
-        Err(syn::Error::new_spanned(&attr, msg))
+        Err(syn::Error::new_spanned(attr, msg))
    } else {
        AttributeArgs::parse_terminated
            .parse(args)
@@ -39,6 +39,13 @@ use proc_macro::TokenStream;
 /// function is called often, it is preferable to create the runtime using the
 /// runtime builder so the runtime can be reused across calls.
 ///
+/// # Non-worker async function
+///
+/// Note that the async function marked with this macro does not run as a
+/// worker. The expectation is that other tasks are spawned by the function here.
+/// Awaiting on other futures from the function provided here will not
+/// perform as fast as those spawned as workers.
+///
 /// # Multi-threaded runtime
 ///
 /// To use the multi-threaded runtime, the macro can be configured using
@@ -168,12 +175,32 @@ use proc_macro::TokenStream;
 ///
 /// Note that `start_paused` requires the `test-util` feature to be enabled.
 ///
-/// ### NOTE:
+/// ### Rename package
 ///
-/// If you rename the Tokio crate in your dependencies this macro will not work.
-/// If you must rename the current version of Tokio because you're also using an
-/// older version of Tokio, you _must_ make the current version of Tokio
-/// available as `tokio` in the module where this macro is expanded.
+/// ```rust
+/// use tokio as tokio1;
+///
+/// #[tokio1::main(crate = "tokio1")]
+/// async fn main() {
+///     println!("Hello world");
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::main]`
+///
+/// ```rust
+/// use tokio as tokio1;
+///
+/// fn main() {
+///     tokio1::runtime::Builder::new_multi_thread()
+///         .enable_all()
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             println!("Hello world");
+///         })
+/// }
+/// ```
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
@@ -213,23 +240,52 @@ pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
 /// }
 /// ```
 ///
-/// ### NOTE:
+/// ### Rename package
 ///
-/// If you rename the Tokio crate in your dependencies this macro will not work.
-/// If you must rename the current version of Tokio because you're also using an
-/// older version of Tokio, you _must_ make the current version of Tokio
-/// available as `tokio` in the module where this macro is expanded.
+/// ```rust
+/// use tokio as tokio1;
+///
+/// #[tokio1::main(crate = "tokio1")]
+/// async fn main() {
+///     println!("Hello world");
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::main]`
+///
+/// ```rust
+/// use tokio as tokio1;
+///
+/// fn main() {
+///     tokio1::runtime::Builder::new_multi_thread()
+///         .enable_all()
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             println!("Hello world");
+///         })
+/// }
+/// ```
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::main(args, item, false)
 }

-/// Marks async function to be executed by runtime, suitable to test environment
+/// Marks async function to be executed by runtime, suitable to test environment.
+/// This macro helps set up a `Runtime` without requiring the user to use
+/// [Runtime](../tokio/runtime/struct.Runtime.html) or
+/// [Builder](../tokio/runtime/struct.Builder.html) directly.
 ///
-/// ## Usage
+/// Note: This macro is designed to be simplistic and targets applications that
+/// do not require a complex setup. If the provided functionality is not
+/// sufficient, you may be interested in using
+/// [Builder](../tokio/runtime/struct.Builder.html), which provides a more
+/// powerful interface.
 ///
-/// ### Multi-thread runtime
+/// # Multi-threaded runtime
+///
+/// To use the multi-threaded runtime, the macro can be configured using
 ///
 /// ```no_run
 /// #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -238,9 +294,17 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// }
 /// ```
 ///
-/// ### Using default
+/// The `worker_threads` option configures the number of worker threads, and
+/// defaults to the number of cpus on the system. This is the default
+/// flavor.
 ///
-/// The default test runtime is single-threaded.
+/// Note: The multi-threaded runtime requires the `rt-multi-thread` feature
+/// flag.
+///
+/// # Current thread runtime
+///
+/// The default test runtime is single-threaded. Each test gets a
+/// separate current-thread runtime.
 ///
 /// ```no_run
 /// #[tokio::test]
@@ -249,6 +313,81 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// }
 /// ```
 ///
+/// ## Usage
+///
+/// ### Using the multi-thread runtime
+///
+/// ```no_run
+/// #[tokio::test(flavor = "multi_thread")]
+/// async fn my_test() {
+///     assert!(true);
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::test]`
+///
+/// ```no_run
+/// #[test]
+/// fn my_test() {
+///     tokio::runtime::Builder::new_multi_thread()
+///         .enable_all()
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             assert!(true);
+///         })
+/// }
+/// ```
+///
+/// ### Using current thread runtime
+///
+/// ```no_run
+/// #[tokio::test]
+/// async fn my_test() {
+///     assert!(true);
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::test]`
+///
+/// ```no_run
+/// #[test]
+/// fn my_test() {
+///     tokio::runtime::Builder::new_current_thread()
+///         .enable_all()
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             assert!(true);
+///         })
+/// }
+/// ```
+///
+/// ### Set number of worker threads
+///
+/// ```no_run
+/// #[tokio::test(flavor ="multi_thread", worker_threads = 2)]
+/// async fn my_test() {
+///     assert!(true);
+/// }
+/// ```
+///
+/// Equivalent code not using `#[tokio::test]`
+///
+/// ```no_run
+/// #[test]
+/// fn my_test() {
+///     tokio::runtime::Builder::new_multi_thread()
+///         .worker_threads(2)
+///         .enable_all()
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             assert!(true);
+///         })
+/// }
+/// ```
+///
 /// ### Configure the runtime to start with time paused
 ///
 /// ```no_run
@@ -258,14 +397,34 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// }
 /// ```
 ///
+/// Equivalent code not using `#[tokio::test]`
+///
+/// ```no_run
+/// #[test]
+/// fn my_test() {
+///     tokio::runtime::Builder::new_current_thread()
+///         .enable_all()
+///         .start_paused(true)
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             assert!(true);
+///         })
+/// }
+/// ```
+///
 /// Note that `start_paused` requires the `test-util` feature to be enabled.
 ///
-/// ### NOTE:
+/// ### Rename package
 ///
-/// If you rename the Tokio crate in your dependencies this macro will not work.
-/// If you must rename the current version of Tokio because you're also using an
-/// older version of Tokio, you _must_ make the current version of Tokio
-/// available as `tokio` in the module where this macro is expanded.
+/// ```rust
+/// use tokio as tokio1;
+///
+/// #[tokio1::test(crate = "tokio1")]
+/// async fn my_test() {
+///     println!("Hello world");
+/// }
+/// ```
 #[proc_macro_attribute]
 pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::test(args, item, true)
@@ -281,13 +440,6 @@ pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
 ///     assert!(true);
 /// }
 /// ```
-///
-/// ### NOTE:
-///
-/// If you rename the Tokio crate in your dependencies this macro will not work.
-/// If you must rename the current version of Tokio because you're also using an
-/// older version of Tokio, you _must_ make the current version of Tokio
-/// available as `tokio` in the module where this macro is expanded.
 #[proc_macro_attribute]
 pub fn test_rt(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::test(args, item, false)
@@ -100,10 +100,10 @@ fn clean_pattern(pat: &mut syn::Pat) {
        }
        syn::Pat::Reference(reference) => {
            reference.mutability = None;
-            clean_pattern(&mut *reference.pat);
+            clean_pattern(&mut reference.pat);
        }
        syn::Pat::Type(type_pat) => {
-            clean_pattern(&mut *type_pat.pat);
+            clean_pattern(&mut type_pat.pat);
        }
        _ => {}
    }
@@ -1,3 +1,47 @@
+# 0.1.12 (January 20, 2023)
+
+- time: remove `Unpin` bound on `Throttle` methods ([#5105])
+- time: document that `throttle` operates on ms granularity ([#5101])
+- sync: add `WatchStream::from_changes` ([#5432])
+
+[#5105]: https://github.com/tokio-rs/tokio/pull/5105
+[#5101]: https://github.com/tokio-rs/tokio/pull/5101
+[#5432]: https://github.com/tokio-rs/tokio/pull/5432
+
+# 0.1.11 (October 11, 2022)
+
+- time: allow `StreamExt::chunks_timeout` outside of a runtime ([#5036])
+
+[#5036]: https://github.com/tokio-rs/tokio/pull/5036
+
+# 0.1.10 (Sept 18, 2022)
+
+- time: add `StreamExt::chunks_timeout` ([#4695])
+- stream: add track_caller to public APIs ([#4786])
+
+[#4695]: https://github.com/tokio-rs/tokio/pull/4695
+[#4786]: https://github.com/tokio-rs/tokio/pull/4786
+
+# 0.1.9 (June 4, 2022)
+
+- deps: upgrade `tokio-util` dependency to `0.7.x` ([#3762])
+- stream: add `StreamExt::map_while` ([#4351])
+- stream: add `StreamExt::then` ([#4355])
+- stream: add cancel-safety docs to `StreamExt::next` and `try_next` ([#4715])
+- stream: expose `Elapsed` error ([#4502])
+- stream: expose `Timeout` ([#4601])
+- stream: implement `Extend` for `StreamMap` ([#4272])
+- sync: add `Clone` to `RecvError` types ([#4560])
+
+[#3762]: https://github.com/tokio-rs/tokio/pull/3762
+[#4272]: https://github.com/tokio-rs/tokio/pull/4272
+[#4351]: https://github.com/tokio-rs/tokio/pull/4351
+[#4355]: https://github.com/tokio-rs/tokio/pull/4355
+[#4502]: https://github.com/tokio-rs/tokio/pull/4502
+[#4560]: https://github.com/tokio-rs/tokio/pull/4560
+[#4601]: https://github.com/tokio-rs/tokio/pull/4601
+[#4715]: https://github.com/tokio-rs/tokio/pull/4715
+
 # 0.1.8 (October 29, 2021)

 - stream: add `From<Receiver<T>>` impl for receiver streams ([#4080])
@@ -4,9 +4,9 @@ name = "tokio-stream"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.8"
+version = "0.1.12"
 edition = "2018"
-rust-version = "1.46"
+rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -34,11 +34,10 @@ tokio-util = { version = "0.7.0", path = "../tokio-util", optional = true }
 [dev-dependencies]
 tokio = { version = "1.2.0", path = "../tokio", features = ["full", "test-util"] }
 async-stream = "0.3"
+parking_lot = "0.12.0"
 tokio-test = { path = "../tokio-test" }
 futures = { version = "0.3", default-features = false }

-proptest = "1"
-
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
@@ -1,4 +1,4 @@
-Copyright (c) 2021 Tokio Contributors
+Copyright (c) 2023 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -0,0 +1,4 @@
+target
+corpus
+artifacts
+coverage
@@ -0,0 +1,29 @@
+[package]
+name = "tokio-stream-fuzz"
+version = "0.0.0"
+publish = false
+edition = "2018"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies]
+libfuzzer-sys = "0.4"
+tokio-test = { path = "../../tokio-test" }
+
+[dependencies.tokio-stream]
+path = ".."
+
+
+# Prevent this from interfering with workspaces
+[workspace]
+members = ["."]
+
+[profile.release]
+debug = 1
+
+[[bin]]
+name = "fuzz_stream_map"
+path = "fuzz_targets/fuzz_stream_map.rs"
+test = false
+doc = false
@@ -0,0 +1,80 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use std::pin::Pin;
+
+use tokio_stream::{self as stream, pending, Stream, StreamExt, StreamMap};
+use tokio_test::{assert_ok, assert_pending, assert_ready, task};
+
+macro_rules! assert_ready_some {
+    ($($t:tt)*) => {
+        match assert_ready!($($t)*) {
+            Some(v) => v,
+            None => panic!("expected `Some`, got `None`"),
+        }
+    };
+}
+
+macro_rules! assert_ready_none {
+    ($($t:tt)*) => {
+        match assert_ready!($($t)*) {
+            None => {}
+            Some(v) => panic!("expected `None`, got `Some({:?})`", v),
+        }
+    };
+}
+
+fn pin_box<T: Stream<Item = U> + 'static, U>(s: T) -> Pin<Box<dyn Stream<Item = U>>> {
+    Box::pin(s)
+}
+
+fuzz_target!(|data: &[u8]| {
+    use std::task::{Context, Poll};
+
+    struct DidPoll<T> {
+        did_poll: bool,
+        inner: T,
+    }
+
+    impl<T: Stream + Unpin> Stream for DidPoll<T> {
+        type Item = T::Item;
+
+        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T::Item>> {
+            self.did_poll = true;
+            Pin::new(&mut self.inner).poll_next(cx)
+        }
+    }
+
+    for _ in 0..10 {
+        let mut map = task::spawn(StreamMap::new());
+        let mut expect = 0;
+
+        for (i, is_empty) in data.iter().map(|x| *x != 0).enumerate() {
+            let inner = if is_empty {
+                pin_box(stream::empty::<()>())
+            } else {
+                expect += 1;
+                pin_box(stream::pending::<()>())
+            };
+
+            let stream = DidPoll {
+                did_poll: false,
+                inner,
+            };
+
+            map.insert(i, stream);
+        }
+
+        if expect == 0 {
+            assert_ready_none!(map.poll_next());
+        } else {
+            assert_pending!(map.poll_next());
+
+            assert_eq!(expect, map.values().count());
+
+            for stream in map.values() {
+                assert!(stream.did_poll);
+            }
+        }
+    }
+});
@@ -77,6 +77,9 @@ pub mod wrappers;

 mod stream_ext;
 pub use stream_ext::{collect::FromStream, StreamExt};
+cfg_time! {
+    pub use stream_ext::timeout::{Elapsed, Timeout};
+}

 mod empty;
 pub use empty::{empty, Empty};
@@ -56,11 +56,13 @@ mod try_next;
 use try_next::TryNext;

 cfg_time! {
-    mod timeout;
+    pub(crate) mod timeout;
    use timeout::Timeout;
    use tokio::time::Duration;
    mod throttle;
    use throttle::{throttle, Throttle};
+    mod chunks_timeout;
+    use chunks_timeout::ChunksTimeout;
 }

 /// An extension trait for the [`Stream`] trait that provides a variety of
@@ -113,6 +115,12 @@ pub trait StreamExt: Stream {
    /// pinning it to the stack using the `pin_mut!` macro from the `pin_utils`
    /// crate.
    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. The returned future only
+    /// holds onto a reference to the underlying stream,
+    /// so dropping it will never lose a value.
+    ///
    /// # Examples
    ///
    /// ```
@@ -149,6 +157,12 @@ pub trait StreamExt: Stream {
    /// an [`Option<Result<T, E>>`](Option), making for easy use
    /// with the [`?`](std::ops::Try) operator.
    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. The returned future only
+    /// holds onto a reference to the underlying stream,
+    /// so dropping it will never lose a value.
+    ///
    /// # Examples
    ///
    /// ```
@@ -968,6 +982,8 @@ pub trait StreamExt: Stream {

    /// Slows down a stream by enforcing a delay between items.
    ///
+    /// The underlying timer behind this utility has a granularity of one millisecond.
+    ///
    /// # Example
    ///
    /// Create a throttled stream.
@@ -993,6 +1009,63 @@ pub trait StreamExt: Stream {
    {
        throttle(duration, self)
    }
+
+    /// Batches the items in the given stream using a maximum duration and size for each batch.
+    ///
+    /// This stream returns the next batch of items in the following situations:
+    ///  1. The inner stream has returned at least `max_size` many items since the last batch.
+    ///  2. The time since the first item of a batch is greater than the given duration.
+    ///  3. The end of the stream is reached.
+    ///
+    /// The length of the returned vector is never empty or greater than the maximum size. Empty batches
+    /// will not be emitted if no items are received upstream.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if `max_size` is zero
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use std::time::Duration;
+    /// use tokio::time;
+    /// use tokio_stream::{self as stream, StreamExt};
+    /// use futures::FutureExt;
+    ///
+    /// #[tokio::main]
+    /// # async fn _unused() {}
+    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
+    /// async fn main() {
+    ///     let iter = vec![1, 2, 3, 4].into_iter();
+    ///     let stream0 = stream::iter(iter);
+    ///
+    ///     let iter = vec![5].into_iter();
+    ///     let stream1 = stream::iter(iter)
+    ///          .then(move |n| time::sleep(Duration::from_secs(5)).map(move |_| n));
+    ///
+    ///     let chunk_stream = stream0
+    ///         .chain(stream1)
+    ///         .chunks_timeout(3, Duration::from_secs(2));
+    ///     tokio::pin!(chunk_stream);
+    ///
+    ///     // a full batch was received
+    ///     assert_eq!(chunk_stream.next().await, Some(vec![1,2,3]));
+    ///     // deadline was reached before max_size was reached
+    ///     assert_eq!(chunk_stream.next().await, Some(vec![4]));
+    ///     // last element in the stream
+    ///     assert_eq!(chunk_stream.next().await, Some(vec![5]));
+    /// }
+    /// ```
+    #[cfg(feature = "time")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
+    #[track_caller]
+    fn chunks_timeout(self, max_size: usize, duration: Duration) -> ChunksTimeout<Self>
+    where
+        Self: Sized,
+    {
+        assert!(max_size > 0, "`max_size` must be non-zero.");
+        ChunksTimeout::new(self, max_size, duration)
+    }
 }

 impl<St: ?Sized> StreamExt for St where St: Stream {}
@@ -1000,10 +1073,10 @@ impl<St: ?Sized> StreamExt for St where St: Stream {}
 /// Merge the size hints from two streams.
 fn merge_size_hints(
    (left_low, left_high): (usize, Option<usize>),
-    (right_low, right_hign): (usize, Option<usize>),
+    (right_low, right_high): (usize, Option<usize>),
 ) -> (usize, Option<usize>) {
    let low = left_low.saturating_add(right_low);
-    let high = match (left_high, right_hign) {
+    let high = match (left_high, right_high) {
        (Some(h1), Some(h2)) => h1.checked_add(h2),
        _ => None,
    };
@@ -0,0 +1,86 @@
+use crate::stream_ext::Fuse;
+use crate::Stream;
+use tokio::time::{sleep, Sleep};
+
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+use pin_project_lite::pin_project;
+use std::time::Duration;
+
+pin_project! {
+    /// Stream returned by the [`chunks_timeout`](super::StreamExt::chunks_timeout) method.
+    #[must_use = "streams do nothing unless polled"]
+    #[derive(Debug)]
+    pub struct ChunksTimeout<S: Stream> {
+        #[pin]
+        stream: Fuse<S>,
+        #[pin]
+        deadline: Option<Sleep>,
+        duration: Duration,
+        items: Vec<S::Item>,
+        cap: usize, // https://github.com/rust-lang/futures-rs/issues/1475
+    }
+}
+
+impl<S: Stream> ChunksTimeout<S> {
+    pub(super) fn new(stream: S, max_size: usize, duration: Duration) -> Self {
+        ChunksTimeout {
+            stream: Fuse::new(stream),
+            deadline: None,
+            duration,
+            items: Vec::with_capacity(max_size),
+            cap: max_size,
+        }
+    }
+}
+
+impl<S: Stream> Stream for ChunksTimeout<S> {
+    type Item = Vec<S::Item>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut me = self.as_mut().project();
+        loop {
+            match me.stream.as_mut().poll_next(cx) {
+                Poll::Pending => break,
+                Poll::Ready(Some(item)) => {
+                    if me.items.is_empty() {
+                        me.deadline.set(Some(sleep(*me.duration)));
+                        me.items.reserve_exact(*me.cap);
+                    }
+                    me.items.push(item);
+                    if me.items.len() >= *me.cap {
+                        return Poll::Ready(Some(std::mem::take(me.items)));
+                    }
+                }
+                Poll::Ready(None) => {
+                    // Returning Some here is only correct because we fuse the inner stream.
+                    let last = if me.items.is_empty() {
+                        None
+                    } else {
+                        Some(std::mem::take(me.items))
+                    };
+
+                    return Poll::Ready(last);
+                }
+            }
+        }
+
+        if !me.items.is_empty() {
+            if let Some(deadline) = me.deadline.as_pin_mut() {
+                ready!(deadline.poll(cx));
+            }
+            return Poll::Ready(Some(std::mem::take(me.items)));
+        }
+
+        Poll::Pending
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = (lower / self.cap).saturating_add(chunk_len);
+        let upper = upper.and_then(|x| x.checked_add(chunk_len));
+        (lower, upper)
+    }
+}
@@ -195,11 +195,7 @@ where
        } else {
            let res = mem::replace(collection, Ok(U::initialize(sealed::Internal, 0, Some(0))));

-            if let Err(err) = res {
-                Err(err)
-            } else {
-                unreachable!();
-            }
+            Err(res.map(drop).unwrap_err())
        }
    }
 }
@@ -8,6 +8,13 @@ use pin_project_lite::pin_project;

 pin_project! {
    /// Future for the [`next`](super::StreamExt::next) method.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. It only
+    /// holds onto a reference to the underlying stream,
+    /// so dropping it will never lose a value.
+    ///
    #[derive(Debug)]
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    pub struct Next<'a, St: ?Sized> {
@@ -72,7 +72,7 @@ where
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
-        let future_len = if self.future.is_some() { 1 } else { 0 };
+        let future_len = usize::from(self.future.is_some());
        let (lower, upper) = self.stream.size_hint();

        let lower = lower.saturating_add(future_len);
@@ -4,7 +4,6 @@ use crate::Stream;
 use tokio::time::{Duration, Instant, Sleep};

 use std::future::Future;
-use std::marker::Unpin;
 use std::pin::Pin;
 use std::task::{self, Poll};

@@ -41,8 +40,7 @@ pin_project! {
    }
 }

-// XXX: are these safe if `T: !Unpin`?
-impl<T: Unpin> Throttle<T> {
+impl<T> Throttle<T> {
    /// Acquires a reference to the underlying stream that this combinator is
    /// pulling from.
    pub fn get_ref(&self) -> &T {
@@ -24,7 +24,7 @@ pin_project! {
 }

 /// Error returned by `Timeout`.
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct Elapsed(());

 impl<S: Stream> Timeout<S> {
@@ -9,6 +9,12 @@ use pin_project_lite::pin_project;

 pin_project! {
    /// Future for the [`try_next`](super::StreamExt::try_next) method.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. It only
+    /// holds onto a reference to the underlying stream,
+    /// so dropping it will never lose a value.
    #[derive(Debug)]
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    pub struct TryNext<'a, St: ?Sized> {
@@ -14,11 +14,11 @@ use std::task::{Context, Poll};
 /// [`Stream`]: trait@crate::Stream
 #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
 pub struct BroadcastStream<T> {
-    inner: ReusableBoxFuture<(Result<T, RecvError>, Receiver<T>)>,
+    inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,
 }

 /// An error returned from the inner stream of a [`BroadcastStream`].
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq, Clone)]
 pub enum BroadcastStreamRecvError {
    /// The receiver lagged too far behind. Attempting to receive again will
    /// return the oldest message still retained by the channel.
@@ -10,8 +10,9 @@ use tokio::sync::watch::error::RecvError;

 /// A wrapper around [`tokio::sync::watch::Receiver`] that implements [`Stream`].
 ///
-/// This stream will always start by yielding the current value when the WatchStream is polled,
-/// regardless of whether it was the initial value or sent afterwards.
+/// This stream will start by yielding the current value when the WatchStream is polled,
+/// regardless of whether it was the initial value or sent afterwards,
+/// unless you use [`WatchStream<T>::from_changes`].
 ///
 /// # Examples
 ///
@@ -40,6 +41,28 @@ use tokio::sync::watch::error::RecvError;
 /// let (tx, rx) = watch::channel("hello");
 /// let mut rx = WatchStream::new(rx);
 ///
+/// // existing rx output with "hello" is ignored here
+///
+/// tx.send("goodbye").unwrap();
+/// assert_eq!(rx.next().await, Some("goodbye"));
+/// # }
+/// ```
+///
+/// Example with [`WatchStream<T>::from_changes`]:
+///
+/// ```
+/// # #[tokio::main]
+/// # async fn main() {
+/// use futures::future::FutureExt;
+/// use tokio::sync::watch;
+/// use tokio_stream::{StreamExt, wrappers::WatchStream};
+///
+/// let (tx, rx) = watch::channel("hello");
+/// let mut rx = WatchStream::from_changes(rx);
+///
+/// // no output from rx is available at this point - let's check this:
+/// assert!(rx.next().now_or_never().is_none());
+///
 /// tx.send("goodbye").unwrap();
 /// assert_eq!(rx.next().await, Some("goodbye"));
 /// # }
@@ -49,7 +72,7 @@ use tokio::sync::watch::error::RecvError;
 /// [`Stream`]: trait@crate::Stream
 #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
 pub struct WatchStream<T> {
-    inner: ReusableBoxFuture<(Result<(), RecvError>, Receiver<T>)>,
+    inner: ReusableBoxFuture<'static, (Result<(), RecvError>, Receiver<T>)>,
 }

 async fn make_future<T: Clone + Send + Sync>(
@@ -66,6 +89,13 @@ impl<T: 'static + Clone + Send + Sync> WatchStream<T> {
            inner: ReusableBoxFuture::new(async move { (Ok(()), rx) }),
        }
    }
+
+    /// Create a new `WatchStream` that waits for the value to be changed.
+    pub fn from_changes(rx: Receiver<T>) -> Self {
+        Self {
+            inner: ReusableBoxFuture::new(make_future(rx)),
+        }
+    }
 }

 impl<T: Clone + 'static + Send + Sync> Stream for WatchStream<T> {
@@ -0,0 +1,84 @@
+#![warn(rust_2018_idioms)]
+#![cfg(all(feature = "time", feature = "sync", feature = "io-util"))]
+
+use tokio::time;
+use tokio_stream::{self as stream, StreamExt};
+use tokio_test::assert_pending;
+use tokio_test::task;
+
+use futures::FutureExt;
+use std::time::Duration;
+
+#[tokio::test(start_paused = true)]
+async fn usage() {
+    let iter = vec![1, 2, 3].into_iter();
+    let stream0 = stream::iter(iter);
+
+    let iter = vec![4].into_iter();
+    let stream1 =
+        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(3)).map(move |_| n));
+
+    let chunk_stream = stream0
+        .chain(stream1)
+        .chunks_timeout(4, Duration::from_secs(2));
+
+    let mut chunk_stream = task::spawn(chunk_stream);
+
+    assert_pending!(chunk_stream.poll_next());
+    time::advance(Duration::from_secs(2)).await;
+    assert_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));
+
+    assert_pending!(chunk_stream.poll_next());
+    time::advance(Duration::from_secs(2)).await;
+    assert_eq!(chunk_stream.next().await, Some(vec![4]));
+}
+
+#[tokio::test(start_paused = true)]
+async fn full_chunk_with_timeout() {
+    let iter = vec![1, 2].into_iter();
+    let stream0 = stream::iter(iter);
+
+    let iter = vec![3].into_iter();
+    let stream1 =
+        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(1)).map(move |_| n));
+
+    let iter = vec![4].into_iter();
+    let stream2 =
+        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(3)).map(move |_| n));
+
+    let chunk_stream = stream0
+        .chain(stream1)
+        .chain(stream2)
+        .chunks_timeout(3, Duration::from_secs(2));
+
+    let mut chunk_stream = task::spawn(chunk_stream);
+
+    assert_pending!(chunk_stream.poll_next());
+    time::advance(Duration::from_secs(2)).await;
+    assert_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));
+
+    assert_pending!(chunk_stream.poll_next());
+    time::advance(Duration::from_secs(2)).await;
+    assert_eq!(chunk_stream.next().await, Some(vec![4]));
+}
+
+#[tokio::test]
+#[ignore]
+async fn real_time() {
+    let iter = vec![1, 2, 3, 4].into_iter();
+    let stream0 = stream::iter(iter);
+
+    let iter = vec![5].into_iter();
+    let stream1 =
+        stream::iter(iter).then(move |n| time::sleep(Duration::from_secs(5)).map(move |_| n));
+
+    let chunk_stream = stream0
+        .chain(stream1)
+        .chunks_timeout(3, Duration::from_secs(2));
+
+    let mut chunk_stream = task::spawn(chunk_stream);
+
+    assert_eq!(chunk_stream.next().await, Some(vec![1, 2, 3]));
+    assert_eq!(chunk_stream.next().await, Some(vec![4]));
+    assert_eq!(chunk_stream.next().await, Some(vec![5]));
+}
@@ -0,0 +1,55 @@
+#![warn(rust_2018_idioms)]
+#![cfg(all(feature = "time", not(target_os = "wasi")))] // Wasi does not support panic recovery
+
+use parking_lot::{const_mutex, Mutex};
+use std::error::Error;
+use std::panic;
+use std::sync::Arc;
+use tokio::time::Duration;
+use tokio_stream::{self as stream, StreamExt};
+
+fn test_panic<Func: FnOnce() + panic::UnwindSafe>(func: Func) -> Option<String> {
+    static PANIC_MUTEX: Mutex<()> = const_mutex(());
+
+    {
+        let _guard = PANIC_MUTEX.lock();
+        let panic_file: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
+
+        let prev_hook = panic::take_hook();
+        {
+            let panic_file = panic_file.clone();
+            panic::set_hook(Box::new(move |panic_info| {
+                let panic_location = panic_info.location().unwrap();
+                panic_file
+                    .lock()
+                    .clone_from(&Some(panic_location.file().to_string()));
+            }));
+        }
+
+        let result = panic::catch_unwind(func);
+        // Return to the previously set panic hook (maybe default) so that we get nice error
+        // messages in the tests.
+        panic::set_hook(prev_hook);
+
+        if result.is_err() {
+            panic_file.lock().clone()
+        } else {
+            None
+        }
+    }
+}
+
+#[test]
+fn stream_chunks_timeout_panic_caller() -> Result<(), Box<dyn Error>> {
+    let panic_location_file = test_panic(|| {
+        let iter = vec![1, 2, 3].into_iter();
+        let stream0 = stream::iter(iter);
+
+        let _chunk_stream = stream0.chunks_timeout(0, Duration::from_secs(2));
+    });
+
+    // The panic location should be in this file
+    assert_eq!(&panic_location_file.unwrap(), file!());
+
+    Ok(())
+}
@@ -325,62 +325,6 @@ fn one_ready_many_none() {
    }
 }

-proptest::proptest! {
-    #[test]
-    fn fuzz_pending_complete_mix(kinds: Vec<bool>) {
-        use std::task::{Context, Poll};
-
-        struct DidPoll<T> {
-            did_poll: bool,
-            inner: T,
-        }
-
-        impl<T: Stream + Unpin> Stream for DidPoll<T> {
-            type Item = T::Item;
-
-            fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>)
-                -> Poll<Option<T::Item>>
-            {
-                self.did_poll = true;
-                Pin::new(&mut self.inner).poll_next(cx)
-            }
-        }
-
-        for _ in 0..10 {
-            let mut map = task::spawn(StreamMap::new());
-            let mut expect = 0;
-
-            for (i, &is_empty) in kinds.iter().enumerate() {
-                let inner = if is_empty {
-                    pin_box(stream::empty::<()>())
-                } else {
-                    expect += 1;
-                    pin_box(stream::pending::<()>())
-                };
-
-                let stream = DidPoll {
-                    did_poll: false,
-                    inner,
-                };
-
-                map.insert(i, stream);
-            }
-
-            if expect == 0 {
-                assert_ready_none!(map.poll_next());
-            } else {
-                assert_pending!(map.poll_next());
-
-                assert_eq!(expect, map.values().count());
-
-                for stream in map.values() {
-                    assert!(stream.did_poll);
-                }
-            }
-        }
-    }
-}
-
 fn pin_box<T: Stream<Item = U> + 'static, U>(s: T) -> Pin<Box<dyn Stream<Item = U>>> {
    Box::pin(s)
 }
@@ -1,10 +1,10 @@
-#![cfg(feature = "full")]
+#![cfg(all(feature = "time", feature = "sync", feature = "io-util"))]

 use tokio::time::{self, sleep, Duration};
 use tokio_stream::{self, StreamExt};
 use tokio_test::*;

-use futures::StreamExt as _;
+use futures::stream;

 async fn maybe_sleep(idx: i32) -> i32 {
    if idx % 2 == 0 {
@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(feature = "full")]
+#![cfg(all(feature = "time", feature = "sync", feature = "io-util"))]

 use tokio::time;
 use tokio_stream::StreamExt;
@@ -3,9 +3,11 @@
 use tokio::sync::watch;
 use tokio_stream::wrappers::WatchStream;
 use tokio_stream::StreamExt;
+use tokio_test::assert_pending;
+use tokio_test::task::spawn;

 #[tokio::test]
-async fn message_not_twice() {
+async fn watch_stream_message_not_twice() {
    let (tx, rx) = watch::channel("hello");

    let mut counter = 0;
@@ -27,3 +29,29 @@ async fn message_not_twice() {
    drop(tx);
    task.await.unwrap();
 }
+
+#[tokio::test]
+async fn watch_stream_from_rx() {
+    let (tx, rx) = watch::channel("hello");
+
+    let mut stream = WatchStream::from(rx);
+
+    assert_eq!(stream.next().await.unwrap(), "hello");
+
+    tx.send("bye").unwrap();
+
+    assert_eq!(stream.next().await.unwrap(), "bye");
+}
+
+#[tokio::test]
+async fn watch_stream_from_changes() {
+    let (tx, rx) = watch::channel("hello");
+
+    let mut stream = WatchStream::from_changes(rx);
+
+    assert_pending!(spawn(&mut stream).poll_next());
+
+    tx.send("bye").unwrap();
+
+    assert_eq!(stream.next().await.unwrap(), "bye");
+}
@@ -6,7 +6,7 @@ name = "tokio-test"
 # - Create "tokio-test-0.4.x" git tag.
 version = "0.4.2"
 edition = "2018"
-rust-version = "1.46"
+rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -19,7 +19,7 @@ categories = ["asynchronous", "testing"]
 [dependencies]
 tokio = { version = "1.2.0", path = "../tokio", features = ["rt", "sync", "time", "test-util"] }
 tokio-stream = { version = "0.1.1", path = "../tokio-stream" }
-async-stream = "0.3"
+async-stream = "0.3.3"

 bytes = "1.0.0"
 futures-core = "0.3.0"
@@ -1,4 +1,4 @@
-Copyright (c) 2021 Tokio Contributors
+Copyright (c) 2023 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -260,7 +260,7 @@ macro_rules! assert_err {
    }};
 }

-/// Asserts that an exact duration has elapsed since since the start instant ±1ms.
+/// Asserts that an exact duration has elapsed since the start instant ±1ms.
 ///
 /// ```rust
 /// use tokio::time::{self, Instant};
@@ -1,4 +1,29 @@
-//! Futures task based helpers
+//! Futures task based helpers to easily test futures and manually written futures.
+//!
+//! The [`Spawn`] type is used as a mock task harness that allows you to poll futures
+//! without needing to setup pinning or context. Any future can be polled but if the
+//! future requires the tokio async context you will need to ensure that you poll the
+//! [`Spawn`] within a tokio context, this means that as long as you are inside the
+//! runtime it will work and you can poll it via [`Spawn`].
+//!
+//! [`Spawn`] also supports [`Stream`] to call `poll_next` without pinning
+//! or context.
+//!
+//! In addition to circumventing the need for pinning and context, [`Spawn`] also tracks
+//! the amount of times the future/task was woken. This can be useful to track if some
+//! leaf future notified the root task correctly.
+//!
+//! # Example
+//!
+//! ```
+//! use tokio_test::task;
+//!
+//! let fut = async {};
+//!
+//! let mut task = task::spawn(fut);
+//!
+//! assert!(task.poll().is_ready(), "Task was not ready!");
+//! ```

 #![allow(clippy::mutex_atomic)]

@@ -11,7 +36,11 @@ use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};

 use tokio_stream::Stream;

-/// TODO: dox
+/// Spawn a future into a [`Spawn`] which wraps the future in a mocked executor.
+///
+/// This can be used to spawn a [`Future`] or a [`Stream`].
+///
+/// For more information, check the module docs.
 pub fn spawn<T>(task: T) -> Spawn<T> {
    Spawn {
        task: MockTask::new(),
@@ -19,16 +48,14 @@ pub fn spawn<T>(task: T) -> Spawn<T> {
    }
 }

-/// Future spawned on a mock task
+/// Future spawned on a mock task that can be used to poll the future or stream
+/// without needing pinning or context types.
 #[derive(Debug)]
 pub struct Spawn<T> {
    task: MockTask,
    future: Pin<Box<T>>,
 }

-/// Mock task
-///
-/// A mock task is able to intercept and track wake notifications.
 #[derive(Debug, Clone)]
 struct MockTask {
    waker: Arc<ThreadWaker>,
@@ -91,7 +118,8 @@ impl<T: Unpin> ops::DerefMut for Spawn<T> {
 }

 impl<T: Future> Spawn<T> {
-    /// Polls a future
+    /// If `T` is a [`Future`] then poll it. This will handle pinning and the context
+    /// type for the future.
    pub fn poll(&mut self) -> Poll<T::Output> {
        let fut = self.future.as_mut();
        self.task.enter(|cx| fut.poll(cx))
@@ -99,7 +127,8 @@ impl<T: Future> Spawn<T> {
 }

 impl<T: Stream> Spawn<T> {
-    /// Polls a stream
+    /// If `T` is a [`Stream`] then poll_next it. This will handle pinning and the context
+    /// type for the stream.
    pub fn poll_next(&mut self) -> Poll<Option<T::Item>> {
        let stream = self.future.as_mut();
        self.task.enter(|cx| stream.poll_next(cx))
@@ -1,3 +1,153 @@
+# 0.7.7 (February 12, 2023)
+
+This release reverts the removal of the `Encoder` bound on the `FramedParts`
+constructor from [#5280] since it turned out to be a breaking change. ([#5450])
+
+[#5450]: https://github.com/tokio-rs/tokio/pull/5450
+
+# 0.7.6 (February 10, 2023)
+
+This release fixes a compilation failure in 0.7.5 when it is used together with
+Tokio version 1.21 and unstable features are enabled. ([#5445])
+
+[#5445]: https://github.com/tokio-rs/tokio/pull/5445
+
+# 0.7.5 (February 9, 2023)
+
+This release fixes an accidental breaking change where `UnwindSafe` was
+accidentally removed from `CancellationToken`.
+
+### Added
+- codec: add `Framed::backpressure_boundary` ([#5124])
+- io: add `InspectReader` and `InspectWriter` ([#5033])
+- io: add `tokio_util::io::{CopyToBytes, SinkWriter}` ([#5070], [#5436])
+- io: impl `std::io::BufRead` on `SyncIoBridge` ([#5265])
+- sync: add `PollSemaphore::poll_acquire_many` ([#5137])
+- sync: add owned future for `CancellationToken` ([#5153])
+- time: add `DelayQueue::try_remove` ([#5052])
+
+### Fixed
+- codec: fix `LengthDelimitedCodec` buffer over-reservation ([#4997])
+- sync: impl `UnwindSafe` on `CancellationToken` ([#5438])
+- util: remove `Encoder` bound on `FramedParts` constructor ([#5280])
+
+### Documented
+- io: add lines example for `StreamReader` ([#5145])
+
+[#4997]: https://github.com/tokio-rs/tokio/pull/4997
+[#5033]: https://github.com/tokio-rs/tokio/pull/5033
+[#5052]: https://github.com/tokio-rs/tokio/pull/5052
+[#5070]: https://github.com/tokio-rs/tokio/pull/5070
+[#5124]: https://github.com/tokio-rs/tokio/pull/5124
+[#5137]: https://github.com/tokio-rs/tokio/pull/5137
+[#5145]: https://github.com/tokio-rs/tokio/pull/5145
+[#5153]: https://github.com/tokio-rs/tokio/pull/5153
+[#5265]: https://github.com/tokio-rs/tokio/pull/5265
+[#5280]: https://github.com/tokio-rs/tokio/pull/5280
+[#5436]: https://github.com/tokio-rs/tokio/pull/5436
+[#5438]: https://github.com/tokio-rs/tokio/pull/5438
+
+# 0.7.4 (September 8, 2022)
+
+### Added
+
+- io: add `SyncIoBridge::shutdown()` ([#4938])
+- task: improve `LocalPoolHandle` ([#4680])
+
+### Fixed
+
+- util: add `track_caller` to public APIs ([#4785])
+
+### Unstable
+
+- task: fix compilation errors in `JoinMap` with Tokio v1.21.0 ([#4755])
+- task: remove the unstable, deprecated `JoinMap::join_one` ([#4920])
+
+[#4680]: https://github.com/tokio-rs/tokio/pull/4680
+[#4755]: https://github.com/tokio-rs/tokio/pull/4755
+[#4785]: https://github.com/tokio-rs/tokio/pull/4785
+[#4920]: https://github.com/tokio-rs/tokio/pull/4920
+[#4938]: https://github.com/tokio-rs/tokio/pull/4938
+
+# 0.7.3 (June 4, 2022)
+
+### Changed
+
+- tracing: don't require default tracing features ([#4592])
+- util: simplify implementation of `ReusableBoxFuture` ([#4675])
+
+### Added (unstable)
+
+- task: add `JoinMap` ([#4640], [#4697])
+
+[#4592]: https://github.com/tokio-rs/tokio/pull/4592
+[#4640]: https://github.com/tokio-rs/tokio/pull/4640
+[#4675]: https://github.com/tokio-rs/tokio/pull/4675
+[#4697]: https://github.com/tokio-rs/tokio/pull/4697
+
+# 0.7.2 (May 14, 2022)
+
+This release contains a rewrite of `CancellationToken` that fixes a memory leak. ([#4652])
+
+[#4652]: https://github.com/tokio-rs/tokio/pull/4652
+
+# 0.7.1 (February 21, 2022)
+
+### Added
+
+- codec: add `length_field_type` to `LengthDelimitedCodec` builder ([#4508])
+- io: add `StreamReader::into_inner_with_chunk()` ([#4559])
+
+### Changed
+
+- switch from log to tracing ([#4539])
+
+### Fixed
+
+- sync: fix waker update condition in `CancellationToken` ([#4497])
+- bumped tokio dependency to 1.6 to satisfy minimum requirements ([#4490])
+
+[#4490]: https://github.com/tokio-rs/tokio/pull/4490
+[#4497]: https://github.com/tokio-rs/tokio/pull/4497
+[#4508]: https://github.com/tokio-rs/tokio/pull/4508
+[#4539]: https://github.com/tokio-rs/tokio/pull/4539
+[#4559]: https://github.com/tokio-rs/tokio/pull/4559
+
+# 0.7.0 (February 9, 2022)
+
+### Added
+
+- task: add `spawn_pinned` ([#3370])
+- time: add `shrink_to_fit` and `compact` methods to `DelayQueue` ([#4170])
+- codec: improve `Builder::max_frame_length` docs ([#4352])
+- codec: add mutable reference getters for codecs to pinned `Framed` ([#4372])
+- net: add generic trait to combine `UnixListener` and `TcpListener` ([#4385])
+- codec: implement `Framed::map_codec` ([#4427])
+- codec: implement `Encoder<BytesMut>` for `BytesCodec` ([#4465])
+
+### Changed
+
+- sync: add lifetime parameter to `ReusableBoxFuture` ([#3762])
+- sync: refactored `PollSender<T>` to fix a subtly broken `Sink<T>` implementation ([#4214])
+- time: remove error case from the infallible `DelayQueue::poll_elapsed` ([#4241])
+
+[#3370]: https://github.com/tokio-rs/tokio/pull/3370
+[#4170]: https://github.com/tokio-rs/tokio/pull/4170
+[#4352]: https://github.com/tokio-rs/tokio/pull/4352
+[#4372]: https://github.com/tokio-rs/tokio/pull/4372
+[#4385]: https://github.com/tokio-rs/tokio/pull/4385
+[#4427]: https://github.com/tokio-rs/tokio/pull/4427
+[#4465]: https://github.com/tokio-rs/tokio/pull/4465
+[#3762]: https://github.com/tokio-rs/tokio/pull/3762
+[#4214]: https://github.com/tokio-rs/tokio/pull/4214
+[#4241]: https://github.com/tokio-rs/tokio/pull/4241
+
+# 0.6.10 (May 14, 2021)
+
+This is a backport for the memory leak in `CancellationToken` that was originally fixed in 0.7.2. ([#4652])
+
+[#4652]: https://github.com/tokio-rs/tokio/pull/4652
+
 # 0.6.9 (October 29, 2021)

 ### Added
@@ -4,9 +4,9 @@ name = "tokio-util"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.0"
+version = "0.7.7"
 edition = "2018"
-rust-version = "1.46"
+rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -15,7 +15,6 @@ description = """
 Additional utilities for working with Tokio.
 """
 categories = ["asynchronous"]
-publish = false

 [features]
 # No features on by default
@@ -26,25 +25,27 @@ full = ["codec", "compat", "io-util", "time", "net", "rt"]

 net = ["tokio/net"]
 compat = ["futures-io",]
-codec = []
+codec = ["tracing"]
 time = ["tokio/time","slab"]
 io = []
 io-util = ["io", "tokio/rt", "tokio/io-util"]
-rt = ["tokio/rt"]
+rt = ["tokio/rt", "tokio/sync", "futures-util", "hashbrown"]

 __docs_rs = ["futures-util"]

 [dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["sync"] }
-
+tokio = { version = "1.22.0", path = "../tokio", features = ["sync"] }
 bytes = "1.0.0"
 futures-core = "0.3.0"
 futures-sink = "0.3.0"
 futures-io = { version = "0.3.0", optional = true }
 futures-util = { version = "0.3.0", optional = true }
-log = "0.4"
 pin-project-lite = "0.2.0"
-slab = { version = "0.4.1", optional = true } # Backs `DelayQueue`
+slab = { version = "0.4.4", optional = true } # Backs `DelayQueue`
+tracing = { version = "0.1.25", default-features = false, features = ["std"], optional = true }
+
+[target.'cfg(tokio_unstable)'.dependencies]
+hashbrown = { version = "0.12.0", optional = true }

 [dev-dependencies]
 tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
@@ -54,7 +55,12 @@ tokio-stream = { version = "0.1", path = "../tokio-stream" }
 async-stream = "0.3.0"
 futures = "0.3.0"
 futures-test = "0.3.5"
+parking_lot = "0.12.0"

 [package.metadata.docs.rs]
 all-features = true
-rustdoc-args = ["--cfg", "docsrs"]
+# enable unstable features in the documentation
+rustdoc-args = ["--cfg", "docsrs", "--cfg", "tokio_unstable"]
+# it's necessary to _also_ pass `--cfg tokio_unstable` to rustc, or else
+# dependencies will not be enabled, and the docs build will fail.
+rustc-args = ["--cfg", "docsrs", "--cfg", "tokio_unstable"]
@@ -1,4 +1,4 @@
-Copyright (c) 2021 Tokio Contributors
+Copyright (c) 2023 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -74,3 +74,13 @@ impl Encoder<Bytes> for BytesCodec {
        Ok(())
    }
 }
+
+impl Encoder<BytesMut> for BytesCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, data: BytesMut, buf: &mut BytesMut) -> Result<(), io::Error> {
+        buf.reserve(data.len());
+        buf.put(data);
+        Ok(())
+    }
+}
@@ -20,7 +20,7 @@ use std::io;
 /// it's possible to temporarily read 0 bytes by reaching EOF.
 ///
 /// In these cases `decode_eof` will be called until it signals
-/// fullfillment of all closing frames by returning `Ok(None)`.
+/// fulfillment of all closing frames by returning `Ok(None)`.
 /// After that, repeated attempts to read from the [`Framed`] or [`FramedRead`]
 /// will not invoke `decode` or `decode_eof` again, until data can be read
 /// during a retry.
@@ -204,6 +204,26 @@ impl<T, U> Framed<T, U> {
        &mut self.inner.codec
    }

+    /// Maps the codec `U` to `C`, preserving the read and write buffers
+    /// wrapped by `Framed`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying codec
+    /// as it may corrupt the stream of frames otherwise being worked with.
+    pub fn map_codec<C, F>(self, map: F) -> Framed<T, C>
+    where
+        F: FnOnce(U) -> C,
+    {
+        // This could be potentially simplified once rust-lang/rust#86555 hits stable
+        let parts = self.into_parts();
+        Framed::from_parts(FramedParts {
+            io: parts.io,
+            codec: map(parts.codec),
+            read_buf: parts.read_buf,
+            write_buf: parts.write_buf,
+            _priv: (),
+        })
+    }
+
    /// Returns a mutable reference to the underlying codec wrapped by
    /// `Framed`.
    ///
@@ -233,6 +253,16 @@ impl<T, U> Framed<T, U> {
        &mut self.inner.state.write.buffer
    }

+    /// Returns backpressure boundary
+    pub fn backpressure_boundary(&self) -> usize {
+        self.inner.state.write.backpressure_boundary
+    }
+
+    /// Updates backpressure boundary
+    pub fn set_backpressure_boundary(&mut self, boundary: usize) {
+        self.inner.state.write.backpressure_boundary = boundary;
+    }
+
    /// Consumes the `Framed`, returning its underlying I/O stream.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
@@ -7,12 +7,12 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use bytes::BytesMut;
 use futures_core::ready;
 use futures_sink::Sink;
-use log::trace;
 use pin_project_lite::pin_project;
 use std::borrow::{Borrow, BorrowMut};
 use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use tracing::trace;

 pin_project! {
    #[derive(Debug)]
@@ -25,7 +25,6 @@ pin_project! {
 }

 const INITIAL_CAPACITY: usize = 8 * 1024;
-const BACKPRESSURE_BOUNDARY: usize = INITIAL_CAPACITY;

 #[derive(Debug)]
 pub(crate) struct ReadFrame {
@@ -37,6 +36,7 @@ pub(crate) struct ReadFrame {

 pub(crate) struct WriteFrame {
    pub(crate) buffer: BytesMut,
+    pub(crate) backpressure_boundary: usize,
 }

 #[derive(Default)]
@@ -60,6 +60,7 @@ impl Default for WriteFrame {
    fn default() -> Self {
        Self {
            buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
+            backpressure_boundary: INITIAL_CAPACITY,
        }
    }
 }
@@ -87,7 +88,10 @@ impl From<BytesMut> for WriteFrame {
            buffer.reserve(INITIAL_CAPACITY - size);
        }

-        Self { buffer }
+        Self {
+            buffer,
+            backpressure_boundary: INITIAL_CAPACITY,
+        }
    }
 }

@@ -256,7 +260,7 @@ where
    type Error = U::Error;

    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if self.state.borrow().buffer.len() >= BACKPRESSURE_BOUNDARY {
+        if self.state.borrow().buffer.len() >= self.state.borrow().backpressure_boundary {
            self.as_mut().poll_flush(cx)
        } else {
            Poll::Ready(Ok(()))
@@ -277,8 +281,8 @@ where
        let mut pinned = self.project();

        while !pinned.state.borrow_mut().buffer.is_empty() {
-            let WriteFrame { buffer } = pinned.state.borrow_mut();
-            trace!("writing; remaining={}", buffer.len());
+            let WriteFrame { buffer, .. } = pinned.state.borrow_mut();
+            trace!(remaining = buffer.len(), "writing;");

            let n = ready!(poll_write_buf(pinned.inner.as_mut(), cx, buffer))?;

@@ -108,6 +108,27 @@ impl<T, D> FramedRead<T, D> {
        &mut self.inner.codec
    }

+    /// Maps the decoder `D` to `C`, preserving the read buffer
+    /// wrapped by `Framed`.
+    pub fn map_decoder<C, F>(self, map: F) -> FramedRead<T, C>
+    where
+        F: FnOnce(D) -> C,
+    {
+        // This could be potentially simplified once rust-lang/rust#86555 hits stable
+        let FramedImpl {
+            inner,
+            state,
+            codec,
+        } = self.inner;
+        FramedRead {
+            inner: FramedImpl {
+                inner,
+                state,
+                codec: map(codec),
+            },
+        }
+    }
+
    /// Returns a mutable reference to the underlying decoder.
    pub fn decoder_pin_mut(self: Pin<&mut Self>) -> &mut D {
        self.project().inner.project().codec
@@ -88,6 +88,27 @@ impl<T, E> FramedWrite<T, E> {
        &mut self.inner.codec
    }

+    /// Maps the encoder `E` to `C`, preserving the write buffer
+    /// wrapped by `Framed`.
+    pub fn map_encoder<C, F>(self, map: F) -> FramedWrite<T, C>
+    where
+        F: FnOnce(E) -> C,
+    {
+        // This could be potentially simplified once rust-lang/rust#86555 hits stable
+        let FramedImpl {
+            inner,
+            state,
+            codec,
+        } = self.inner;
+        FramedWrite {
+            inner: FramedImpl {
+                inner,
+                state,
+                codec: map(codec),
+            },
+        }
+    }
+
    /// Returns a mutable reference to the underlying encoder.
    pub fn encoder_pin_mut(self: Pin<&mut Self>) -> &mut E {
        self.project().inner.project().codec
@@ -102,6 +123,16 @@ impl<T, E> FramedWrite<T, E> {
    pub fn write_buffer_mut(&mut self) -> &mut BytesMut {
        &mut self.inner.state.buffer
    }
+
+    /// Returns backpressure boundary
+    pub fn backpressure_boundary(&self) -> usize {
+        self.inner.state.backpressure_boundary
+    }
+
+    /// Updates backpressure boundary
+    pub fn set_backpressure_boundary(&mut self, boundary: usize) {
+        self.inner.state.backpressure_boundary = boundary;
+    }
 }

 // This impl just defers to the underlying FramedImpl
@@ -84,7 +84,7 @@
 //! # fn bind_read<T: AsyncRead>(io: T) {
 //! LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
-//!     .length_field_length(2)
+//!     .length_field_type::<u16>()
 //!     .length_adjustment(0)   // default value
 //!     .num_skip(0) // Do not strip frame header
 //!     .new_read(io);
@@ -118,7 +118,7 @@
 //! # fn bind_read<T: AsyncRead>(io: T) {
 //! LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
-//!     .length_field_length(2)
+//!     .length_field_type::<u16>()
 //!     .length_adjustment(0)   // default value
 //!     // `num_skip` is not needed, the default is to skip
 //!     .new_read(io);
@@ -150,7 +150,7 @@
 //! # fn bind_read<T: AsyncRead>(io: T) {
 //! LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
-//!     .length_field_length(2)
+//!     .length_field_type::<u16>()
 //!     .length_adjustment(-2)  // size of head
 //!     .num_skip(0)
 //!     .new_read(io);
@@ -228,7 +228,7 @@
 //! # fn bind_read<T: AsyncRead>(io: T) {
 //! LengthDelimitedCodec::builder()
 //!     .length_field_offset(1) // length of hdr1
-//!     .length_field_length(2)
+//!     .length_field_type::<u16>()
 //!     .length_adjustment(1)  // length of hdr2
 //!     .num_skip(3) // length of hdr1 + LEN
 //!     .new_read(io);
@@ -274,7 +274,7 @@
 //! # fn bind_read<T: AsyncRead>(io: T) {
 //! LengthDelimitedCodec::builder()
 //!     .length_field_offset(1) // length of hdr1
-//!     .length_field_length(2)
+//!     .length_field_type::<u16>()
 //!     .length_adjustment(-3)  // length of hdr1 + LEN, negative
 //!     .num_skip(3)
 //!     .new_read(io);
@@ -350,7 +350,7 @@
 //! # fn write_frame<T: AsyncWrite>(io: T) {
 //! # let _ =
 //! LengthDelimitedCodec::builder()
-//!     .length_field_length(2)
+//!     .length_field_type::<u16>()
 //!     .new_write(io);
 //! # }
 //! # pub fn main() {}
@@ -379,7 +379,7 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use std::error::Error as StdError;
 use std::io::{self, Cursor};
-use std::{cmp, fmt};
+use std::{cmp, fmt, mem};

 /// Configure length delimited `LengthDelimitedCodec`s.
 ///
@@ -522,15 +522,11 @@ impl LengthDelimitedCodec {
            }
        };

-        let num_skip = self.builder.get_num_skip();
-
-        if num_skip > 0 {
-            src.advance(num_skip);
-        }
+        src.advance(self.builder.get_num_skip());

        // Ensure that the buffer has enough space to read the incoming
        // payload
-        src.reserve(n);
+        src.reserve(n.saturating_sub(src.len()));

        Ok(Some(n))
    }
@@ -568,7 +564,7 @@ impl Decoder for LengthDelimitedCodec {
                self.state = DecodeState::Head;

                // Make sure the buffer has enough space to read the next head
-                src.reserve(self.builder.num_head_bytes());
+                src.reserve(self.builder.num_head_bytes().saturating_sub(src.len()));

                Ok(Some(data))
            }
@@ -629,6 +625,24 @@ impl Default for LengthDelimitedCodec {

 // ===== impl Builder =====

+mod builder {
+    /// Types that can be used with `Builder::length_field_type`.
+    pub trait LengthFieldType {}
+
+    impl LengthFieldType for u8 {}
+    impl LengthFieldType for u16 {}
+    impl LengthFieldType for u32 {}
+    impl LengthFieldType for u64 {}
+
+    #[cfg(any(
+        target_pointer_width = "8",
+        target_pointer_width = "16",
+        target_pointer_width = "32",
+        target_pointer_width = "64",
+    ))]
+    impl LengthFieldType for usize {}
+}
+
 impl Builder {
    /// Creates a new length delimited codec builder with default configuration
    /// values.
@@ -642,7 +656,7 @@ impl Builder {
    /// # fn bind_read<T: AsyncRead>(io: T) {
    /// LengthDelimitedCodec::builder()
    ///     .length_field_offset(0)
-    ///     .length_field_length(2)
+    ///     .length_field_type::<u16>()
    ///     .length_adjustment(0)
    ///     .num_skip(0)
    ///     .new_read(io);
@@ -777,6 +791,42 @@ impl Builder {
        self
    }

+    /// Sets the unsigned integer type used to represent the length field.
+    ///
+    /// The default type is [`u32`]. The max type is [`u64`] (or [`usize`] on
+    /// 64-bit targets).
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use tokio::io::AsyncRead;
+    /// use tokio_util::codec::LengthDelimitedCodec;
+    ///
+    /// # fn bind_read<T: AsyncRead>(io: T) {
+    /// LengthDelimitedCodec::builder()
+    ///     .length_field_type::<u32>()
+    ///     .new_read(io);
+    /// # }
+    /// # pub fn main() {}
+    /// ```
+    ///
+    /// Unlike [`Builder::length_field_length`], this does not fail at runtime
+    /// and instead produces a compile error:
+    ///
+    /// ```compile_fail
+    /// # use tokio::io::AsyncRead;
+    /// # use tokio_util::codec::LengthDelimitedCodec;
+    /// # fn bind_read<T: AsyncRead>(io: T) {
+    /// LengthDelimitedCodec::builder()
+    ///     .length_field_type::<u128>()
+    ///     .new_read(io);
+    /// # }
+    /// # pub fn main() {}
+    /// ```
+    pub fn length_field_type<T: builder::LengthFieldType>(&mut self) -> &mut Self {
+        self.length_field_length(mem::size_of::<T>())
+    }
+
    /// Sets the number of bytes used to represent the length field
    ///
    /// The default value is `4`. The max value is `8`.
@@ -878,7 +928,7 @@ impl Builder {
    /// # pub fn main() {
    /// LengthDelimitedCodec::builder()
    ///     .length_field_offset(0)
-    ///     .length_field_length(2)
+    ///     .length_field_type::<u16>()
    ///     .length_adjustment(0)
    ///     .num_skip(0)
    ///     .new_codec();
@@ -902,7 +952,7 @@ impl Builder {
    /// # fn bind_read<T: AsyncRead>(io: T) {
    /// LengthDelimitedCodec::builder()
    ///     .length_field_offset(0)
-    ///     .length_field_length(2)
+    ///     .length_field_type::<u16>()
    ///     .length_adjustment(0)
    ///     .num_skip(0)
    ///     .new_read(io);
@@ -925,7 +975,7 @@ impl Builder {
    /// # use tokio_util::codec::LengthDelimitedCodec;
    /// # fn write_frame<T: AsyncWrite>(io: T) {
    /// LengthDelimitedCodec::builder()
-    ///     .length_field_length(2)
+    ///     .length_field_type::<u16>()
    ///     .new_write(io);
    /// # }
    /// # pub fn main() {}
@@ -947,7 +997,7 @@ impl Builder {
    /// # fn write_frame<T: AsyncRead + AsyncWrite>(io: T) {
    /// # let _ =
    /// LengthDelimitedCodec::builder()
-    ///     .length_field_length(2)
+    ///     .length_field_type::<u16>()
    ///     .new_framed(io);
    /// # }
    /// # pub fn main() {}
@@ -0,0 +1,68 @@
+use bytes::Bytes;
+use futures_sink::Sink;
+use pin_project_lite::pin_project;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A helper that wraps a [`Sink`]`<`[`Bytes`]`>` and converts it into a
+    /// [`Sink`]`<&'a [u8]>` by copying each byte slice into an owned [`Bytes`].
+    ///
+    /// See the documentation for [`SinkWriter`] for an example.
+    ///
+    /// [`Bytes`]: bytes::Bytes
+    /// [`SinkWriter`]: crate::io::SinkWriter
+    /// [`Sink`]: futures_sink::Sink
+    #[derive(Debug)]
+    pub struct CopyToBytes<S> {
+        #[pin]
+        inner: S,
+    }
+}
+
+impl<S> CopyToBytes<S> {
+    /// Creates a new [`CopyToBytes`].
+    pub fn new(inner: S) -> Self {
+        Self { inner }
+    }
+
+    /// Gets a reference to the underlying sink.
+    pub fn get_ref(&self) -> &S {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying sink.
+    pub fn get_mut(&mut self) -> &mut S {
+        &mut self.inner
+    }
+
+    /// Consumes this [`CopyToBytes`], returning the underlying sink.
+    pub fn into_inner(self) -> S {
+        self.inner
+    }
+}
+
+impl<'a, S> Sink<&'a [u8]> for CopyToBytes<S>
+where
+    S: Sink<Bytes>,
+{
+    type Error = S::Error;
+
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().inner.poll_ready(cx)
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: &'a [u8]) -> Result<(), Self::Error> {
+        self.project()
+            .inner
+            .start_send(Bytes::copy_from_slice(item))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().inner.poll_flush(cx)
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().inner.poll_close(cx)
+    }
+}
@@ -0,0 +1,134 @@
+use futures_core::ready;
+use pin_project_lite::pin_project;
+use std::io::{IoSlice, Result};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+pin_project! {
+    /// An adapter that lets you inspect the data that's being read.
+    ///
+    /// This is useful for things like hashing data as it's read in.
+    pub struct InspectReader<R, F> {
+        #[pin]
+        reader: R,
+        f: F,
+    }
+}
+
+impl<R, F> InspectReader<R, F> {
+    /// Create a new InspectReader, wrapping `reader` and calling `f` for the
+    /// new data supplied by each read call.
+    ///
+    /// The closure will only be called with an empty slice if the inner reader
+    /// returns without reading data into the buffer. This happens at EOF, or if
+    /// `poll_read` is called with a zero-size buffer.
+    pub fn new(reader: R, f: F) -> InspectReader<R, F>
+    where
+        R: AsyncRead,
+        F: FnMut(&[u8]),
+    {
+        InspectReader { reader, f }
+    }
+
+    /// Consumes the `InspectReader`, returning the wrapped reader
+    pub fn into_inner(self) -> R {
+        self.reader
+    }
+}
+
+impl<R: AsyncRead, F: FnMut(&[u8])> AsyncRead for InspectReader<R, F> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<Result<()>> {
+        let me = self.project();
+        let filled_length = buf.filled().len();
+        ready!(me.reader.poll_read(cx, buf))?;
+        (me.f)(&buf.filled()[filled_length..]);
+        Poll::Ready(Ok(()))
+    }
+}
+
+pin_project! {
+    /// An adapter that lets you inspect the data that's being written.
+    ///
+    /// This is useful for things like hashing data as it's written out.
+    pub struct InspectWriter<W, F> {
+        #[pin]
+        writer: W,
+        f: F,
+    }
+}
+
+impl<W, F> InspectWriter<W, F> {
+    /// Create a new InspectWriter, wrapping `write` and calling `f` for the
+    /// data successfully written by each write call.
+    ///
+    /// The closure `f` will never be called with an empty slice. A vectored
+    /// write can result in multiple calls to `f` - at most one call to `f` per
+    /// buffer supplied to `poll_write_vectored`.
+    pub fn new(writer: W, f: F) -> InspectWriter<W, F>
+    where
+        W: AsyncWrite,
+        F: FnMut(&[u8]),
+    {
+        InspectWriter { writer, f }
+    }
+
+    /// Consumes the `InspectWriter`, returning the wrapped writer
+    pub fn into_inner(self) -> W {
+        self.writer
+    }
+}
+
+impl<W: AsyncWrite, F: FnMut(&[u8])> AsyncWrite for InspectWriter<W, F> {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize>> {
+        let me = self.project();
+        let res = me.writer.poll_write(cx, buf);
+        if let Poll::Ready(Ok(count)) = res {
+            if count != 0 {
+                (me.f)(&buf[..count]);
+            }
+        }
+        res
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        let me = self.project();
+        me.writer.poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        let me = self.project();
+        me.writer.poll_shutdown(cx)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<Result<usize>> {
+        let me = self.project();
+        let res = me.writer.poll_write_vectored(cx, bufs);
+        if let Poll::Ready(Ok(mut count)) = res {
+            for buf in bufs {
+                if count == 0 {
+                    break;
+                }
+                let size = count.min(buf.len());
+                if size != 0 {
+                    (me.f)(&buf[..size]);
+                    count -= size;
+                }
+            }
+        }
+        res
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.writer.is_write_vectored()
+    }
+}
@@ -10,15 +10,22 @@
 //! [`Body`]: https://docs.rs/hyper/0.13/hyper/struct.Body.html
 //! [`AsyncRead`]: tokio::io::AsyncRead

+mod copy_to_bytes;
+mod inspect;
 mod read_buf;
 mod reader_stream;
+mod sink_writer;
 mod stream_reader;
+
 cfg_io_util! {
    mod sync_bridge;
    pub use self::sync_bridge::SyncIoBridge;
 }

+pub use self::copy_to_bytes::CopyToBytes;
+pub use self::inspect::{InspectReader, InspectWriter};
 pub use self::read_buf::read_buf;
 pub use self::reader_stream::ReaderStream;
+pub use self::sink_writer::SinkWriter;
 pub use self::stream_reader::StreamReader;
 pub use crate::util::{poll_read_buf, poll_write_buf};
@@ -0,0 +1,117 @@
+use futures_core::ready;
+use futures_sink::Sink;
+
+use pin_project_lite::pin_project;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::AsyncWrite;
+
+pin_project! {
+    /// Convert a [`Sink`] of byte chunks into an [`AsyncWrite`].
+    ///
+    /// Whenever you write to this [`SinkWriter`], the supplied bytes are
+    /// forwarded to the inner [`Sink`]. When `shutdown` is called on this
+    /// [`SinkWriter`], the inner sink is closed.
+    ///
+    /// This adapter takes a `Sink<&[u8]>` and provides an [`AsyncWrite`] impl
+    /// for it. Because of the lifetime, this trait is relatively rarely
+    /// implemented. The main ways to get a `Sink<&[u8]>` that you can use with
+    /// this type are:
+    ///
+    ///  * With the codec module by implementing the [`Encoder`]`<&[u8]>` trait.
+    ///  * By wrapping a `Sink<Bytes>` in a [`CopyToBytes`].
+    ///  * Manually implementing `Sink<&[u8]>` directly.
+    ///
+    /// The opposite conversion of implementing `Sink<_>` for an [`AsyncWrite`]
+    /// is done using the [`codec`] module.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use bytes::Bytes;
+    /// use futures_util::SinkExt;
+    /// use std::io::{Error, ErrorKind};
+    /// use tokio::io::AsyncWriteExt;
+    /// use tokio_util::io::{SinkWriter, CopyToBytes};
+    /// use tokio_util::sync::PollSender;
+    ///
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() -> Result<(), Error> {
+    /// // We use an mpsc channel as an example of a `Sink<Bytes>`.
+    /// let (tx, mut rx) = tokio::sync::mpsc::channel::<Bytes>(1);
+    /// let sink = PollSender::new(tx).sink_map_err(|_| Error::from(ErrorKind::BrokenPipe));
+    ///
+    /// // Wrap it in `CopyToBytes` to get a `Sink<&[u8]>`.
+    /// let mut writer = SinkWriter::new(CopyToBytes::new(sink));
+    ///
+    /// // Write data to our interface...
+    /// let data: [u8; 4] = [1, 2, 3, 4];
+    /// let _ = writer.write(&data).await?;
+    ///
+    /// // ... and receive it.
+    /// assert_eq!(data.as_slice(), &*rx.recv().await.unwrap());
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// [`AsyncWrite`]: tokio::io::AsyncWrite
+    /// [`CopyToBytes`]: crate::io::CopyToBytes
+    /// [`Encoder`]: crate::codec::Encoder
+    /// [`Sink`]: futures_sink::Sink
+    /// [`codec`]: tokio_util::codec
+    #[derive(Debug)]
+    pub struct SinkWriter<S> {
+        #[pin]
+        inner: S,
+    }
+}
+
+impl<S> SinkWriter<S> {
+    /// Creates a new [`SinkWriter`].
+    pub fn new(sink: S) -> Self {
+        Self { inner: sink }
+    }
+
+    /// Gets a reference to the underlying sink.
+    pub fn get_ref(&self) -> &S {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying sink.
+    pub fn get_mut(&mut self) -> &mut S {
+        &mut self.inner
+    }
+
+    /// Consumes this [`SinkWriter`], returning the underlying sink.
+    pub fn into_inner(self) -> S {
+        self.inner
+    }
+}
+impl<S, E> AsyncWrite for SinkWriter<S>
+where
+    for<'a> S: Sink<&'a [u8], Error = E>,
+    E: Into<io::Error>,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        let mut this = self.project();
+
+        ready!(this.inner.as_mut().poll_ready(cx).map_err(Into::into))?;
+        match this.inner.as_mut().start_send(buf) {
+            Ok(()) => Poll::Ready(Ok(buf.len())),
+            Err(e) => Poll::Ready(Err(e.into())),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        self.project().inner.poll_flush(cx).map_err(Into::into)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        self.project().inner.poll_close(cx).map_err(Into::into)
+    }
+}
@@ -1,64 +1,162 @@
 use bytes::Buf;
 use futures_core::stream::Stream;
-use pin_project_lite::pin_project;
 use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::io::{AsyncBufRead, AsyncRead, ReadBuf};

-pin_project! {
-    /// Convert a [`Stream`] of byte chunks into an [`AsyncRead`].
-    ///
-    /// This type performs the inverse operation of [`ReaderStream`].
-    ///
-    /// # Example
-    ///
-    /// ```
-    /// use bytes::Bytes;
-    /// use tokio::io::{AsyncReadExt, Result};
-    /// use tokio_util::io::StreamReader;
-    /// # #[tokio::main]
-    /// # async fn main() -> std::io::Result<()> {
-    ///
-    /// // Create a stream from an iterator.
-    /// let stream = tokio_stream::iter(vec![
-    ///     Result::Ok(Bytes::from_static(&[0, 1, 2, 3])),
-    ///     Result::Ok(Bytes::from_static(&[4, 5, 6, 7])),
-    ///     Result::Ok(Bytes::from_static(&[8, 9, 10, 11])),
-    /// ]);
-    ///
-    /// // Convert it to an AsyncRead.
-    /// let mut read = StreamReader::new(stream);
-    ///
-    /// // Read five bytes from the stream.
-    /// let mut buf = [0; 5];
-    /// read.read_exact(&mut buf).await?;
-    /// assert_eq!(buf, [0, 1, 2, 3, 4]);
-    ///
-    /// // Read the rest of the current chunk.
-    /// assert_eq!(read.read(&mut buf).await?, 3);
-    /// assert_eq!(&buf[..3], [5, 6, 7]);
-    ///
-    /// // Read the next chunk.
-    /// assert_eq!(read.read(&mut buf).await?, 4);
-    /// assert_eq!(&buf[..4], [8, 9, 10, 11]);
-    ///
-    /// // We have now reached the end.
-    /// assert_eq!(read.read(&mut buf).await?, 0);
-    ///
-    /// # Ok(())
-    /// # }
-    /// ```
-    ///
-    /// [`AsyncRead`]: tokio::io::AsyncRead
-    /// [`Stream`]: futures_core::Stream
-    /// [`ReaderStream`]: crate::io::ReaderStream
-    #[derive(Debug)]
-    pub struct StreamReader<S, B> {
-        #[pin]
-        inner: S,
-        chunk: Option<B>,
-    }
+/// Convert a [`Stream`] of byte chunks into an [`AsyncRead`].
+///
+/// This type performs the inverse operation of [`ReaderStream`].
+///
+/// This type also implements the [`AsyncBufRead`] trait, so you can use it
+/// to read a `Stream` of byte chunks line-by-line. See the examples below.
+///
+/// # Example
+///
+/// ```
+/// use bytes::Bytes;
+/// use tokio::io::{AsyncReadExt, Result};
+/// use tokio_util::io::StreamReader;
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+///
+/// // Create a stream from an iterator.
+/// let stream = tokio_stream::iter(vec![
+///     Result::Ok(Bytes::from_static(&[0, 1, 2, 3])),
+///     Result::Ok(Bytes::from_static(&[4, 5, 6, 7])),
+///     Result::Ok(Bytes::from_static(&[8, 9, 10, 11])),
+/// ]);
+///
+/// // Convert it to an AsyncRead.
+/// let mut read = StreamReader::new(stream);
+///
+/// // Read five bytes from the stream.
+/// let mut buf = [0; 5];
+/// read.read_exact(&mut buf).await?;
+/// assert_eq!(buf, [0, 1, 2, 3, 4]);
+///
+/// // Read the rest of the current chunk.
+/// assert_eq!(read.read(&mut buf).await?, 3);
+/// assert_eq!(&buf[..3], [5, 6, 7]);
+///
+/// // Read the next chunk.
+/// assert_eq!(read.read(&mut buf).await?, 4);
+/// assert_eq!(&buf[..4], [8, 9, 10, 11]);
+///
+/// // We have now reached the end.
+/// assert_eq!(read.read(&mut buf).await?, 0);
+///
+/// # Ok(())
+/// # }
+/// ```
+///
+/// If the stream produces errors which are not [`std::io::Error`],
+/// the errors can be converted using [`StreamExt`] to map each
+/// element.
+///
+/// ```
+/// use bytes::Bytes;
+/// use tokio::io::AsyncReadExt;
+/// use tokio_util::io::StreamReader;
+/// use tokio_stream::StreamExt;
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+///
+/// // Create a stream from an iterator, including an error.
+/// let stream = tokio_stream::iter(vec![
+///     Result::Ok(Bytes::from_static(&[0, 1, 2, 3])),
+///     Result::Ok(Bytes::from_static(&[4, 5, 6, 7])),
+///     Result::Err("Something bad happened!")
+/// ]);
+///
+/// // Use StreamExt to map the stream and error to a std::io::Error
+/// let stream = stream.map(|result| result.map_err(|err| {
+///     std::io::Error::new(std::io::ErrorKind::Other, err)
+/// }));
+///
+/// // Convert it to an AsyncRead.
+/// let mut read = StreamReader::new(stream);
+///
+/// // Read five bytes from the stream.
+/// let mut buf = [0; 5];
+/// read.read_exact(&mut buf).await?;
+/// assert_eq!(buf, [0, 1, 2, 3, 4]);
+///
+/// // Read the rest of the current chunk.
+/// assert_eq!(read.read(&mut buf).await?, 3);
+/// assert_eq!(&buf[..3], [5, 6, 7]);
+///
+/// // Reading the next chunk will produce an error
+/// let error = read.read(&mut buf).await.unwrap_err();
+/// assert_eq!(error.kind(), std::io::ErrorKind::Other);
+/// assert_eq!(error.into_inner().unwrap().to_string(), "Something bad happened!");
+///
+/// // We have now reached the end.
+/// assert_eq!(read.read(&mut buf).await?, 0);
+///
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Using the [`AsyncBufRead`] impl, you can read a `Stream` of byte chunks
+/// line-by-line. Note that you will usually also need to convert the error
+/// type when doing this. See the second example for an explanation of how
+/// to do this.
+///
+/// ```
+/// use tokio::io::{Result, AsyncBufReadExt};
+/// use tokio_util::io::StreamReader;
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> std::io::Result<()> {
+///
+/// // Create a stream of byte chunks.
+/// let stream = tokio_stream::iter(vec![
+///     Result::Ok(b"The first line.\n".as_slice()),
+///     Result::Ok(b"The second line.".as_slice()),
+///     Result::Ok(b"\nThe third".as_slice()),
+///     Result::Ok(b" line.\nThe fourth line.\nThe fifth line.\n".as_slice()),
+/// ]);
+///
+/// // Convert it to an AsyncRead.
+/// let mut read = StreamReader::new(stream);
+///
+/// // Loop through the lines from the `StreamReader`.
+/// let mut line = String::new();
+/// let mut lines = Vec::new();
+/// loop {
+///     line.clear();
+///     let len = read.read_line(&mut line).await?;
+///     if len == 0 { break; }
+///     lines.push(line.clone());
+/// }
+///
+/// // Verify that we got the lines we expected.
+/// assert_eq!(
+///     lines,
+///     vec![
+///         "The first line.\n",
+///         "The second line.\n",
+///         "The third line.\n",
+///         "The fourth line.\n",
+///         "The fifth line.\n",
+///     ]
+/// );
+/// # Ok(())
+/// # }
+/// ```
+///
+/// [`AsyncRead`]: tokio::io::AsyncRead
+/// [`AsyncBufRead`]: tokio::io::AsyncBufRead
+/// [`Stream`]: futures_core::Stream
+/// [`ReaderStream`]: crate::io::ReaderStream
+/// [`StreamExt`]: https://docs.rs/tokio-stream/latest/tokio_stream/trait.StreamExt.html
+#[derive(Debug)]
+pub struct StreamReader<S, B> {
+    // This field is pinned.
+    inner: S,
+    // This field is not pinned.
+    chunk: Option<B>,
 }

 impl<S, B, E> StreamReader<S, B>
@@ -84,13 +182,24 @@ where
    }

    /// Do we have a chunk and is it non-empty?
-    fn has_chunk(self: Pin<&mut Self>) -> bool {
-        if let Some(chunk) = self.project().chunk {
+    fn has_chunk(&self) -> bool {
+        if let Some(ref chunk) = self.chunk {
            chunk.remaining() > 0
        } else {
            false
        }
    }
+
+    /// Consumes this `StreamReader`, returning a Tuple consisting
+    /// of the underlying stream and an Option of the internal buffer,
+    /// which is Some in case the buffer contains elements.
+    pub fn into_inner_with_chunk(self) -> (S, Option<B>) {
+        if self.has_chunk() {
+            (self.inner, self.chunk)
+        } else {
+            (self.inner, None)
+        }
+    }
 }

 impl<S, B> StreamReader<S, B> {
@@ -118,6 +227,10 @@ impl<S, B> StreamReader<S, B> {
    /// Consumes this `BufWriter`, returning the underlying stream.
    ///
    /// Note that any leftover data in the internal buffer is lost.
+    /// If you additionally want access to the internal buffer use
+    /// [`into_inner_with_chunk`].
+    ///
+    /// [`into_inner_with_chunk`]: crate::io::StreamReader::into_inner_with_chunk
    pub fn into_inner(self) -> S {
        self.inner
    }
@@ -186,3 +299,28 @@ where
        }
    }
 }
+
+// The code below is a manual expansion of the code that pin-project-lite would
+// generate. This is done because pin-project-lite fails by hitting the recusion
+// limit on this struct. (Every line of documentation is handled recursively by
+// the macro.)
+
+impl<S: Unpin, B> Unpin for StreamReader<S, B> {}
+
+struct StreamReaderProject<'a, S, B> {
+    inner: Pin<&'a mut S>,
+    chunk: &'a mut Option<B>,
+}
+
+impl<S, B> StreamReader<S, B> {
+    #[inline]
+    fn project(self: Pin<&mut Self>) -> StreamReaderProject<'_, S, B> {
+        // SAFETY: We define that only `inner` should be pinned when `Self` is
+        // and have an appropriate `impl Unpin` for this.
+        let me = unsafe { Pin::into_inner_unchecked(self) };
+        StreamReaderProject {
+            inner: unsafe { Pin::new_unchecked(&mut me.inner) },
+            chunk: &mut me.chunk,
+        }
+    }
+}
@@ -1,5 +1,7 @@
-use std::io::{Read, Write};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use std::io::{BufRead, Read, Write};
+use tokio::io::{
+    AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt,
+};

 /// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
 /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
@@ -9,6 +11,28 @@ pub struct SyncIoBridge<T> {
    rt: tokio::runtime::Handle,
 }

+impl<T: AsyncBufRead + Unpin> BufRead for SyncIoBridge<T> {
+    fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
+        let src = &mut self.src;
+        self.rt.block_on(AsyncBufReadExt::fill_buf(src))
+    }
+
+    fn consume(&mut self, amt: usize) {
+        let src = &mut self.src;
+        AsyncBufReadExt::consume(src, amt)
+    }
+
+    fn read_until(&mut self, byte: u8, buf: &mut Vec<u8>) -> std::io::Result<usize> {
+        let src = &mut self.src;
+        self.rt
+            .block_on(AsyncBufReadExt::read_until(src, byte, buf))
+    }
+    fn read_line(&mut self, buf: &mut String) -> std::io::Result<usize> {
+        let src = &mut self.src;
+        self.rt.block_on(AsyncBufReadExt::read_line(src, buf))
+    }
+}
+
 impl<T: AsyncRead + Unpin> Read for SyncIoBridge<T> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let src = &mut self.src;
@@ -66,6 +90,21 @@ impl<T: AsyncWrite> SyncIoBridge<T> {
    }
 }

+impl<T: AsyncWrite + Unpin> SyncIoBridge<T> {
+    /// Shutdown this writer. This method provides a way to call the [`AsyncWriteExt::shutdown`]
+    /// function of the inner [`tokio::io::AsyncWrite`] instance.
+    ///
+    /// # Errors
+    ///
+    /// This method returns the same errors as [`AsyncWriteExt::shutdown`].
+    ///
+    /// [`AsyncWriteExt::shutdown`]: tokio::io::AsyncWriteExt::shutdown
+    pub fn shutdown(&mut self) -> std::io::Result<()> {
+        let src = &mut self.src;
+        self.rt.block_on(src.shutdown())
+    }
+}
+
 impl<T: Unpin> SyncIoBridge<T> {
    /// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
    /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
@@ -85,9 +124,10 @@ impl<T: Unpin> SyncIoBridge<T> {
    ///
    /// Use e.g. `SyncIoBridge::new(Box::pin(src))`.
    ///
-    /// # Panic
+    /// # Panics
    ///
    /// This will panic if called outside the context of a Tokio runtime.
+    #[track_caller]
    pub fn new(src: T) -> Self {
        Self::new_with_handle(src, tokio::runtime::Handle::current())
    }
@@ -29,7 +29,9 @@ cfg_codec! {
 }

 cfg_net! {
+    #[cfg(not(target_arch = "wasm32"))]
    pub mod udp;
+    pub mod net;
 }

 cfg_compat! {
@@ -42,6 +44,7 @@ cfg_io! {

 cfg_rt! {
    pub mod context;
+    pub mod task;
 }

 cfg_time! {
@@ -113,6 +116,9 @@ mod util {

        let n = {
            let dst = buf.chunk_mut();
+
+            // Safety: `chunk_mut()` returns a `&mut UninitSlice`, and `UninitSlice` is a
+            // transparent wrapper around `[MaybeUninit<u8>]`.
            let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
            let mut buf = ReadBuf::uninit(dst);
            let ptr = buf.filled().as_ptr();
@@ -0,0 +1,97 @@
+//! TCP/UDP/Unix helpers for tokio.
+
+use crate::either::Either;
+use std::future::Future;
+use std::io::Result;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[cfg(unix)]
+pub mod unix;
+
+/// A trait for a listener: `TcpListener` and `UnixListener`.
+pub trait Listener {
+    /// The stream's type of this listener.
+    type Io: tokio::io::AsyncRead + tokio::io::AsyncWrite;
+    /// The socket address type of this listener.
+    type Addr;
+
+    /// Polls to accept a new incoming connection to this listener.
+    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>>;
+
+    /// Accepts a new incoming connection from this listener.
+    fn accept(&mut self) -> ListenerAcceptFut<'_, Self>
+    where
+        Self: Sized,
+    {
+        ListenerAcceptFut { listener: self }
+    }
+
+    /// Returns the local address that this listener is bound to.
+    fn local_addr(&self) -> Result<Self::Addr>;
+}
+
+impl Listener for tokio::net::TcpListener {
+    type Io = tokio::net::TcpStream;
+    type Addr = std::net::SocketAddr;
+
+    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
+        Self::poll_accept(self, cx)
+    }
+
+    fn local_addr(&self) -> Result<Self::Addr> {
+        self.local_addr().map(Into::into)
+    }
+}
+
+/// Future for accepting a new connection from a listener.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ListenerAcceptFut<'a, L> {
+    listener: &'a mut L,
+}
+
+impl<'a, L> Future for ListenerAcceptFut<'a, L>
+where
+    L: Listener,
+{
+    type Output = Result<(L::Io, L::Addr)>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.listener.poll_accept(cx)
+    }
+}
+
+impl<L, R> Either<L, R>
+where
+    L: Listener,
+    R: Listener,
+{
+    /// Accepts a new incoming connection from this listener.
+    pub async fn accept(&mut self) -> Result<Either<(L::Io, L::Addr), (R::Io, R::Addr)>> {
+        match self {
+            Either::Left(listener) => {
+                let (stream, addr) = listener.accept().await?;
+                Ok(Either::Left((stream, addr)))
+            }
+            Either::Right(listener) => {
+                let (stream, addr) = listener.accept().await?;
+                Ok(Either::Right((stream, addr)))
+            }
+        }
+    }
+
+    /// Returns the local address that this listener is bound to.
+    pub fn local_addr(&self) -> Result<Either<L::Addr, R::Addr>> {
+        match self {
+            Either::Left(listener) => {
+                let addr = listener.local_addr()?;
+                Ok(Either::Left(addr))
+            }
+            Either::Right(listener) => {
+                let addr = listener.local_addr()?;
+                Ok(Either::Right(addr))
+            }
+        }
+    }
+}
@@ -0,0 +1,18 @@
+//! Unix domain socket helpers.
+
+use super::Listener;
+use std::io::Result;
+use std::task::{Context, Poll};
+
+impl Listener for tokio::net::UnixListener {
+    type Io = tokio::net::UnixStream;
+    type Addr = tokio::net::unix::SocketAddr;
+
+    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
+        Self::poll_accept(self, cx)
+    }
+
+    fn local_addr(&self) -> Result<Self::Addr> {
+        self.local_addr().map(Into::into)
+    }
+}
@@ -1,18 +1,15 @@
 //! An asynchronously awaitable `CancellationToken`.
 //! The token allows to signal a cancellation request to one or more tasks.
 pub(crate) mod guard;
+mod tree_node;

-use crate::loom::sync::atomic::AtomicUsize;
-use crate::loom::sync::Mutex;
-use crate::sync::intrusive_double_linked_list::{LinkedList, ListNode};
-
+use crate::loom::sync::Arc;
 use core::future::Future;
 use core::pin::Pin;
-use core::ptr::NonNull;
-use core::sync::atomic::Ordering;
-use core::task::{Context, Poll, Waker};
+use core::task::{Context, Poll};

 use guard::DropGuard;
+use pin_project_lite::pin_project;

 /// A token which can be used to signal a cancellation request to one or more
 /// tasks.
@@ -24,9 +21,9 @@ use guard::DropGuard;
 ///
 /// # Examples
 ///
-/// ```ignore
+/// ```no_run
 /// use tokio::select;
-/// use tokio::scope::CancellationToken;
+/// use tokio_util::sync::CancellationToken;
 ///
 /// #[tokio::main]
 /// async fn main() {
@@ -55,30 +52,39 @@ use guard::DropGuard;
 /// }
 /// ```
 pub struct CancellationToken {
-    inner: NonNull<CancellationTokenState>,
+    inner: Arc<tree_node::TreeNode>,
 }

-// Safety: The CancellationToken is thread-safe and can be moved between threads,
-// since all methods are internally synchronized.
-unsafe impl Send for CancellationToken {}
-unsafe impl Sync for CancellationToken {}
+impl std::panic::UnwindSafe for CancellationToken {}
+impl std::panic::RefUnwindSafe for CancellationToken {}

-/// A Future that is resolved once the corresponding [`CancellationToken`]
-/// was cancelled
-#[must_use = "futures do nothing unless polled"]
-pub struct WaitForCancellationFuture<'a> {
-    /// The CancellationToken that is associated with this WaitForCancellationFuture
-    cancellation_token: Option<&'a CancellationToken>,
-    /// Node for waiting at the cancellation_token
-    wait_node: ListNode<WaitQueueEntry>,
-    /// Whether this future was registered at the token yet as a waiter
-    is_registered: bool,
+pin_project! {
+    /// A Future that is resolved once the corresponding [`CancellationToken`]
+    /// is cancelled.
+    #[must_use = "futures do nothing unless polled"]
+    pub struct WaitForCancellationFuture<'a> {
+        cancellation_token: &'a CancellationToken,
+        #[pin]
+        future: tokio::sync::futures::Notified<'a>,
+    }
 }

-// Safety: Futures can be sent between threads as long as the underlying
-// cancellation_token is thread-safe (Sync),
-// which allows to poll/register/unregister from a different thread.
-unsafe impl<'a> Send for WaitForCancellationFuture<'a> {}
+pin_project! {
+    /// A Future that is resolved once the corresponding [`CancellationToken`]
+    /// is cancelled.
+    ///
+    /// This is the counterpart to [`WaitForCancellationFuture`] that takes
+    /// [`CancellationToken`] by value instead of using a reference.
+    #[must_use = "futures do nothing unless polled"]
+    pub struct WaitForCancellationFutureOwned {
+        // Since `future` is the first field, it is dropped before the
+        // cancellation_token field. This ensures that the reference inside the
+        // `Notified` remains valid.
+        #[pin]
+        future: tokio::sync::futures::Notified<'static>,
+        cancellation_token: CancellationToken,
+    }
+}

 // ===== impl CancellationToken =====

@@ -92,43 +98,16 @@ impl core::fmt::Debug for CancellationToken {

 impl Clone for CancellationToken {
    fn clone(&self) -> Self {
-        // Safety: The state inside a `CancellationToken` is always valid, since
-        // is reference counted
-        let inner = self.state();
-
-        // Tokens are cloned by increasing their refcount
-        let current_state = inner.snapshot();
-        inner.increment_refcount(current_state);
-
-        CancellationToken { inner: self.inner }
+        tree_node::increase_handle_refcount(&self.inner);
+        CancellationToken {
+            inner: self.inner.clone(),
+        }
    }
 }

 impl Drop for CancellationToken {
    fn drop(&mut self) {
-        let token_state_pointer = self.inner;
-
-        // Safety: The state inside a `CancellationToken` is always valid, since
-        // is reference counted
-        let inner = unsafe { &mut *self.inner.as_ptr() };
-
-        let mut current_state = inner.snapshot();
-
-        // We need to safe the parent, since the state might be released by the
-        // next call
-        let parent = inner.parent;
-
-        // Drop our own refcount
-        current_state = inner.decrement_refcount(current_state);
-
-        // If this was the last reference, unregister from the parent
-        if current_state.refcount == 0 {
-            if let Some(mut parent) = parent {
-                // Safety: Since we still retain a reference on the parent, it must be valid.
-                let parent = unsafe { parent.as_mut() };
-                parent.unregister_child(token_state_pointer, current_state);
-            }
-        }
+        tree_node::decrease_handle_refcount(&self.inner);
    }
 }

@@ -141,29 +120,11 @@ impl Default for CancellationToken {
 impl CancellationToken {
    /// Creates a new CancellationToken in the non-cancelled state.
    pub fn new() -> CancellationToken {
-        let state = Box::new(CancellationTokenState::new(
-            None,
-            StateSnapshot {
-                cancel_state: CancellationState::NotCancelled,
-                has_parent_ref: false,
-                refcount: 1,
-            },
-        ));
-
-        // Safety: We just created the Box. The pointer is guaranteed to be
-        // not null
        CancellationToken {
-            inner: unsafe { NonNull::new_unchecked(Box::into_raw(state)) },
+            inner: Arc::new(tree_node::TreeNode::new()),
        }
    }

-    /// Returns a reference to the utilized `CancellationTokenState`.
-    fn state(&self) -> &CancellationTokenState {
-        // Safety: The state inside a `CancellationToken` is always valid, since
-        // is reference counted
-        unsafe { &*self.inner.as_ptr() }
-    }
-
    /// Creates a `CancellationToken` which will get cancelled whenever the
    /// current token gets cancelled.
    ///
@@ -172,9 +133,9 @@ impl CancellationToken {
    ///
    /// # Examples
    ///
-    /// ```ignore
+    /// ```no_run
    /// use tokio::select;
-    /// use tokio::scope::CancellationToken;
+    /// use tokio_util::sync::CancellationToken;
    ///
    /// #[tokio::main]
    /// async fn main() {
@@ -203,56 +164,8 @@ impl CancellationToken {
    /// }
    /// ```
    pub fn child_token(&self) -> CancellationToken {
-        let inner = self.state();
-
-        // Increment the refcount of this token. It will be referenced by the
-        // child, independent of whether the child is immediately cancelled or
-        // not.
-        let _current_state = inner.increment_refcount(inner.snapshot());
-
-        let mut unpacked_child_state = StateSnapshot {
-            has_parent_ref: true,
-            refcount: 1,
-            cancel_state: CancellationState::NotCancelled,
-        };
-        let mut child_token_state = Box::new(CancellationTokenState::new(
-            Some(self.inner),
-            unpacked_child_state,
-        ));
-
-        {
-            let mut guard = inner.synchronized.lock().unwrap();
-            if guard.is_cancelled {
-                // This task was already cancelled. In this case we should not
-                // insert the child into the list, since it would never get removed
-                // from the list.
-                (*child_token_state.synchronized.lock().unwrap()).is_cancelled = true;
-                unpacked_child_state.cancel_state = CancellationState::Cancelled;
-                // Since it's not in the list, the parent doesn't need to retain
-                // a reference to it.
-                unpacked_child_state.has_parent_ref = false;
-                child_token_state
-                    .state
-                    .store(unpacked_child_state.pack(), Ordering::SeqCst);
-            } else {
-                if let Some(mut first_child) = guard.first_child {
-                    child_token_state.from_parent.next_peer = Some(first_child);
-                    // Safety: We manipulate other child task inside the Mutex
-                    // and retain a parent reference on it. The child token can't
-                    // get invalidated while the Mutex is held.
-                    unsafe {
-                        first_child.as_mut().from_parent.prev_peer =
-                            Some((&mut *child_token_state).into())
-                    };
-                }
-                guard.first_child = Some((&mut *child_token_state).into());
-            }
-        };
-
-        let child_token_ptr = Box::into_raw(child_token_state);
-        // Safety: We just created the pointer from a `Box`
        CancellationToken {
-            inner: unsafe { NonNull::new_unchecked(child_token_ptr) },
+            inner: tree_node::child_node(&self.inner),
        }
    }

@@ -260,24 +173,51 @@ impl CancellationToken {
    /// derived from it.
    ///
    /// This will wake up all tasks which are waiting for cancellation.
+    ///
+    /// Be aware that cancellation is not an atomic operation. It is possible
+    /// for another thread running in parallel with a call to `cancel` to first
+    /// receive `true` from `is_cancelled` on one child node, and then receive
+    /// `false` from `is_cancelled` on another child node. However, once the
+    /// call to `cancel` returns, all child nodes have been fully cancelled.
    pub fn cancel(&self) {
-        self.state().cancel();
+        tree_node::cancel(&self.inner);
    }

-    /// Returns `true` if the `CancellationToken` had been cancelled
+    /// Returns `true` if the `CancellationToken` is cancelled.
    pub fn is_cancelled(&self) -> bool {
-        self.state().is_cancelled()
+        tree_node::is_cancelled(&self.inner)
    }

    /// Returns a `Future` that gets fulfilled when cancellation is requested.
+    ///
+    /// The future will complete immediately if the token is already cancelled
+    /// when this method is called.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe.
    pub fn cancelled(&self) -> WaitForCancellationFuture<'_> {
        WaitForCancellationFuture {
-            cancellation_token: Some(self),
-            wait_node: ListNode::new(WaitQueueEntry::new()),
-            is_registered: false,
+            cancellation_token: self,
+            future: self.inner.notified(),
        }
    }

+    /// Returns a `Future` that gets fulfilled when cancellation is requested.
+    ///
+    /// The future will complete immediately if the token is already cancelled
+    /// when this method is called.
+    ///
+    /// The function takes self by value and returns a future that owns the
+    /// token.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe.
+    pub fn cancelled_owned(self) -> WaitForCancellationFutureOwned {
+        WaitForCancellationFutureOwned::new(self)
+    }
+
    /// Creates a `DropGuard` for this token.
    ///
    /// Returned guard will cancel this token (and all its children) on drop
@@ -285,26 +225,6 @@ impl CancellationToken {
    pub fn drop_guard(self) -> DropGuard {
        DropGuard { inner: Some(self) }
    }
-
-    unsafe fn register(
-        &self,
-        wait_node: &mut ListNode<WaitQueueEntry>,
-        cx: &mut Context<'_>,
-    ) -> Poll<()> {
-        self.state().register(wait_node, cx)
-    }
-
-    fn check_for_cancellation(
-        &self,
-        wait_node: &mut ListNode<WaitQueueEntry>,
-        cx: &mut Context<'_>,
-    ) -> Poll<()> {
-        self.state().check_for_cancellation(wait_node, cx)
-    }
-
-    fn unregister(&self, wait_node: &mut ListNode<WaitQueueEntry>) {
-        self.state().unregister(wait_node)
-    }
 }

 // ===== impl WaitForCancellationFuture =====
@@ -319,560 +239,86 @@ impl<'a> Future for WaitForCancellationFuture<'a> {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
-        // Safety: We do not move anything out of `WaitForCancellationFuture`
-        let mut_self: &mut WaitForCancellationFuture<'_> = unsafe { Pin::get_unchecked_mut(self) };
-
-        let cancellation_token = mut_self
-            .cancellation_token
-            .expect("polled WaitForCancellationFuture after completion");
-
-        let poll_res = if !mut_self.is_registered {
-            // Safety: The `ListNode` is pinned through the Future,
-            // and we will unregister it in `WaitForCancellationFuture::drop`
-            // before the Future is dropped and the memory reference is invalidated.
-            unsafe { cancellation_token.register(&mut mut_self.wait_node, cx) }
-        } else {
-            cancellation_token.check_for_cancellation(&mut mut_self.wait_node, cx)
-        };
-
-        if let Poll::Ready(()) = poll_res {
-            // The cancellation_token was signalled
-            mut_self.cancellation_token = None;
-            // A signalled Token means the Waker won't be enqueued anymore
-            mut_self.is_registered = false;
-            mut_self.wait_node.task = None;
-        } else {
-            // This `Future` and its stored `Waker` stay registered at the
-            // `CancellationToken`
-            mut_self.is_registered = true;
-        }
-
-        poll_res
-    }
-}
-
-impl<'a> Drop for WaitForCancellationFuture<'a> {
-    fn drop(&mut self) {
-        // If this WaitForCancellationFuture has been polled and it was added to the
-        // wait queue at the cancellation_token, it must be removed before dropping.
-        // Otherwise the cancellation_token would access invalid memory.
-        if let Some(token) = self.cancellation_token {
-            if self.is_registered {
-                token.unregister(&mut self.wait_node);
-            }
-        }
-    }
-}
-
-/// Tracks how the future had interacted with the [`CancellationToken`]
-#[derive(Copy, Clone, Debug, PartialEq, Eq)]
-enum PollState {
-    /// The task has never interacted with the [`CancellationToken`].
-    New,
-    /// The task was added to the wait queue at the [`CancellationToken`].
-    Waiting,
-    /// The task has been polled to completion.
-    Done,
-}
-
-/// Tracks the WaitForCancellationFuture waiting state.
-/// Access to this struct is synchronized through the mutex in the CancellationToken.
-struct WaitQueueEntry {
-    /// The task handle of the waiting task
-    task: Option<Waker>,
-    // Current polling state. This state is only updated inside the Mutex of
-    // the CancellationToken.
-    state: PollState,
-}
-
-impl WaitQueueEntry {
-    /// Creates a new WaitQueueEntry
-    fn new() -> WaitQueueEntry {
-        WaitQueueEntry {
-            task: None,
-            state: PollState::New,
-        }
-    }
-}
-
-struct SynchronizedState {
-    waiters: LinkedList<WaitQueueEntry>,
-    first_child: Option<NonNull<CancellationTokenState>>,
-    is_cancelled: bool,
-}
-
-impl SynchronizedState {
-    fn new() -> Self {
-        Self {
-            waiters: LinkedList::new(),
-            first_child: None,
-            is_cancelled: false,
-        }
-    }
-}
-
-/// Information embedded in child tokens which is synchronized through the Mutex
-/// in their parent.
-struct SynchronizedThroughParent {
-    next_peer: Option<NonNull<CancellationTokenState>>,
-    prev_peer: Option<NonNull<CancellationTokenState>>,
-}
-
-/// Possible states of a `CancellationToken`
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-enum CancellationState {
-    NotCancelled = 0,
-    Cancelling = 1,
-    Cancelled = 2,
-}
-
-impl CancellationState {
-    fn pack(self) -> usize {
-        self as usize
-    }
-
-    fn unpack(value: usize) -> Self {
-        match value {
-            0 => CancellationState::NotCancelled,
-            1 => CancellationState::Cancelling,
-            2 => CancellationState::Cancelled,
-            _ => unreachable!("Invalid value"),
-        }
-    }
-}
-
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-struct StateSnapshot {
-    /// The amount of references to this particular CancellationToken.
-    /// `CancellationToken` structs hold these references to a `CancellationTokenState`.
-    /// Also the state is referenced by the state of each child.
-    refcount: usize,
-    /// Whether the state is still referenced by it's parent and can therefore
-    /// not be freed.
-    has_parent_ref: bool,
-    /// Whether the token is cancelled
-    cancel_state: CancellationState,
-}
-
-impl StateSnapshot {
-    /// Packs the snapshot into a `usize`
-    fn pack(self) -> usize {
-        self.refcount << 3 | if self.has_parent_ref { 4 } else { 0 } | self.cancel_state.pack()
-    }
-
-    /// Unpacks the snapshot from a `usize`
-    fn unpack(value: usize) -> Self {
-        let refcount = value >> 3;
-        let has_parent_ref = value & 4 != 0;
-        let cancel_state = CancellationState::unpack(value & 0x03);
-
-        StateSnapshot {
-            refcount,
-            has_parent_ref,
-            cancel_state,
-        }
-    }
-
-    /// Whether this `CancellationTokenState` is still referenced by any
-    /// `CancellationToken`.
-    fn has_refs(&self) -> bool {
-        self.refcount != 0 || self.has_parent_ref
-    }
-}
-
-/// The maximum permitted amount of references to a CancellationToken. This
-/// is derived from the intent to never use more than 32bit in the `Snapshot`.
-const MAX_REFS: u32 = (std::u32::MAX - 7) >> 3;
-
-/// Internal state of the `CancellationToken` pair above
-struct CancellationTokenState {
-    state: AtomicUsize,
-    parent: Option<NonNull<CancellationTokenState>>,
-    from_parent: SynchronizedThroughParent,
-    synchronized: Mutex<SynchronizedState>,
-}
-
-impl CancellationTokenState {
-    fn new(
-        parent: Option<NonNull<CancellationTokenState>>,
-        state: StateSnapshot,
-    ) -> CancellationTokenState {
-        CancellationTokenState {
-            parent,
-            from_parent: SynchronizedThroughParent {
-                prev_peer: None,
-                next_peer: None,
-            },
-            state: AtomicUsize::new(state.pack()),
-            synchronized: Mutex::new(SynchronizedState::new()),
-        }
-    }
-
-    /// Returns a snapshot of the current atomic state of the token
-    fn snapshot(&self) -> StateSnapshot {
-        StateSnapshot::unpack(self.state.load(Ordering::SeqCst))
-    }
-
-    fn atomic_update_state<F>(&self, mut current_state: StateSnapshot, func: F) -> StateSnapshot
-    where
-        F: Fn(StateSnapshot) -> StateSnapshot,
-    {
-        let mut current_packed_state = current_state.pack();
+        let mut this = self.project();
        loop {
-            let next_state = func(current_state);
-            match self.state.compare_exchange(
-                current_packed_state,
-                next_state.pack(),
-                Ordering::SeqCst,
-                Ordering::SeqCst,
-            ) {
-                Ok(_) => {
-                    return next_state;
-                }
-                Err(actual) => {
-                    current_packed_state = actual;
-                    current_state = StateSnapshot::unpack(actual);
-                }
-            }
-        }
-    }
-
-    fn increment_refcount(&self, current_state: StateSnapshot) -> StateSnapshot {
-        self.atomic_update_state(current_state, |mut state: StateSnapshot| {
-            if state.refcount >= MAX_REFS as usize {
-                eprintln!("[ERROR] Maximum reference count for CancellationToken was exceeded");
-                std::process::abort();
-            }
-            state.refcount += 1;
-            state
-        })
-    }
-
-    fn decrement_refcount(&self, current_state: StateSnapshot) -> StateSnapshot {
-        let current_state = self.atomic_update_state(current_state, |mut state: StateSnapshot| {
-            state.refcount -= 1;
-            state
-        });
-
-        // Drop the State if it is not referenced anymore
-        if !current_state.has_refs() {
-            // Safety: `CancellationTokenState` is always stored in refcounted
-            // Boxes
-            let _ = unsafe { Box::from_raw(self as *const Self as *mut Self) };
-        }
-
-        current_state
-    }
-
-    fn remove_parent_ref(&self, current_state: StateSnapshot) -> StateSnapshot {
-        let current_state = self.atomic_update_state(current_state, |mut state: StateSnapshot| {
-            state.has_parent_ref = false;
-            state
-        });
-
-        // Drop the State if it is not referenced anymore
-        if !current_state.has_refs() {
-            // Safety: `CancellationTokenState` is always stored in refcounted
-            // Boxes
-            let _ = unsafe { Box::from_raw(self as *const Self as *mut Self) };
-        }
-
-        current_state
-    }
-
-    /// Unregisters a child from the parent token.
-    /// The child tokens state is not exactly known at this point in time.
-    /// If the parent token is cancelled, the child token gets removed from the
-    /// parents list, and might therefore already have been freed. If the parent
-    /// token is not cancelled, the child token is still valid.
-    fn unregister_child(
-        &mut self,
-        mut child_state: NonNull<CancellationTokenState>,
-        current_child_state: StateSnapshot,
-    ) {
-        let removed_child = {
-            // Remove the child toke from the parents linked list
-            let mut guard = self.synchronized.lock().unwrap();
-            if !guard.is_cancelled {
-                // Safety: Since the token was not cancelled, the child must
-                // still be in the list and valid.
-                let mut child_state = unsafe { child_state.as_mut() };
-                debug_assert!(child_state.snapshot().has_parent_ref);
-
-                if guard.first_child == Some(child_state.into()) {
-                    guard.first_child = child_state.from_parent.next_peer;
-                }
-                // Safety: If peers wouldn't be valid anymore, they would try
-                // to remove themselves from the list. This would require locking
-                // the Mutex that we currently own.
-                unsafe {
-                    if let Some(mut prev_peer) = child_state.from_parent.prev_peer {
-                        prev_peer.as_mut().from_parent.next_peer =
-                            child_state.from_parent.next_peer;
-                    }
-                    if let Some(mut next_peer) = child_state.from_parent.next_peer {
-                        next_peer.as_mut().from_parent.prev_peer =
-                            child_state.from_parent.prev_peer;
-                    }
-                }
-                child_state.from_parent.prev_peer = None;
-                child_state.from_parent.next_peer = None;
-
-                // The child is no longer referenced by the parent, since we were able
-                // to remove its reference from the parents list.
-                true
-            } else {
-                // Do not touch the linked list anymore. If the parent is cancelled
-                // it will move all childs outside of the Mutex and manipulate
-                // the pointers there. Manipulating the pointers here too could
-                // lead to races. Therefore leave them just as as and let the
-                // parent deal with it. The parent will make sure to retain a
-                // reference to this state as long as it manipulates the list
-                // pointers. Therefore the pointers are not dangling.
-                false
-            }
-        };
-
-        if removed_child {
-            // If the token removed itself from the parents list, it can reset
-            // the parent ref status. If it is isn't able to do so, because the
-            // parent removed it from the list, there is no need to do this.
-            // The parent ref acts as as another reference count. Therefore
-            // removing this reference can free the object.
-            // Safety: The token was in the list. This means the parent wasn't
-            // cancelled before, and the token must still be alive.
-            unsafe { child_state.as_mut().remove_parent_ref(current_child_state) };
-        }
-
-        // Decrement the refcount on the parent and free it if necessary
-        self.decrement_refcount(self.snapshot());
-    }
-
-    fn cancel(&self) {
-        // Move the state of the CancellationToken from `NotCancelled` to `Cancelling`
-        let mut current_state = self.snapshot();
-
-        let state_after_cancellation = loop {
-            if current_state.cancel_state != CancellationState::NotCancelled {
-                // Another task already initiated the cancellation
-                return;
+            if this.cancellation_token.is_cancelled() {
+                return Poll::Ready(());
            }

-            let mut next_state = current_state;
-            next_state.cancel_state = CancellationState::Cancelling;
-            match self.state.compare_exchange(
-                current_state.pack(),
-                next_state.pack(),
-                Ordering::SeqCst,
-                Ordering::SeqCst,
-            ) {
-                Ok(_) => break next_state,
-                Err(actual) => current_state = StateSnapshot::unpack(actual),
+            // No wakeups can be lost here because there is always a call to
+            // `is_cancelled` between the creation of the future and the call to
+            // `poll`, and the code that sets the cancelled flag does so before
+            // waking the `Notified`.
+            if this.future.as_mut().poll(cx).is_pending() {
+                return Poll::Pending;
            }
-        };

-        // This task cancelled the token
-
-        // Take the task list out of the Token
-        // We do not want to cancel child token inside this lock. If one of the
-        // child tasks would have additional child tokens, we would recursively
-        // take locks.
-
-        // Doing this action has an impact if the child token is dropped concurrently:
-        // It will try to deregister itself from the parent task, but can not find
-        // itself in the task list anymore. Therefore it needs to assume the parent
-        // has extracted the list and will process it. It may not modify the list.
-        // This is OK from a memory safety perspective, since the parent still
-        // retains a reference to the child task until it finished iterating over
-        // it.
-
-        let mut first_child = {
-            let mut guard = self.synchronized.lock().unwrap();
-            // Save the cancellation also inside the Mutex
-            // This allows child tokens which want to detach themselves to detect
-            // that this is no longer required since the parent cleared the list.
-            guard.is_cancelled = true;
-
-            // Wakeup all waiters
-            // This happens inside the lock to make cancellation reliable
-            // If we would access waiters outside of the lock, the pointers
-            // may no longer be valid.
-            // Typically this shouldn't be an issue, since waking a task should
-            // only move it from the blocked into the ready state and not have
-            // further side effects.
-
-            // Use a reverse iterator, so that the oldest waiter gets
-            // scheduled first
-            guard.waiters.reverse_drain(|waiter| {
-                // We are not allowed to move the `Waker` out of the list node.
-                // The `Future` relies on the fact that the old `Waker` stays there
-                // as long as the `Future` has not completed in order to perform
-                // the `will_wake()` check.
-                // Therefore `wake_by_ref` is used instead of `wake()`
-                if let Some(handle) = &mut waiter.task {
-                    handle.wake_by_ref();
-                }
-                // Mark the waiter to have been removed from the list.
-                waiter.state = PollState::Done;
-            });
-
-            guard.first_child.take()
-        };
-
-        while let Some(mut child) = first_child {
-            // Safety: We know this is a valid pointer since it is in our child pointer
-            // list. It can't have been freed in between, since we retain a a reference
-            // to each child.
-            let mut_child = unsafe { child.as_mut() };
-
-            // Get the next child and clean up list pointers
-            first_child = mut_child.from_parent.next_peer;
-            mut_child.from_parent.prev_peer = None;
-            mut_child.from_parent.next_peer = None;
-
-            // Cancel the child task
-            mut_child.cancel();
-
-            // Drop the parent reference. This `CancellationToken` is not interested
-            // in interacting with the child anymore.
-            // This is ONLY allowed once we promised not to touch the state anymore
-            // after this interaction.
-            mut_child.remove_parent_ref(mut_child.snapshot());
+            this.future.set(this.cancellation_token.inner.notified());
+        }
+    }
+}
+
+// ===== impl WaitForCancellationFutureOwned =====
+
+impl core::fmt::Debug for WaitForCancellationFutureOwned {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("WaitForCancellationFutureOwned").finish()
+    }
+}
+
+impl WaitForCancellationFutureOwned {
+    fn new(cancellation_token: CancellationToken) -> Self {
+        WaitForCancellationFutureOwned {
+            // cancellation_token holds a heap allocation and is guaranteed to have a
+            // stable deref, thus it would be ok to move the cancellation_token while
+            // the future holds a reference to it.
+            //
+            // # Safety
+            //
+            // cancellation_token is dropped after future due to the field ordering.
+            future: unsafe { Self::new_future(&cancellation_token) },
+            cancellation_token,
+        }
+    }
+
+    /// # Safety
+    /// The returned future must be destroyed before the cancellation token is
+    /// destroyed.
+    unsafe fn new_future(
+        cancellation_token: &CancellationToken,
+    ) -> tokio::sync::futures::Notified<'static> {
+        let inner_ptr = Arc::as_ptr(&cancellation_token.inner);
+        // SAFETY: The `Arc::as_ptr` method guarantees that `inner_ptr` remains
+        // valid until the strong count of the Arc drops to zero, and the caller
+        // guarantees that they will drop the future before that happens.
+        (*inner_ptr).notified()
+    }
+}
+
+impl Future for WaitForCancellationFutureOwned {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        let mut this = self.project();
+
+        loop {
+            if this.cancellation_token.is_cancelled() {
+                return Poll::Ready(());
+            }
+
+            // No wakeups can be lost here because there is always a call to
+            // `is_cancelled` between the creation of the future and the call to
+            // `poll`, and the code that sets the cancelled flag does so before
+            // waking the `Notified`.
+            if this.future.as_mut().poll(cx).is_pending() {
+                return Poll::Pending;
+            }
+
+            // # Safety
+            //
+            // cancellation_token is dropped after future due to the field ordering.
+            this.future
+                .set(unsafe { Self::new_future(this.cancellation_token) });
        }
-
-        // The cancellation has completed
-        // At this point in time tasks which registered a wait node can be sure
-        // that this wait node already had been dequeued from the list without
-        // needing to inspect the list.
-        self.atomic_update_state(state_after_cancellation, |mut state| {
-            state.cancel_state = CancellationState::Cancelled;
-            state
-        });
-    }
-
-    /// Returns `true` if the `CancellationToken` had been cancelled
-    fn is_cancelled(&self) -> bool {
-        let current_state = self.snapshot();
-        current_state.cancel_state != CancellationState::NotCancelled
-    }
-
-    /// Registers a waiting task at the `CancellationToken`.
-    /// Safety: This method is only safe as long as the waiting waiting task
-    /// will properly unregister the wait node before it gets moved.
-    unsafe fn register(
-        &self,
-        wait_node: &mut ListNode<WaitQueueEntry>,
-        cx: &mut Context<'_>,
-    ) -> Poll<()> {
-        debug_assert_eq!(PollState::New, wait_node.state);
-        let current_state = self.snapshot();
-
-        // Perform an optimistic cancellation check before. This is not strictly
-        // necessary since we also check for cancellation in the Mutex, but
-        // reduces the necessary work to be performed for tasks which already
-        // had been cancelled.
-        if current_state.cancel_state != CancellationState::NotCancelled {
-            return Poll::Ready(());
-        }
-
-        // So far the token is not cancelled. However it could be cancelled before
-        // we get the chance to store the `Waker`. Therefore we need to check
-        // for cancellation again inside the mutex.
-        let mut guard = self.synchronized.lock().unwrap();
-        if guard.is_cancelled {
-            // Cancellation was signalled
-            wait_node.state = PollState::Done;
-            Poll::Ready(())
-        } else {
-            // Added the task to the wait queue
-            wait_node.task = Some(cx.waker().clone());
-            wait_node.state = PollState::Waiting;
-            guard.waiters.add_front(wait_node);
-            Poll::Pending
-        }
-    }
-
-    fn check_for_cancellation(
-        &self,
-        wait_node: &mut ListNode<WaitQueueEntry>,
-        cx: &mut Context<'_>,
-    ) -> Poll<()> {
-        debug_assert!(
-            wait_node.task.is_some(),
-            "Method can only be called after task had been registered"
-        );
-
-        let current_state = self.snapshot();
-
-        if current_state.cancel_state != CancellationState::NotCancelled {
-            // If the cancellation had been fully completed we know that our `Waker`
-            // is no longer registered at the `CancellationToken`.
-            // Otherwise the cancel call may or may not yet have iterated
-            // through the waiters list and removed the wait nodes.
-            // If it hasn't yet, we need to remove it. Otherwise an attempt to
-            // reuse the `wait_node´ might get freed due to the `WaitForCancellationFuture`
-            // getting dropped before the cancellation had interacted with it.
-            if current_state.cancel_state != CancellationState::Cancelled {
-                self.unregister(wait_node);
-            }
-            Poll::Ready(())
-        } else {
-            // Check if we need to swap the `Waker`. This will make the check more
-            // expensive, since the `Waker` is synchronized through the Mutex.
-            // If we don't need to perform a `Waker` update, an atomic check for
-            // cancellation is sufficient.
-            let need_waker_update = wait_node
-                .task
-                .as_ref()
-                .map(|waker| waker.will_wake(cx.waker()))
-                .unwrap_or(true);
-
-            if need_waker_update {
-                let guard = self.synchronized.lock().unwrap();
-                if guard.is_cancelled {
-                    // Cancellation was signalled. Since this cancellation signal
-                    // is set inside the Mutex, the old waiter must already have
-                    // been removed from the waiting list
-                    debug_assert_eq!(PollState::Done, wait_node.state);
-                    wait_node.task = None;
-                    Poll::Ready(())
-                } else {
-                    // The WaitForCancellationFuture is already in the queue.
-                    // The CancellationToken can't have been cancelled,
-                    // since this would change the is_cancelled flag inside the mutex.
-                    // Therefore we just have to update the Waker. A follow-up
-                    // cancellation will always use the new waker.
-                    wait_node.task = Some(cx.waker().clone());
-                    Poll::Pending
-                }
-            } else {
-                // Do nothing. If the token gets cancelled, this task will get
-                // woken again and can fetch the cancellation.
-                Poll::Pending
-            }
-        }
-    }
-
-    fn unregister(&self, wait_node: &mut ListNode<WaitQueueEntry>) {
-        debug_assert!(
-            wait_node.task.is_some(),
-            "waiter can not be active without task"
-        );
-
-        let mut guard = self.synchronized.lock().unwrap();
-        // WaitForCancellationFuture only needs to get removed if it has been added to
-        // the wait queue of the CancellationToken.
-        // This has happened in the PollState::Waiting case.
-        if let PollState::Waiting = wait_node.state {
-            // Safety: Due to the state, we know that the node must be part
-            // of the waiter list
-            if !unsafe { guard.waiters.remove(wait_node) } {
-                // Panic if the address isn't found. This can only happen if the contract was
-                // violated, e.g. the WaitQueueEntry got moved after the initial poll.
-                panic!("Future could not be removed from wait queue");
-            }
-            wait_node.state = PollState::Done;
-        }
-        wait_node.task = None;
    }
 }
@@ -0,0 +1,373 @@
+//! This mod provides the logic for the inner tree structure of the CancellationToken.
+//!
+//! CancellationTokens are only light handles with references to TreeNode.
+//! All the logic is actually implemented in the TreeNode.
+//!
+//! A TreeNode is part of the cancellation tree and may have one parent and an arbitrary number of
+//! children.
+//!
+//! A TreeNode can receive the request to perform a cancellation through a CancellationToken.
+//! This cancellation request will cancel the node and all of its descendants.
+//!
+//! As soon as a node cannot get cancelled any more (because it was already cancelled or it has no
+//! more CancellationTokens pointing to it any more), it gets removed from the tree, to keep the
+//! tree as small as possible.
+//!
+//! # Invariants
+//!
+//! Those invariants shall be true at any time.
+//!
+//! 1. A node that has no parents and no handles can no longer be cancelled.
+//!     This is important during both cancellation and refcounting.
+//!
+//! 2. If node B *is* or *was* a child of node A, then node B was created *after* node A.
+//!     This is important for deadlock safety, as it is used for lock order.
+//!     Node B can only become the child of node A in two ways:
+//!         - being created with `child_node()`, in which case it is trivially true that
+//!           node A already existed when node B was created
+//!         - being moved A->C->B to A->B because node C was removed in `decrease_handle_refcount()`
+//!           or `cancel()`. In this case the invariant still holds, as B was younger than C, and C
+//!           was younger than A, therefore B is also younger than A.
+//!
+//! 3. If two nodes are both unlocked and node A is the parent of node B, then node B is a child of
+//!    node A. It is important to always restore that invariant before dropping the lock of a node.
+//!
+//! # Deadlock safety
+//!
+//! We always lock in the order of creation time. We can prove this through invariant #2.
+//! Specifically, through invariant #2, we know that we always have to lock a parent
+//! before its child.
+//!
+use crate::loom::sync::{Arc, Mutex, MutexGuard};
+
+/// A node of the cancellation tree structure
+///
+/// The actual data it holds is wrapped inside a mutex for synchronization.
+pub(crate) struct TreeNode {
+    inner: Mutex<Inner>,
+    waker: tokio::sync::Notify,
+}
+impl TreeNode {
+    pub(crate) fn new() -> Self {
+        Self {
+            inner: Mutex::new(Inner {
+                parent: None,
+                parent_idx: 0,
+                children: vec![],
+                is_cancelled: false,
+                num_handles: 1,
+            }),
+            waker: tokio::sync::Notify::new(),
+        }
+    }
+
+    pub(crate) fn notified(&self) -> tokio::sync::futures::Notified<'_> {
+        self.waker.notified()
+    }
+}
+
+/// The data contained inside a TreeNode.
+///
+/// This struct exists so that the data of the node can be wrapped
+/// in a Mutex.
+struct Inner {
+    parent: Option<Arc<TreeNode>>,
+    parent_idx: usize,
+    children: Vec<Arc<TreeNode>>,
+    is_cancelled: bool,
+    num_handles: usize,
+}
+
+/// Returns whether or not the node is cancelled
+pub(crate) fn is_cancelled(node: &Arc<TreeNode>) -> bool {
+    node.inner.lock().unwrap().is_cancelled
+}
+
+/// Creates a child node
+pub(crate) fn child_node(parent: &Arc<TreeNode>) -> Arc<TreeNode> {
+    let mut locked_parent = parent.inner.lock().unwrap();
+
+    // Do not register as child if we are already cancelled.
+    // Cancelled trees can never be uncancelled and therefore
+    // need no connection to parents or children any more.
+    if locked_parent.is_cancelled {
+        return Arc::new(TreeNode {
+            inner: Mutex::new(Inner {
+                parent: None,
+                parent_idx: 0,
+                children: vec![],
+                is_cancelled: true,
+                num_handles: 1,
+            }),
+            waker: tokio::sync::Notify::new(),
+        });
+    }
+
+    let child = Arc::new(TreeNode {
+        inner: Mutex::new(Inner {
+            parent: Some(parent.clone()),
+            parent_idx: locked_parent.children.len(),
+            children: vec![],
+            is_cancelled: false,
+            num_handles: 1,
+        }),
+        waker: tokio::sync::Notify::new(),
+    });
+
+    locked_parent.children.push(child.clone());
+
+    child
+}
+
+/// Disconnects the given parent from all of its children.
+///
+/// Takes a reference to [Inner] to make sure the parent is already locked.
+fn disconnect_children(node: &mut Inner) {
+    for child in std::mem::take(&mut node.children) {
+        let mut locked_child = child.inner.lock().unwrap();
+        locked_child.parent_idx = 0;
+        locked_child.parent = None;
+    }
+}
+
+/// Figures out the parent of the node and locks the node and its parent atomically.
+///
+/// The basic principle of preventing deadlocks in the tree is
+/// that we always lock the parent first, and then the child.
+/// For more info look at *deadlock safety* and *invariant #2*.
+///
+/// Sadly, it's impossible to figure out the parent of a node without
+/// locking it. To then achieve locking order consistency, the node
+/// has to be unlocked before the parent gets locked.
+/// This leaves a small window where we already assume that we know the parent,
+/// but neither the parent nor the node is locked. Therefore, the parent could change.
+///
+/// To prevent that this problem leaks into the rest of the code, it is abstracted
+/// in this function.
+///
+/// The locked child and optionally its locked parent, if a parent exists, get passed
+/// to the `func` argument via (node, None) or (node, Some(parent)).
+fn with_locked_node_and_parent<F, Ret>(node: &Arc<TreeNode>, func: F) -> Ret
+where
+    F: FnOnce(MutexGuard<'_, Inner>, Option<MutexGuard<'_, Inner>>) -> Ret,
+{
+    let mut potential_parent = {
+        let locked_node = node.inner.lock().unwrap();
+        match locked_node.parent.clone() {
+            Some(parent) => parent,
+            // If we locked the node and its parent is `None`, we are in a valid state
+            // and can return.
+            None => return func(locked_node, None),
+        }
+    };
+
+    loop {
+        // Deadlock safety:
+        //
+        // Due to invariant #2, we know that we have to lock the parent first, and then the child.
+        // This is true even if the potential_parent is no longer the current parent or even its
+        // sibling, as the invariant still holds.
+        let locked_parent = potential_parent.inner.lock().unwrap();
+        let locked_node = node.inner.lock().unwrap();
+
+        let actual_parent = match locked_node.parent.clone() {
+            Some(parent) => parent,
+            // If we locked the node and its parent is `None`, we are in a valid state
+            // and can return.
+            None => {
+                // Was the wrong parent, so unlock it before calling `func`
+                drop(locked_parent);
+                return func(locked_node, None);
+            }
+        };
+
+        // Loop until we managed to lock both the node and its parent
+        if Arc::ptr_eq(&actual_parent, &potential_parent) {
+            return func(locked_node, Some(locked_parent));
+        }
+
+        // Drop locked_parent before reassigning to potential_parent,
+        // as potential_parent is borrowed in it
+        drop(locked_node);
+        drop(locked_parent);
+
+        potential_parent = actual_parent;
+    }
+}
+
+/// Moves all children from `node` to `parent`.
+///
+/// `parent` MUST have been a parent of the node when they both got locked,
+/// otherwise there is a potential for a deadlock as invariant #2 would be violated.
+///
+/// To acquire the locks for node and parent, use [with_locked_node_and_parent].
+fn move_children_to_parent(node: &mut Inner, parent: &mut Inner) {
+    // Pre-allocate in the parent, for performance
+    parent.children.reserve(node.children.len());
+
+    for child in std::mem::take(&mut node.children) {
+        {
+            let mut child_locked = child.inner.lock().unwrap();
+            child_locked.parent = node.parent.clone();
+            child_locked.parent_idx = parent.children.len();
+        }
+        parent.children.push(child);
+    }
+}
+
+/// Removes a child from the parent.
+///
+/// `parent` MUST be the parent of `node`.
+/// To acquire the locks for node and parent, use [with_locked_node_and_parent].
+fn remove_child(parent: &mut Inner, mut node: MutexGuard<'_, Inner>) {
+    // Query the position from where to remove a node
+    let pos = node.parent_idx;
+    node.parent = None;
+    node.parent_idx = 0;
+
+    // Unlock node, so that only one child at a time is locked.
+    // Otherwise we would violate the lock order (see 'deadlock safety') as we
+    // don't know the creation order of the child nodes
+    drop(node);
+
+    // If `node` is the last element in the list, we don't need any swapping
+    if parent.children.len() == pos + 1 {
+        parent.children.pop().unwrap();
+    } else {
+        // If `node` is not the last element in the list, we need to
+        // replace it with the last element
+        let replacement_child = parent.children.pop().unwrap();
+        replacement_child.inner.lock().unwrap().parent_idx = pos;
+        parent.children[pos] = replacement_child;
+    }
+
+    let len = parent.children.len();
+    if 4 * len <= parent.children.capacity() {
+        // equal to:
+        //    parent.children.shrink_to(2 * len);
+        // but shrink_to was not yet stabilized in our minimal compatible version
+        let old_children = std::mem::replace(&mut parent.children, Vec::with_capacity(2 * len));
+        parent.children.extend(old_children);
+    }
+}
+
+/// Increases the reference count of handles.
+pub(crate) fn increase_handle_refcount(node: &Arc<TreeNode>) {
+    let mut locked_node = node.inner.lock().unwrap();
+
+    // Once no handles are left over, the node gets detached from the tree.
+    // There should never be a new handle once all handles are dropped.
+    assert!(locked_node.num_handles > 0);
+
+    locked_node.num_handles += 1;
+}
+
+/// Decreases the reference count of handles.
+///
+/// Once no handle is left, we can remove the node from the
+/// tree and connect its parent directly to its children.
+pub(crate) fn decrease_handle_refcount(node: &Arc<TreeNode>) {
+    let num_handles = {
+        let mut locked_node = node.inner.lock().unwrap();
+        locked_node.num_handles -= 1;
+        locked_node.num_handles
+    };
+
+    if num_handles == 0 {
+        with_locked_node_and_parent(node, |mut node, parent| {
+            // Remove the node from the tree
+            match parent {
+                Some(mut parent) => {
+                    // As we want to remove ourselves from the tree,
+                    // we have to move the children to the parent, so that
+                    // they still receive the cancellation event without us.
+                    // Moving them does not violate invariant #1.
+                    move_children_to_parent(&mut node, &mut parent);
+
+                    // Remove the node from the parent
+                    remove_child(&mut parent, node);
+                }
+                None => {
+                    // Due to invariant #1, we can assume that our
+                    // children can no longer be cancelled through us.
+                    // (as we now have neither a parent nor handles)
+                    // Therefore we can disconnect them.
+                    disconnect_children(&mut node);
+                }
+            }
+        });
+    }
+}
+
+/// Cancels a node and its children.
+pub(crate) fn cancel(node: &Arc<TreeNode>) {
+    let mut locked_node = node.inner.lock().unwrap();
+
+    if locked_node.is_cancelled {
+        return;
+    }
+
+    // One by one, adopt grandchildren and then cancel and detach the child
+    while let Some(child) = locked_node.children.pop() {
+        // This can't deadlock because the mutex we are already
+        // holding is the parent of child.
+        let mut locked_child = child.inner.lock().unwrap();
+
+        // Detach the child from node
+        // No need to modify node.children, as the child already got removed with `.pop`
+        locked_child.parent = None;
+        locked_child.parent_idx = 0;
+
+        // If child is already cancelled, detaching is enough
+        if locked_child.is_cancelled {
+            continue;
+        }
+
+        // Cancel or adopt grandchildren
+        while let Some(grandchild) = locked_child.children.pop() {
+            // This can't deadlock because the two mutexes we are already
+            // holding is the parent and grandparent of grandchild.
+            let mut locked_grandchild = grandchild.inner.lock().unwrap();
+
+            // Detach the grandchild
+            locked_grandchild.parent = None;
+            locked_grandchild.parent_idx = 0;
+
+            // If grandchild is already cancelled, detaching is enough
+            if locked_grandchild.is_cancelled {
+                continue;
+            }
+
+            // For performance reasons, only adopt grandchildren that have children.
+            // Otherwise, just cancel them right away, no need for another iteration.
+            if locked_grandchild.children.is_empty() {
+                // Cancel the grandchild
+                locked_grandchild.is_cancelled = true;
+                locked_grandchild.children = Vec::new();
+                drop(locked_grandchild);
+                grandchild.waker.notify_waiters();
+            } else {
+                // Otherwise, adopt grandchild
+                locked_grandchild.parent = Some(node.clone());
+                locked_grandchild.parent_idx = locked_node.children.len();
+                drop(locked_grandchild);
+                locked_node.children.push(grandchild);
+            }
+        }
+
+        // Cancel the child
+        locked_child.is_cancelled = true;
+        locked_child.children = Vec::new();
+        drop(locked_child);
+        child.waker.notify_waiters();
+
+        // Now the child is cancelled and detached and all its children are adopted.
+        // Just continue until all (including adopted) children are cancelled and detached.
+    }
+
+    // Cancel the node itself.
+    locked_node.is_cancelled = true;
+    locked_node.children = Vec::new();
+    drop(locked_node);
+    node.waker.notify_waiters();
+}
@@ -1,788 +0,0 @@
-//! An intrusive double linked list of data
-
-#![allow(dead_code, unreachable_pub)]
-
-use core::{
-    marker::PhantomPinned,
-    ops::{Deref, DerefMut},
-    ptr::NonNull,
-};
-
-/// A node which carries data of type `T` and is stored in an intrusive list
-#[derive(Debug)]
-pub struct ListNode<T> {
-    /// The previous node in the list. `None` if there is no previous node.
-    prev: Option<NonNull<ListNode<T>>>,
-    /// The next node in the list. `None` if there is no previous node.
-    next: Option<NonNull<ListNode<T>>>,
-    /// The data which is associated to this list item
-    data: T,
-    /// Prevents `ListNode`s from being `Unpin`. They may never be moved, since
-    /// the list semantics require addresses to be stable.
-    _pin: PhantomPinned,
-}
-
-impl<T> ListNode<T> {
-    /// Creates a new node with the associated data
-    pub fn new(data: T) -> ListNode<T> {
-        Self {
-            prev: None,
-            next: None,
-            data,
-            _pin: PhantomPinned,
-        }
-    }
-}
-
-impl<T> Deref for ListNode<T> {
-    type Target = T;
-
-    fn deref(&self) -> &T {
-        &self.data
-    }
-}
-
-impl<T> DerefMut for ListNode<T> {
-    fn deref_mut(&mut self) -> &mut T {
-        &mut self.data
-    }
-}
-
-/// An intrusive linked list of nodes, where each node carries associated data
-/// of type `T`.
-#[derive(Debug)]
-pub struct LinkedList<T> {
-    head: Option<NonNull<ListNode<T>>>,
-    tail: Option<NonNull<ListNode<T>>>,
-}
-
-impl<T> LinkedList<T> {
-    /// Creates an empty linked list
-    pub fn new() -> Self {
-        LinkedList::<T> {
-            head: None,
-            tail: None,
-        }
-    }
-
-    /// Adds a node at the front of the linked list.
-    /// Safety: This function is only safe as long as `node` is guaranteed to
-    /// get removed from the list before it gets moved or dropped.
-    /// In addition to this `node` may not be added to another other list before
-    /// it is removed from the current one.
-    pub unsafe fn add_front(&mut self, node: &mut ListNode<T>) {
-        node.next = self.head;
-        node.prev = None;
-        if let Some(mut head) = self.head {
-            head.as_mut().prev = Some(node.into())
-        };
-        self.head = Some(node.into());
-        if self.tail.is_none() {
-            self.tail = Some(node.into());
-        }
-    }
-
-    /// Inserts a node into the list in a way that the list keeps being sorted.
-    /// Safety: This function is only safe as long as `node` is guaranteed to
-    /// get removed from the list before it gets moved or dropped.
-    /// In addition to this `node` may not be added to another other list before
-    /// it is removed from the current one.
-    pub unsafe fn add_sorted(&mut self, node: &mut ListNode<T>)
-    where
-        T: PartialOrd,
-    {
-        if self.head.is_none() {
-            // First node in the list
-            self.head = Some(node.into());
-            self.tail = Some(node.into());
-            return;
-        }
-
-        let mut prev: Option<NonNull<ListNode<T>>> = None;
-        let mut current = self.head;
-
-        while let Some(mut current_node) = current {
-            if node.data < current_node.as_ref().data {
-                // Need to insert before the current node
-                current_node.as_mut().prev = Some(node.into());
-                match prev {
-                    Some(mut prev) => {
-                        prev.as_mut().next = Some(node.into());
-                    }
-                    None => {
-                        // We are inserting at the beginning of the list
-                        self.head = Some(node.into());
-                    }
-                }
-                node.next = current;
-                node.prev = prev;
-                return;
-            }
-            prev = current;
-            current = current_node.as_ref().next;
-        }
-
-        // We looped through the whole list and the nodes data is bigger or equal
-        // than everything we found up to now.
-        // Insert at the end. Since we checked before that the list isn't empty,
-        // tail always has a value.
-        node.prev = self.tail;
-        node.next = None;
-        self.tail.as_mut().unwrap().as_mut().next = Some(node.into());
-        self.tail = Some(node.into());
-    }
-
-    /// Returns the first node in the linked list without removing it from the list
-    /// The function is only safe as long as valid pointers are stored inside
-    /// the linked list.
-    /// The returned pointer is only guaranteed to be valid as long as the list
-    /// is not mutated
-    pub fn peek_first(&self) -> Option<&mut ListNode<T>> {
-        // Safety: When the node was inserted it was promised that it is alive
-        // until it gets removed from the list.
-        // The returned node has a pointer which constrains it to the lifetime
-        // of the list. This is ok, since the Node is supposed to outlive
-        // its insertion in the list.
-        unsafe {
-            self.head
-                .map(|mut node| &mut *(node.as_mut() as *mut ListNode<T>))
-        }
-    }
-
-    /// Returns the last node in the linked list without removing it from the list
-    /// The function is only safe as long as valid pointers are stored inside
-    /// the linked list.
-    /// The returned pointer is only guaranteed to be valid as long as the list
-    /// is not mutated
-    pub fn peek_last(&self) -> Option<&mut ListNode<T>> {
-        // Safety: When the node was inserted it was promised that it is alive
-        // until it gets removed from the list.
-        // The returned node has a pointer which constrains it to the lifetime
-        // of the list. This is ok, since the Node is supposed to outlive
-        // its insertion in the list.
-        unsafe {
-            self.tail
-                .map(|mut node| &mut *(node.as_mut() as *mut ListNode<T>))
-        }
-    }
-
-    /// Removes the first node from the linked list
-    pub fn remove_first(&mut self) -> Option<&mut ListNode<T>> {
-        #![allow(clippy::debug_assert_with_mut_call)]
-
-        // Safety: When the node was inserted it was promised that it is alive
-        // until it gets removed from the list
-        unsafe {
-            let mut head = self.head?;
-            self.head = head.as_mut().next;
-
-            let first_ref = head.as_mut();
-            match first_ref.next {
-                None => {
-                    // This was the only node in the list
-                    debug_assert_eq!(Some(first_ref.into()), self.tail);
-                    self.tail = None;
-                }
-                Some(mut next) => {
-                    next.as_mut().prev = None;
-                }
-            }
-
-            first_ref.prev = None;
-            first_ref.next = None;
-            Some(&mut *(first_ref as *mut ListNode<T>))
-        }
-    }
-
-    /// Removes the last node from the linked list and returns it
-    pub fn remove_last(&mut self) -> Option<&mut ListNode<T>> {
-        #![allow(clippy::debug_assert_with_mut_call)]
-
-        // Safety: When the node was inserted it was promised that it is alive
-        // until it gets removed from the list
-        unsafe {
-            let mut tail = self.tail?;
-            self.tail = tail.as_mut().prev;
-
-            let last_ref = tail.as_mut();
-            match last_ref.prev {
-                None => {
-                    // This was the last node in the list
-                    debug_assert_eq!(Some(last_ref.into()), self.head);
-                    self.head = None;
-                }
-                Some(mut prev) => {
-                    prev.as_mut().next = None;
-                }
-            }
-
-            last_ref.prev = None;
-            last_ref.next = None;
-            Some(&mut *(last_ref as *mut ListNode<T>))
-        }
-    }
-
-    /// Returns whether the linked list does not contain any node
-    pub fn is_empty(&self) -> bool {
-        if self.head.is_some() {
-            return false;
-        }
-
-        debug_assert!(self.tail.is_none());
-        true
-    }
-
-    /// Removes the given `node` from the linked list.
-    /// Returns whether the `node` was removed.
-    /// It is also only safe if it is known that the `node` is either part of this
-    /// list, or of no list at all. If `node` is part of another list, the
-    /// behavior is undefined.
-    pub unsafe fn remove(&mut self, node: &mut ListNode<T>) -> bool {
-        #![allow(clippy::debug_assert_with_mut_call)]
-
-        match node.prev {
-            None => {
-                // This might be the first node in the list. If it is not, the
-                // node is not in the list at all. Since our precondition is that
-                // the node must either be in this list or in no list, we check that
-                // the node is really in no list.
-                if self.head != Some(node.into()) {
-                    debug_assert!(node.next.is_none());
-                    return false;
-                }
-                self.head = node.next;
-            }
-            Some(mut prev) => {
-                debug_assert_eq!(prev.as_ref().next, Some(node.into()));
-                prev.as_mut().next = node.next;
-            }
-        }
-
-        match node.next {
-            None => {
-                // This must be the last node in our list. Otherwise the list
-                // is inconsistent.
-                debug_assert_eq!(self.tail, Some(node.into()));
-                self.tail = node.prev;
-            }
-            Some(mut next) => {
-                debug_assert_eq!(next.as_mut().prev, Some(node.into()));
-                next.as_mut().prev = node.prev;
-            }
-        }
-
-        node.next = None;
-        node.prev = None;
-
-        true
-    }
-
-    /// Drains the list iby calling a callback on each list node
-    ///
-    /// The method does not return an iterator since stopping or deferring
-    /// draining the list is not permitted. If the method would push nodes to
-    /// an iterator we could not guarantee that the nodes do not get utilized
-    /// after having been removed from the list anymore.
-    pub fn drain<F>(&mut self, mut func: F)
-    where
-        F: FnMut(&mut ListNode<T>),
-    {
-        let mut current = self.head;
-        self.head = None;
-        self.tail = None;
-
-        while let Some(mut node) = current {
-            // Safety: The nodes have not been removed from the list yet and must
-            // therefore contain valid data. The nodes can also not be added to
-            // the list again during iteration, since the list is mutably borrowed.
-            unsafe {
-                let node_ref = node.as_mut();
-                current = node_ref.next;
-
-                node_ref.next = None;
-                node_ref.prev = None;
-
-                // Note: We do not reset the pointers from the next element in the
-                // list to the current one since we will iterate over the whole
-                // list anyway, and therefore clean up all pointers.
-
-                func(node_ref);
-            }
-        }
-    }
-
-    /// Drains the list in reverse order by calling a callback on each list node
-    ///
-    /// The method does not return an iterator since stopping or deferring
-    /// draining the list is not permitted. If the method would push nodes to
-    /// an iterator we could not guarantee that the nodes do not get utilized
-    /// after having been removed from the list anymore.
-    pub fn reverse_drain<F>(&mut self, mut func: F)
-    where
-        F: FnMut(&mut ListNode<T>),
-    {
-        let mut current = self.tail;
-        self.head = None;
-        self.tail = None;
-
-        while let Some(mut node) = current {
-            // Safety: The nodes have not been removed from the list yet and must
-            // therefore contain valid data. The nodes can also not be added to
-            // the list again during iteration, since the list is mutably borrowed.
-            unsafe {
-                let node_ref = node.as_mut();
-                current = node_ref.prev;
-
-                node_ref.next = None;
-                node_ref.prev = None;
-
-                // Note: We do not reset the pointers from the next element in the
-                // list to the current one since we will iterate over the whole
-                // list anyway, and therefore clean up all pointers.
-
-                func(node_ref);
-            }
-        }
-    }
-}
-
-#[cfg(all(test, feature = "std"))] // Tests make use of Vec at the moment
-mod tests {
-    use super::*;
-
-    fn collect_list<T: Copy>(mut list: LinkedList<T>) -> Vec<T> {
-        let mut result = Vec::new();
-        list.drain(|node| {
-            result.push(**node);
-        });
-        result
-    }
-
-    fn collect_reverse_list<T: Copy>(mut list: LinkedList<T>) -> Vec<T> {
-        let mut result = Vec::new();
-        list.reverse_drain(|node| {
-            result.push(**node);
-        });
-        result
-    }
-
-    unsafe fn add_nodes(list: &mut LinkedList<i32>, nodes: &mut [&mut ListNode<i32>]) {
-        for node in nodes.iter_mut() {
-            list.add_front(node);
-        }
-    }
-
-    unsafe fn assert_clean<T>(node: &mut ListNode<T>) {
-        assert!(node.next.is_none());
-        assert!(node.prev.is_none());
-    }
-
-    #[test]
-    fn insert_and_iterate() {
-        unsafe {
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-
-            let mut setup = |list: &mut LinkedList<i32>| {
-                assert_eq!(true, list.is_empty());
-                list.add_front(&mut c);
-                assert_eq!(31, **list.peek_first().unwrap());
-                assert_eq!(false, list.is_empty());
-                list.add_front(&mut b);
-                assert_eq!(7, **list.peek_first().unwrap());
-                list.add_front(&mut a);
-                assert_eq!(5, **list.peek_first().unwrap());
-            };
-
-            let mut list = LinkedList::new();
-            setup(&mut list);
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7, 31].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            setup(&mut list);
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([31, 7, 5].to_vec(), items);
-        }
-    }
-
-    #[test]
-    fn add_sorted() {
-        unsafe {
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-            let mut d = ListNode::new(99);
-
-            let mut list = LinkedList::new();
-            list.add_sorted(&mut a);
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            list.add_sorted(&mut a);
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut d, &mut c, &mut b]);
-            list.add_sorted(&mut a);
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7, 31, 99].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut d, &mut c, &mut b]);
-            list.add_sorted(&mut a);
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([99, 31, 7, 5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut d, &mut c, &mut a]);
-            list.add_sorted(&mut b);
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7, 31, 99].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut d, &mut c, &mut a]);
-            list.add_sorted(&mut b);
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([99, 31, 7, 5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut d, &mut b, &mut a]);
-            list.add_sorted(&mut c);
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7, 31, 99].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut d, &mut b, &mut a]);
-            list.add_sorted(&mut c);
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([99, 31, 7, 5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-            list.add_sorted(&mut d);
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7, 31, 99].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-            list.add_sorted(&mut d);
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([99, 31, 7, 5].to_vec(), items);
-        }
-    }
-
-    #[test]
-    fn drain_and_collect() {
-        unsafe {
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-
-            let taken_items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7, 31].to_vec(), taken_items);
-        }
-    }
-
-    #[test]
-    fn peek_last() {
-        unsafe {
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-
-            let last = list.peek_last();
-            assert_eq!(31, **last.unwrap());
-            list.remove_last();
-
-            let last = list.peek_last();
-            assert_eq!(7, **last.unwrap());
-            list.remove_last();
-
-            let last = list.peek_last();
-            assert_eq!(5, **last.unwrap());
-            list.remove_last();
-
-            let last = list.peek_last();
-            assert!(last.is_none());
-        }
-    }
-
-    #[test]
-    fn remove_first() {
-        unsafe {
-            // We iterate forward and backwards through the manipulated lists
-            // to make sure pointers in both directions are still ok.
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-            let removed = list.remove_first().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([7, 31].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-            let removed = list.remove_first().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([31, 7].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut b, &mut a]);
-            let removed = list.remove_first().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([7].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut b, &mut a]);
-            let removed = list.remove_first().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([7].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut a]);
-            let removed = list.remove_first().unwrap();
-            assert_clean(removed);
-            assert!(list.is_empty());
-            let items: Vec<i32> = collect_list(list);
-            assert!(items.is_empty());
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut a]);
-            let removed = list.remove_first().unwrap();
-            assert_clean(removed);
-            assert!(list.is_empty());
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert!(items.is_empty());
-        }
-    }
-
-    #[test]
-    fn remove_last() {
-        unsafe {
-            // We iterate forward and backwards through the manipulated lists
-            // to make sure pointers in both directions are still ok.
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-            let removed = list.remove_last().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5, 7].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-            let removed = list.remove_last().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([7, 5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut b, &mut a]);
-            let removed = list.remove_last().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_list(list);
-            assert_eq!([5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut b, &mut a]);
-            let removed = list.remove_last().unwrap();
-            assert_clean(removed);
-            assert!(!list.is_empty());
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert_eq!([5].to_vec(), items);
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut a]);
-            let removed = list.remove_last().unwrap();
-            assert_clean(removed);
-            assert!(list.is_empty());
-            let items: Vec<i32> = collect_list(list);
-            assert!(items.is_empty());
-
-            let mut list = LinkedList::new();
-            add_nodes(&mut list, &mut [&mut a]);
-            let removed = list.remove_last().unwrap();
-            assert_clean(removed);
-            assert!(list.is_empty());
-            let items: Vec<i32> = collect_reverse_list(list);
-            assert!(items.is_empty());
-        }
-    }
-
-    #[test]
-    fn remove_by_address() {
-        unsafe {
-            let mut a = ListNode::new(5);
-            let mut b = ListNode::new(7);
-            let mut c = ListNode::new(31);
-
-            {
-                // Remove first
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut a));
-                assert_clean((&mut a).into());
-                // a should be no longer there and can't be removed twice
-                assert_eq!(false, list.remove(&mut a));
-                assert_eq!(Some((&mut b).into()), list.head);
-                assert_eq!(Some((&mut c).into()), b.next);
-                assert_eq!(Some((&mut b).into()), c.prev);
-                let items: Vec<i32> = collect_list(list);
-                assert_eq!([7, 31].to_vec(), items);
-
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut a));
-                assert_clean((&mut a).into());
-                // a should be no longer there and can't be removed twice
-                assert_eq!(false, list.remove(&mut a));
-                assert_eq!(Some((&mut c).into()), b.next);
-                assert_eq!(Some((&mut b).into()), c.prev);
-                let items: Vec<i32> = collect_reverse_list(list);
-                assert_eq!([31, 7].to_vec(), items);
-            }
-
-            {
-                // Remove middle
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut b));
-                assert_clean((&mut b).into());
-                assert_eq!(Some((&mut c).into()), a.next);
-                assert_eq!(Some((&mut a).into()), c.prev);
-                let items: Vec<i32> = collect_list(list);
-                assert_eq!([5, 31].to_vec(), items);
-
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut b));
-                assert_clean((&mut b).into());
-                assert_eq!(Some((&mut c).into()), a.next);
-                assert_eq!(Some((&mut a).into()), c.prev);
-                let items: Vec<i32> = collect_reverse_list(list);
-                assert_eq!([31, 5].to_vec(), items);
-            }
-
-            {
-                // Remove last
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut c));
-                assert_clean((&mut c).into());
-                assert!(b.next.is_none());
-                assert_eq!(Some((&mut b).into()), list.tail);
-                let items: Vec<i32> = collect_list(list);
-                assert_eq!([5, 7].to_vec(), items);
-
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut c, &mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut c));
-                assert_clean((&mut c).into());
-                assert!(b.next.is_none());
-                assert_eq!(Some((&mut b).into()), list.tail);
-                let items: Vec<i32> = collect_reverse_list(list);
-                assert_eq!([7, 5].to_vec(), items);
-            }
-
-            {
-                // Remove first of two
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut a));
-                assert_clean((&mut a).into());
-                // a should be no longer there and can't be removed twice
-                assert_eq!(false, list.remove(&mut a));
-                assert_eq!(Some((&mut b).into()), list.head);
-                assert_eq!(Some((&mut b).into()), list.tail);
-                assert!(b.next.is_none());
-                assert!(b.prev.is_none());
-                let items: Vec<i32> = collect_list(list);
-                assert_eq!([7].to_vec(), items);
-
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut a));
-                assert_clean((&mut a).into());
-                // a should be no longer there and can't be removed twice
-                assert_eq!(false, list.remove(&mut a));
-                assert_eq!(Some((&mut b).into()), list.head);
-                assert_eq!(Some((&mut b).into()), list.tail);
-                assert!(b.next.is_none());
-                assert!(b.prev.is_none());
-                let items: Vec<i32> = collect_reverse_list(list);
-                assert_eq!([7].to_vec(), items);
-            }
-
-            {
-                // Remove last of two
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut b));
-                assert_clean((&mut b).into());
-                assert_eq!(Some((&mut a).into()), list.head);
-                assert_eq!(Some((&mut a).into()), list.tail);
-                assert!(a.next.is_none());
-                assert!(a.prev.is_none());
-                let items: Vec<i32> = collect_list(list);
-                assert_eq!([5].to_vec(), items);
-
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut b, &mut a]);
-                assert_eq!(true, list.remove(&mut b));
-                assert_clean((&mut b).into());
-                assert_eq!(Some((&mut a).into()), list.head);
-                assert_eq!(Some((&mut a).into()), list.tail);
-                assert!(a.next.is_none());
-                assert!(a.prev.is_none());
-                let items: Vec<i32> = collect_reverse_list(list);
-                assert_eq!([5].to_vec(), items);
-            }
-
-            {
-                // Remove last item
-                let mut list = LinkedList::new();
-                add_nodes(&mut list, &mut [&mut a]);
-                assert_eq!(true, list.remove(&mut a));
-                assert_clean((&mut a).into());
-                assert!(list.head.is_none());
-                assert!(list.tail.is_none());
-                let items: Vec<i32> = collect_list(list);
-                assert!(items.is_empty());
-            }
-
-            {
-                // Remove missing
-                let mut list = LinkedList::new();
-                list.add_front(&mut b);
-                list.add_front(&mut a);
-                assert_eq!(false, list.remove(&mut c));
-            }
-        }
-    }
-}
@@ -1,12 +1,12 @@
 //! Synchronization primitives

 mod cancellation_token;
-pub use cancellation_token::{guard::DropGuard, CancellationToken, WaitForCancellationFuture};
-
-mod intrusive_double_linked_list;
+pub use cancellation_token::{
+    guard::DropGuard, CancellationToken, WaitForCancellationFuture, WaitForCancellationFutureOwned,
+};

 mod mpsc;
-pub use mpsc::PollSender;
+pub use mpsc::{PollSendError, PollSender};

 mod poll_semaphore;
 pub use poll_semaphore::PollSemaphore;
@@ -1,221 +1,284 @@
-use futures_core::ready;
 use futures_sink::Sink;
 use std::pin::Pin;
-use std::sync::Arc;
 use std::task::{Context, Poll};
-use tokio::sync::mpsc::{error::SendError, Sender};
+use std::{fmt, mem};
+use tokio::sync::mpsc::OwnedPermit;
+use tokio::sync::mpsc::Sender;

 use super::ReusableBoxFuture;

-// This implementation was chosen over something based on permits because to get a
-// `tokio::sync::mpsc::Permit` out of the `inner` future, you must transmute the
-// lifetime on the permit to `'static`.
+/// Error returned by the `PollSender` when the channel is closed.
+#[derive(Debug)]
+pub struct PollSendError<T>(Option<T>);
+
+impl<T> PollSendError<T> {
+    /// Consumes the stored value, if any.
+    ///
+    /// If this error was encountered when calling `start_send`/`send_item`, this will be the item
+    /// that the caller attempted to send.  Otherwise, it will be `None`.
+    pub fn into_inner(self) -> Option<T> {
+        self.0
+    }
+}
+
+impl<T> fmt::Display for PollSendError<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "channel closed")
+    }
+}
+
+impl<T: fmt::Debug> std::error::Error for PollSendError<T> {}
+
+#[derive(Debug)]
+enum State<T> {
+    Idle(Sender<T>),
+    Acquiring,
+    ReadyToSend(OwnedPermit<T>),
+    Closed,
+}

 /// A wrapper around [`mpsc::Sender`] that can be polled.
 ///
 /// [`mpsc::Sender`]: tokio::sync::mpsc::Sender
 #[derive(Debug)]
 pub struct PollSender<T> {
-    /// is none if closed
-    sender: Option<Arc<Sender<T>>>,
-    is_sending: bool,
-    inner: ReusableBoxFuture<Result<(), SendError<T>>>,
+    sender: Option<Sender<T>>,
+    state: State<T>,
+    acquire: ReusableBoxFuture<'static, Result<OwnedPermit<T>, PollSendError<T>>>,
 }

-// By reusing the same async fn for both Some and None, we make sure every
-// future passed to ReusableBoxFuture has the same underlying type, and hence
-// the same size and alignment.
-async fn make_future<T>(data: Option<(Arc<Sender<T>>, T)>) -> Result<(), SendError<T>> {
+// Creates a future for acquiring a permit from the underlying channel.  This is used to ensure
+// there's capacity for a send to complete.
+//
+// By reusing the same async fn for both `Some` and `None`, we make sure every future passed to
+// ReusableBoxFuture has the same underlying type, and hence the same size and alignment.
+async fn make_acquire_future<T>(
+    data: Option<Sender<T>>,
+) -> Result<OwnedPermit<T>, PollSendError<T>> {
    match data {
-        Some((sender, value)) => sender.send(value).await,
-        None => unreachable!(
-            "This future should not be pollable, as is_sending should be set to false."
-        ),
+        Some(sender) => sender
+            .reserve_owned()
+            .await
+            .map_err(|_| PollSendError(None)),
+        None => unreachable!("this future should not be pollable in this state"),
    }
 }

 impl<T: Send + 'static> PollSender<T> {
-    /// Create a new `PollSender`.
+    /// Creates a new `PollSender`.
    pub fn new(sender: Sender<T>) -> Self {
        Self {
-            sender: Some(Arc::new(sender)),
-            is_sending: false,
-            inner: ReusableBoxFuture::new(make_future(None)),
+            sender: Some(sender.clone()),
+            state: State::Idle(sender),
+            acquire: ReusableBoxFuture::new(make_acquire_future(None)),
        }
    }

-    /// Start sending a new item.
+    fn take_state(&mut self) -> State<T> {
+        mem::replace(&mut self.state, State::Closed)
+    }
+
+    /// Attempts to prepare the sender to receive a value.
    ///
-    /// This method panics if a send is currently in progress. To ensure that no
-    /// send is in progress, call `poll_send_done` first until it returns
-    /// `Poll::Ready`.
+    /// This method must be called and return `Poll::Ready(Ok(()))` prior to each call to
+    /// `send_item`.
    ///
-    /// If this method returns an error, that indicates that the channel is
-    /// closed. Note that this method is not guaranteed to return an error if
-    /// the channel is closed, but in that case the error would be reported by
-    /// the first call to `poll_send_done`.
-    pub fn start_send(&mut self, value: T) -> Result<(), SendError<T>> {
-        if self.is_sending {
-            panic!("start_send called while not ready.");
-        }
-        match self.sender.clone() {
-            Some(sender) => {
-                self.inner.set(make_future(Some((sender, value))));
-                self.is_sending = true;
-                Ok(())
+    /// This method returns `Poll::Ready` once the underlying channel is ready to receive a value,
+    /// by reserving a slot in the channel for the item to be sent. If this method returns
+    /// `Poll::Pending`, the current task is registered to be notified (via
+    /// `cx.waker().wake_by_ref()`) when `poll_reserve` should be called again.
+    ///
+    /// # Errors
+    ///
+    /// If the channel is closed, an error will be returned.  This is a permanent state.
+    pub fn poll_reserve(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), PollSendError<T>>> {
+        loop {
+            let (result, next_state) = match self.take_state() {
+                State::Idle(sender) => {
+                    // Start trying to acquire a permit to reserve a slot for our send, and
+                    // immediately loop back around to poll it the first time.
+                    self.acquire.set(make_acquire_future(Some(sender)));
+                    (None, State::Acquiring)
+                }
+                State::Acquiring => match self.acquire.poll(cx) {
+                    // Channel has capacity.
+                    Poll::Ready(Ok(permit)) => {
+                        (Some(Poll::Ready(Ok(()))), State::ReadyToSend(permit))
+                    }
+                    // Channel is closed.
+                    Poll::Ready(Err(e)) => (Some(Poll::Ready(Err(e))), State::Closed),
+                    // Channel doesn't have capacity yet, so we need to wait.
+                    Poll::Pending => (Some(Poll::Pending), State::Acquiring),
+                },
+                // We're closed, either by choice or because the underlying sender was closed.
+                s @ State::Closed => (Some(Poll::Ready(Err(PollSendError(None)))), s),
+                // We're already ready to send an item.
+                s @ State::ReadyToSend(_) => (Some(Poll::Ready(Ok(()))), s),
+            };
+
+            self.state = next_state;
+            if let Some(result) = result {
+                return result;
            }
-            None => Err(SendError(value)),
        }
    }

-    /// If a send is in progress, poll for its completion. If no send is in progress,
-    /// this method returns `Poll::Ready(Ok(()))`.
+    /// Sends an item to the channel.
    ///
-    /// This method can return the following values:
+    /// Before calling `send_item`, `poll_reserve` must be called with a successful return
+    /// value of `Poll::Ready(Ok(()))`.
    ///
-    ///  - `Poll::Ready(Ok(()))` if the in-progress send has been completed, or there is
-    ///    no send in progress (even if the channel is closed).
-    ///  - `Poll::Ready(Err(err))` if the in-progress send failed because the channel has
-    ///    been closed.
-    ///  - `Poll::Pending` if a send is in progress, but it could not complete now.
+    /// # Errors
    ///
-    /// When this method returns `Poll::Pending`, the current task is scheduled
-    /// to receive a wakeup when the message is sent, or when the entire channel
-    /// is closed (but not if just this sender is closed by
-    /// `close_this_sender`). Note that on multiple calls to `poll_send_done`,
-    /// only the `Waker` from the `Context` passed to the most recent call is
-    /// scheduled to receive a wakeup.
+    /// If the channel is closed, an error will be returned.  This is a permanent state.
    ///
-    /// If this method returns `Poll::Ready`, then `start_send` is guaranteed to
-    /// not panic.
-    pub fn poll_send_done(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), SendError<T>>> {
-        if !self.is_sending {
-            return Poll::Ready(Ok(()));
-        }
+    /// # Panics
+    ///
+    /// If `poll_reserve` was not successfully called prior to calling `send_item`, then this method
+    /// will panic.
+    #[track_caller]
+    pub fn send_item(&mut self, value: T) -> Result<(), PollSendError<T>> {
+        let (result, next_state) = match self.take_state() {
+            State::Idle(_) | State::Acquiring => {
+                panic!("`send_item` called without first calling `poll_reserve`")
+            }
+            // We have a permit to send our item, so go ahead, which gets us our sender back.
+            State::ReadyToSend(permit) => (Ok(()), State::Idle(permit.send(value))),
+            // We're closed, either by choice or because the underlying sender was closed.
+            State::Closed => (Err(PollSendError(Some(value))), State::Closed),
+        };

-        let result = self.inner.poll(cx);
-        if result.is_ready() {
-            self.is_sending = false;
-        }
-        if let Poll::Ready(Err(_)) = &result {
-            self.sender = None;
-        }
+        // Handle deferred closing if `close` was called between `poll_reserve` and `send_item`.
+        self.state = if self.sender.is_some() {
+            next_state
+        } else {
+            State::Closed
+        };
        result
    }

-    /// Check whether the channel is ready to send more messages now.
+    /// Checks whether this sender is been closed.
    ///
-    /// If this method returns `true`, then `start_send` is guaranteed to not
-    /// panic.
-    ///
-    /// If the channel is closed, this method returns `true`.
-    pub fn is_ready(&self) -> bool {
-        !self.is_sending
-    }
-
-    /// Check whether the channel has been closed.
+    /// The underlying channel that this sender was wrapping may still be open.
    pub fn is_closed(&self) -> bool {
-        match &self.sender {
-            Some(sender) => sender.is_closed(),
-            None => true,
-        }
+        matches!(self.state, State::Closed) || self.sender.is_none()
    }

-    /// Clone the underlying `Sender`.
+    /// Gets a reference to the `Sender` of the underlying channel.
    ///
-    /// If this method returns `None`, then the channel is closed. (But it is
-    /// not guaranteed to return `None` if the channel is closed.)
-    pub fn clone_inner(&self) -> Option<Sender<T>> {
-        self.sender.as_ref().map(|sender| (&**sender).clone())
+    /// If `PollSender` has been closed, `None` is returned. The underlying channel that this sender
+    /// was wrapping may still be open.
+    pub fn get_ref(&self) -> Option<&Sender<T>> {
+        self.sender.as_ref()
    }

-    /// Access the underlying `Sender`.
+    /// Closes this sender.
    ///
-    /// If this method returns `None`, then the channel is closed. (But it is
-    /// not guaranteed to return `None` if the channel is closed.)
-    pub fn inner_ref(&self) -> Option<&Sender<T>> {
-        self.sender.as_deref()
-    }
-
-    // This operation is supported because it is required by the Sink trait.
-    /// Close this sender. No more messages can be sent from this sender.
+    /// No more messages will be able to be sent from this sender, but the underlying channel will
+    /// remain open until all senders have dropped, or until the [`Receiver`] closes the channel.
    ///
-    /// Note that this only closes the channel from the view-point of this
-    /// sender. The channel remains open until all senders have gone away, or
-    /// until the [`Receiver`] closes the channel.
-    ///
-    /// If there is a send in progress when this method is called, that send is
-    /// unaffected by this operation, and `poll_send_done` can still be called
-    /// to complete that send.
+    /// If a slot was previously reserved by calling `poll_reserve`, then a final call can be made
+    /// to `send_item` in order to consume the reserved slot.  After that, no further sends will be
+    /// possible.  If you do not intend to send another item, you can release the reserved slot back
+    /// to the underlying sender by calling [`abort_send`].
    ///
+    /// [`abort_send`]: crate::sync::PollSender::abort_send
    /// [`Receiver`]: tokio::sync::mpsc::Receiver
-    pub fn close_this_sender(&mut self) {
+    pub fn close(&mut self) {
+        // Mark ourselves officially closed by dropping our main sender.
        self.sender = None;
+
+        // If we're already idle, closed, or we haven't yet reserved a slot, we can quickly
+        // transition to the closed state.  Otherwise, leave the existing permit in place for the
+        // caller if they want to complete the send.
+        match self.state {
+            State::Idle(_) => self.state = State::Closed,
+            State::Acquiring => {
+                self.acquire.set(make_acquire_future(None));
+                self.state = State::Closed;
+            }
+            _ => {}
+        }
    }

-    /// Abort the current in-progress send, if any.
+    /// Aborts the current in-progress send, if any.
    ///
-    /// Returns `true` if a send was aborted.
+    /// Returns `true` if a send was aborted.  If the sender was closed prior to calling
+    /// `abort_send`, then the sender will remain in the closed state, otherwise the sender will be
+    /// ready to attempt another send.
    pub fn abort_send(&mut self) -> bool {
-        if self.is_sending {
-            self.inner.set(make_future(None));
-            self.is_sending = false;
-            true
-        } else {
-            false
-        }
+        // We may have been closed in the meantime, after a call to `poll_reserve` already
+        // succeeded.  We'll check if `self.sender` is `None` to see if we should transition to the
+        // closed state when we actually abort a send, rather than resetting ourselves back to idle.
+
+        let (result, next_state) = match self.take_state() {
+            // We're currently trying to reserve a slot to send into.
+            State::Acquiring => {
+                // Replacing the future drops the in-flight one.
+                self.acquire.set(make_acquire_future(None));
+
+                // If we haven't closed yet, we have to clone our stored sender since we have no way
+                // to get it back from the acquire future we just dropped.
+                let state = match self.sender.clone() {
+                    Some(sender) => State::Idle(sender),
+                    None => State::Closed,
+                };
+                (true, state)
+            }
+            // We got the permit.  If we haven't closed yet, get the sender back.
+            State::ReadyToSend(permit) => {
+                let state = if self.sender.is_some() {
+                    State::Idle(permit.release())
+                } else {
+                    State::Closed
+                };
+                (true, state)
+            }
+            s => (false, s),
+        };
+
+        self.state = next_state;
+        result
    }
 }

 impl<T> Clone for PollSender<T> {
-    /// Clones this `PollSender`. The resulting clone will not have any
-    /// in-progress send operations, even if the current `PollSender` does.
+    /// Clones this `PollSender`.
+    ///
+    /// The resulting `PollSender` will have an initial state identical to calling `PollSender::new`.
    fn clone(&self) -> PollSender<T> {
+        let (sender, state) = match self.sender.clone() {
+            Some(sender) => (Some(sender.clone()), State::Idle(sender)),
+            None => (None, State::Closed),
+        };
+
        Self {
-            sender: self.sender.clone(),
-            is_sending: false,
-            inner: ReusableBoxFuture::new(async { unreachable!() }),
+            sender,
+            state,
+            // We don't use `make_acquire_future` here because our relaxed bounds on `T` are not
+            // compatible with the transitive bounds required by `Sender<T>`.
+            acquire: ReusableBoxFuture::new(async { unreachable!() }),
        }
    }
 }

 impl<T: Send + 'static> Sink<T> for PollSender<T> {
-    type Error = SendError<T>;
+    type Error = PollSendError<T>;

-    /// This is equivalent to calling [`poll_send_done`].
-    ///
-    /// [`poll_send_done`]: PollSender::poll_send_done
    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Pin::into_inner(self).poll_send_done(cx)
+        Pin::into_inner(self).poll_reserve(cx)
    }

-    /// This is equivalent to calling [`poll_send_done`].
-    ///
-    /// [`poll_send_done`]: PollSender::poll_send_done
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Pin::into_inner(self).poll_send_done(cx)
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
    }

-    /// This is equivalent to calling [`start_send`].
-    ///
-    /// [`start_send`]: PollSender::start_send
    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
-        Pin::into_inner(self).start_send(item)
+        Pin::into_inner(self).send_item(item)
    }

-    /// This method will first flush the `PollSender`, and then close it by
-    /// calling [`close_this_sender`].
-    ///
-    /// If a send fails while flushing because the [`Receiver`] has gone away,
-    /// then this function returns an error. The channel is still successfully
-    /// closed in this situation.
-    ///
-    /// [`close_this_sender`]: PollSender::close_this_sender
-    /// [`Receiver`]: tokio::sync::mpsc::Receiver
-    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        ready!(self.as_mut().poll_flush(cx))?;
-
-        Pin::into_inner(self).close_this_sender();
+    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::into_inner(self).close();
        Poll::Ready(Ok(()))
    }
 }
@@ -12,7 +12,10 @@ use super::ReusableBoxFuture;
 /// [`Semaphore`]: tokio::sync::Semaphore
 pub struct PollSemaphore {
    semaphore: Arc<Semaphore>,
-    permit_fut: Option<ReusableBoxFuture<Result<OwnedSemaphorePermit, AcquireError>>>,
+    permit_fut: Option<(
+        u32, // The number of permits requested.
+        ReusableBoxFuture<'static, Result<OwnedSemaphorePermit, AcquireError>>,
+    )>,
 }

 impl PollSemaphore {
@@ -53,25 +56,57 @@ impl PollSemaphore {
    /// the `Waker` from the `Context` passed to the most recent call is
    /// scheduled to receive a wakeup.
    pub fn poll_acquire(&mut self, cx: &mut Context<'_>) -> Poll<Option<OwnedSemaphorePermit>> {
+        self.poll_acquire_many(cx, 1)
+    }
+
+    /// Poll to acquire many permits from the semaphore.
+    ///
+    /// This can return the following values:
+    ///
+    ///  - `Poll::Pending` if a permit is not currently available.
+    ///  - `Poll::Ready(Some(permit))` if a permit was acquired.
+    ///  - `Poll::Ready(None)` if the semaphore has been closed.
+    ///
+    /// When this method returns `Poll::Pending`, the current task is scheduled
+    /// to receive a wakeup when the permits become available, or when the
+    /// semaphore is closed. Note that on multiple calls to `poll_acquire`, only
+    /// the `Waker` from the `Context` passed to the most recent call is
+    /// scheduled to receive a wakeup.
+    pub fn poll_acquire_many(
+        &mut self,
+        cx: &mut Context<'_>,
+        permits: u32,
+    ) -> Poll<Option<OwnedSemaphorePermit>> {
        let permit_future = match self.permit_fut.as_mut() {
-            Some(fut) => fut,
+            Some((prev_permits, fut)) if *prev_permits == permits => fut,
+            Some((old_permits, fut_box)) => {
+                // We're requesting a different number of permits, so replace the future
+                // and record the new amount.
+                let fut = Arc::clone(&self.semaphore).acquire_many_owned(permits);
+                fut_box.set(fut);
+                *old_permits = permits;
+                fut_box
+            }
            None => {
                // avoid allocations completely if we can grab a permit immediately
-                match Arc::clone(&self.semaphore).try_acquire_owned() {
+                match Arc::clone(&self.semaphore).try_acquire_many_owned(permits) {
                    Ok(permit) => return Poll::Ready(Some(permit)),
                    Err(TryAcquireError::Closed) => return Poll::Ready(None),
                    Err(TryAcquireError::NoPermits) => {}
                }

-                let next_fut = Arc::clone(&self.semaphore).acquire_owned();
-                self.permit_fut
-                    .get_or_insert(ReusableBoxFuture::new(next_fut))
+                let next_fut = Arc::clone(&self.semaphore).acquire_many_owned(permits);
+                &mut self
+                    .permit_fut
+                    .get_or_insert((permits, ReusableBoxFuture::new(next_fut)))
+                    .1
            }
        };

        let result = ready!(permit_future.poll(cx));

-        let next_fut = Arc::clone(&self.semaphore).acquire_owned();
+        // Assume we'll request the same amount of permits in a subsequent call.
+        let next_fut = Arc::clone(&self.semaphore).acquire_many_owned(permits);
        permit_future.set(next_fut);

        match result {
@@ -95,7 +130,7 @@ impl PollSemaphore {

    /// Adds `n` new permits to the semaphore.
    ///
-    /// The maximum number of permits is `usize::MAX >> 3`, and this function
+    /// The maximum number of permits is [`Semaphore::MAX_PERMITS`], and this function
    /// will panic if the limit is exceeded.
    ///
    /// This is equivalent to the [`Semaphore::add_permits`] method on the
@@ -131,6 +166,6 @@ impl fmt::Debug for PollSemaphore {

 impl AsRef<Semaphore> for PollSemaphore {
    fn as_ref(&self) -> &Semaphore {
-        &*self.semaphore
+        &self.semaphore
    }
 }
@@ -1,33 +1,29 @@
 use std::alloc::Layout;
+use std::fmt;
 use std::future::Future;
-use std::panic::AssertUnwindSafe;
+use std::marker::PhantomData;
+use std::mem::{self, ManuallyDrop};
 use std::pin::Pin;
-use std::ptr::{self, NonNull};
+use std::ptr;
 use std::task::{Context, Poll};
-use std::{fmt, panic};

-/// A reusable `Pin<Box<dyn Future<Output = T> + Send>>`.
+/// A reusable `Pin<Box<dyn Future<Output = T> + Send + 'a>>`.
 ///
 /// This type lets you replace the future stored in the box without
 /// reallocating when the size and alignment permits this.
-pub struct ReusableBoxFuture<T> {
-    boxed: NonNull<dyn Future<Output = T> + Send>,
+pub struct ReusableBoxFuture<'a, T> {
+    boxed: Pin<Box<dyn Future<Output = T> + Send + 'a>>,
 }

-impl<T> ReusableBoxFuture<T> {
+impl<'a, T> ReusableBoxFuture<'a, T> {
    /// Create a new `ReusableBoxFuture<T>` containing the provided future.
    pub fn new<F>(future: F) -> Self
    where
-        F: Future<Output = T> + Send + 'static,
+        F: Future<Output = T> + Send + 'a,
    {
-        let boxed: Box<dyn Future<Output = T> + Send> = Box::new(future);
-
-        let boxed = Box::into_raw(boxed);
-
-        // SAFETY: Box::into_raw does not return null pointers.
-        let boxed = unsafe { NonNull::new_unchecked(boxed) };
-
-        Self { boxed }
+        Self {
+            boxed: Box::pin(future),
+        }
    }

    /// Replace the future currently stored in this box.
@@ -36,7 +32,7 @@ impl<T> ReusableBoxFuture<T> {
    /// different from the layout of the currently stored future.
    pub fn set<F>(&mut self, future: F)
    where
-        F: Future<Output = T> + Send + 'static,
+        F: Future<Output = T> + Send + 'a,
    {
        if let Err(future) = self.try_set(future) {
            *self = Self::new(future);
@@ -50,64 +46,31 @@ impl<T> ReusableBoxFuture<T> {
    /// future.
    pub fn try_set<F>(&mut self, future: F) -> Result<(), F>
    where
-        F: Future<Output = T> + Send + 'static,
+        F: Future<Output = T> + Send + 'a,
    {
-        // SAFETY: The pointer is not dangling.
-        let self_layout = {
-            let dyn_future: &(dyn Future<Output = T> + Send) = unsafe { self.boxed.as_ref() };
-            Layout::for_value(dyn_future)
-        };
-
-        if Layout::new::<F>() == self_layout {
-            // SAFETY: We just checked that the layout of F is correct.
-            unsafe {
-                self.set_same_layout(future);
-            }
-
-            Ok(())
-        } else {
-            Err(future)
+        // If we try to inline the contents of this function, the type checker complains because
+        // the bound `T: 'a` is not satisfied in the call to `pending()`. But by putting it in an
+        // inner function that doesn't have `T` as a generic parameter, we implicitly get the bound
+        // `F::Output: 'a` transitively through `F: 'a`, allowing us to call `pending()`.
+        #[inline(always)]
+        fn real_try_set<'a, F>(
+            this: &mut ReusableBoxFuture<'a, F::Output>,
+            future: F,
+        ) -> Result<(), F>
+        where
+            F: Future + Send + 'a,
+        {
+            // future::Pending<T> is a ZST so this never allocates.
+            let boxed = mem::replace(&mut this.boxed, Box::pin(Pending(PhantomData)));
+            reuse_pin_box(boxed, future, |boxed| this.boxed = Pin::from(boxed))
        }
-    }

-    /// Set the current future.
-    ///
-    /// # Safety
-    ///
-    /// This function requires that the layout of the provided future is the
-    /// same as `self.layout`.
-    unsafe fn set_same_layout<F>(&mut self, future: F)
-    where
-        F: Future<Output = T> + Send + 'static,
-    {
-        // Drop the existing future, catching any panics.
-        let result = panic::catch_unwind(AssertUnwindSafe(|| {
-            ptr::drop_in_place(self.boxed.as_ptr());
-        }));
-
-        // Overwrite the future behind the pointer. This is safe because the
-        // allocation was allocated with the same size and alignment as the type F.
-        let self_ptr: *mut F = self.boxed.as_ptr() as *mut F;
-        ptr::write(self_ptr, future);
-
-        // Update the vtable of self.boxed. The pointer is not null because we
-        // just got it from self.boxed, which is not null.
-        self.boxed = NonNull::new_unchecked(self_ptr);
-
-        // If the old future's destructor panicked, resume unwinding.
-        match result {
-            Ok(()) => {}
-            Err(payload) => {
-                panic::resume_unwind(payload);
-            }
-        }
+        real_try_set(self, future)
    }

    /// Get a pinned reference to the underlying future.
    pub fn get_pin(&mut self) -> Pin<&mut (dyn Future<Output = T> + Send)> {
-        // SAFETY: The user of this box cannot move the box, and we do not move it
-        // either.
-        unsafe { Pin::new_unchecked(self.boxed.as_mut()) }
+        self.boxed.as_mut()
    }

    /// Poll the future stored inside this box.
@@ -116,7 +79,7 @@ impl<T> ReusableBoxFuture<T> {
    }
 }

-impl<T> Future for ReusableBoxFuture<T> {
+impl<T> Future for ReusableBoxFuture<'_, T> {
    type Output = T;

    /// Poll the future stored inside this box.
@@ -125,27 +88,84 @@ impl<T> Future for ReusableBoxFuture<T> {
    }
 }

-// The future stored inside ReusableBoxFuture<T> must be Send.
-unsafe impl<T> Send for ReusableBoxFuture<T> {}
-
 // The only method called on self.boxed is poll, which takes &mut self, so this
 // struct being Sync does not permit any invalid access to the Future, even if
 // the future is not Sync.
-unsafe impl<T> Sync for ReusableBoxFuture<T> {}
+unsafe impl<T> Sync for ReusableBoxFuture<'_, T> {}

-// Just like a Pin<Box<dyn Future>> is always Unpin, so is this type.
-impl<T> Unpin for ReusableBoxFuture<T> {}
-
-impl<T> Drop for ReusableBoxFuture<T> {
-    fn drop(&mut self) {
-        unsafe {
-            drop(Box::from_raw(self.boxed.as_ptr()));
-        }
-    }
-}
-
-impl<T> fmt::Debug for ReusableBoxFuture<T> {
+impl<T> fmt::Debug for ReusableBoxFuture<'_, T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ReusableBoxFuture").finish()
    }
 }
+
+fn reuse_pin_box<T: ?Sized, U, O, F>(boxed: Pin<Box<T>>, new_value: U, callback: F) -> Result<O, U>
+where
+    F: FnOnce(Box<U>) -> O,
+{
+    let layout = Layout::for_value::<T>(&*boxed);
+    if layout != Layout::new::<U>() {
+        return Err(new_value);
+    }
+
+    // SAFETY: We don't ever construct a non-pinned reference to the old `T` from now on, and we
+    // always drop the `T`.
+    let raw: *mut T = Box::into_raw(unsafe { Pin::into_inner_unchecked(boxed) });
+
+    // When dropping the old value panics, we still want to call `callback` — so move the rest of
+    // the code into a guard type.
+    let guard = CallOnDrop::new(|| {
+        let raw: *mut U = raw.cast::<U>();
+        unsafe { raw.write(new_value) };
+
+        // SAFETY:
+        // - `T` and `U` have the same layout.
+        // - `raw` comes from a `Box` that uses the same allocator as this one.
+        // - `raw` points to a valid instance of `U` (we just wrote it in).
+        let boxed = unsafe { Box::from_raw(raw) };
+
+        callback(boxed)
+    });
+
+    // Drop the old value.
+    unsafe { ptr::drop_in_place(raw) };
+
+    // Run the rest of the code.
+    Ok(guard.call())
+}
+
+struct CallOnDrop<O, F: FnOnce() -> O> {
+    f: ManuallyDrop<F>,
+}
+
+impl<O, F: FnOnce() -> O> CallOnDrop<O, F> {
+    fn new(f: F) -> Self {
+        let f = ManuallyDrop::new(f);
+        Self { f }
+    }
+    fn call(self) -> O {
+        let mut this = ManuallyDrop::new(self);
+        let f = unsafe { ManuallyDrop::take(&mut this.f) };
+        f()
+    }
+}
+
+impl<O, F: FnOnce() -> O> Drop for CallOnDrop<O, F> {
+    fn drop(&mut self) {
+        let f = unsafe { ManuallyDrop::take(&mut self.f) };
+        f();
+    }
+}
+
+/// The same as `std::future::Pending<T>`; we can't use that type directly because on rustc
+/// versions <1.60 it didn't unconditionally implement `Send`.
+// FIXME: use `std::future::Pending<T>` once the MSRV is >=1.60
+struct Pending<T>(PhantomData<fn() -> T>);
+
+impl<T> Future for Pending<T> {
+    type Output = T;
+
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Poll::Pending
+    }
+}
--- a/Show More
+++ b/Show More