chore: prepare Tokio v1.28.1 (#5679)

This reverts commit cf9a03c107.

.cargo/config

@@ -0,0 +1,2 @@
+# [build]
+# rustflags = ["--cfg", "tokio_unstable"]

.circleci/config.yml

@@ -5,7 +5,7 @@ jobs:
       image: ubuntu-2004:202101-01
     resource_class: arm.medium
     environment:
-      # Change to pin rust versino
+      # Change to pin rust version
       RUST_STABLE: stable
     steps:
       - checkout
@@ -22,4 +22,4 @@ jobs:
 workflows:
   ci:
     jobs:
-      - test-arm
+      - test-arm

.cirrus.yml

@@ -1,8 +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-3-release-amd64
+  image_family: freebsd-13-1
 env:
   RUST_STABLE: stable
-  RUST_NIGHTLY: nightly-2022-03-21
+  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 +14,7 @@ env:
 task:
   name: FreeBSD 64-bit
   setup_script:
-    - pkg install -y bash curl
+    - pkg install -y bash
     - curl https://sh.rustup.rs -sSf --output rustup.sh
     - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
     - . $HOME/.cargo/env
@@ -29,7 +31,7 @@ task:
     RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
     RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
   setup_script:
-    - pkg install -y bash curl
+    - pkg install -y bash
     - curl https://sh.rustup.rs -sSf --output rustup.sh
     - sh rustup.sh -y --profile minimal --default-toolchain $RUST_NIGHTLY
     - . $HOME/.cargo/env
@@ -43,7 +45,7 @@ task:
 task:
   name: FreeBSD 32-bit
   setup_script:
-    - pkg install -y bash curl
+    - pkg install -y bash
     - curl https://sh.rustup.rs -sSf --output rustup.sh
     - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
     - . $HOME/.cargo/env

.clippy.toml

@@ -1 +1 @@
-msrv = "1.49"
+msrv = "1.56"

.github/ISSUE_TEMPLATE/config.yml

@@ -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.

.github/ISSUE_TEMPLATE/question.md

@@ -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/

.github/workflows/audit.yml

@@ -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 }}

.github/workflows/ci.yml

@@ -11,39 +11,71 @@ env:
   RUST_BACKTRACE: 1
   # Change to specific Rust release to pin
   rust_stable: stable
-  rust_nightly: nightly-2022-03-21
-  rust_clippy: 1.52.0
-  rust_min: 1.49.0
+  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.56.0
 
 defaults:
   run:
     shell: bash
 
+permissions:
+  contents: read
+
 jobs:
-  # Depends on all action sthat are required for a "successful" CI run.
+  # Depends on all actions that are required for a "successful" CI run.
   tests-pass:
     name: all systems go
     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
+
+  # Basic actions that must pass before we kick off more expensive tests.
+  basics:
+    name: basic checks
+    runs-on: ubuntu-latest
+    needs:
+      - clippy
+      - fmt
+      - docs
+      - minrust
     steps:
       - run: exit 0
 
   test:
+    needs: basics
     name: test tokio full
     runs-on: ${{ matrix.os }}
     strategy:
@@ -53,17 +85,16 @@ jobs:
           - ubuntu-latest
           - macos-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
             toolchain: ${{ env.rust_stable }}
-            override: true
       - 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.
@@ -101,15 +132,15 @@ jobs:
     # This relies on the potentially affected Tokio type being listed in
     # `tokio/tokio/tests/async_send_sync.rs`.
     name: compile tests with parking lot send guards
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
             toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - 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
@@ -118,20 +149,18 @@ jobs:
 
   valgrind:
     name: valgrind
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
             toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - 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
@@ -153,6 +182,7 @@ jobs:
 
   test-unstable:
     name: test tokio full --unstable
+    needs: basics
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
@@ -161,13 +191,12 @@ jobs:
           - ubuntu-latest
           - macos-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
             toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       # Run `tokio` with "unstable" cfg flag.
       - name: test tokio full --cfg unstable
         run: cargo test --all-features
@@ -180,88 +209,167 @@ jobs:
 
   miri:
     name: miri
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_nightly }}
           components: miri
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       - name: miri
         # 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:
-          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-tag-raw-pointers
-          PROPTEST_CASES: 10
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
 
-  san:
-    name: san
+  asan:
+    name: asan
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - 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: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       - name: asan
-        run: cargo test --all-features --target x86_64-unknown-linux-gnu --lib -- --test-threads 1
-        working-directory: tokio
+        run: cargo test --workspace --all-features --target x86_64-unknown-linux-gnu --tests -- --test-threads 1
         env:
           RUSTFLAGS: -Z sanitizer=address
-          ASAN_OPTIONS: detect_leaks=0
+          # Ignore `trybuild` errors as they are irrelevant and flaky on nightly
+          TRYBUILD: overwrite
 
-  cross:
-    name: cross
+  semver:
+    name: semver
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Check semver
+        uses: obi1kenobi/cargo-semver-checks-action@v2
+        with:
+          rust-toolchain: ${{ env.rust_stable }}
+          release-type: minor
+
+  cross-check:
+    name: cross-check
+    needs: basics
     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/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           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
+    needs: basics
+    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
+    needs: basics
+    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
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_nightly }}
           target: ${{ matrix.target }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - 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
 
@@ -269,29 +377,43 @@ jobs:
     name: minrust
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_min }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_min }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: "test --workspace --all-features"
+      - 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
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_nightly }}
-          override: true
-      - 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
       - name: "check --all-features -Z minimal-versions"
         run: |
           # Remove dev-dependencies from Cargo.toml to prevent the next `cargo update`
@@ -315,14 +437,13 @@ jobs:
     name: fmt
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_stable }}
-          override: true
           components: rustfmt
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       # Check fmt
       - name: "rustfmt --check"
         # Workaround for rust-lang/cargo#7732
@@ -336,14 +457,13 @@ jobs:
     name: clippy
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_clippy }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_clippy }}
-          override: true
           components: clippy
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       # Run clippy
       - name: "clippy --all"
         run: cargo clippy --all --tests --all-features
@@ -352,13 +472,12 @@ jobs:
     name: docs
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_nightly }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       - name: "doc --lib --all-features"
         run: cargo doc --lib --no-deps --all-features --document-private-items
         env:
@@ -367,15 +486,15 @@ jobs:
 
   loom-compile:
     name: build loom tests
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_stable }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       - name: build --cfg loom
         run: cargo test --no-run --lib --features full
         working-directory: tokio
@@ -386,7 +505,7 @@ 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
 
@@ -396,6 +515,7 @@ jobs:
 
   test-hyper:
     name: Test hyper
+    needs: basics
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
@@ -404,13 +524,12 @@ jobs:
           - ubuntu-latest
           - macos-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_stable }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - uses: Swatinem/rust-cache@v2
       - name: Test hyper
         run: |
           set -x
@@ -428,19 +547,109 @@ jobs:
           git diff
           cargo test --features full
 
-  wasm32-unknown-unknown:
-    name: test tokio for wasm32-unknown-unknown
+  x86_64-fortanix-unknown-sgx:
+    name: build tokio for x86_64-fortanix-unknown-sgx
+    needs: basics
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - 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
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_stable }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
+      - 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
+    needs: basics
+    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
+    needs: basics
+    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

.github/workflows/labeler.yml

@@ -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:

.github/workflows/loom.yml

@@ -13,11 +13,14 @@ env:
   # 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:
@@ -29,17 +32,17 @@ jobs:
           - loom_pool::group_d
           - time::driver
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
             toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - 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 }}

.github/workflows/pr-audit.yml

@@ -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

.github/workflows/stress-test.yml

@@ -11,8 +11,11 @@ env:
   # 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:
@@ -20,17 +23,14 @@ jobs:
         stress-test:
           - simple_echo_tcp
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@master
         with:
             toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - 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
@@ -38,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 }}

CONTRIBUTING.md

@@ -131,12 +131,33 @@ cargo check --all-features
 cargo test --all-features
 ```
 
+Ideally, you should use the same version of clippy as the one used in CI
+(defined by `env.rust_clippy` in [ci.yml][ci.yml]), because newer versions
+might have new lints:
+
+[ci.yml]: .github/workflows/ci.yml
+
+<!--
+When updating this, also update:
+- .github/workflows/ci.yml
+- README.md
+- tokio/README.md
+- tokio/Cargo.toml
+- tokio-util/Cargo.toml
+- tokio-test/Cargo.toml
+- tokio-stream/Cargo.toml
+-->
+
+```
+cargo +1.65.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 +165,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:
 
@@ -175,7 +190,7 @@ LOOM_MAX_PREEMPTIONS=1 RUSTFLAGS="--cfg loom" \
 
 You can run miri tests with
 ```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" PROPTEST_CASES=10 \
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" \
     cargo +nightly miri test --features full --lib
 ```
 
@@ -185,8 +200,22 @@ If the change being proposed alters code (as opposed to only documentation for
 example), it is either adding new functionality to Tokio or it is fixing
 existing, broken functionality. In both of these cases, the pull request should
 include one or more tests to ensure that Tokio does not regress in the future.
-There are two ways to write tests: integration tests and documentation tests
-(Tokio avoids unit tests as much as possible).
+There are two ways to write tests: [integration tests][integration-tests]
+and [documentation tests][documentation-tests].
+(Tokio avoids [unit tests][unit-tests] as much as possible).
+
+Tokio uses [conditional compilation attributes][conditional-compilation]
+throughout the codebase, to modify rustc's behavior. Code marked with such
+attributes can be enabled using RUSTFLAGS and RUSTDOCFLAGS environment
+variables. One of the most prevalent flags passed in these variables is
+the `--cfg` option. To run tests in a particular file, check first what
+options #![cfg] declaration defines for that file.
+
+For instance, to run a test marked with the 'tokio_unstable' cfg option,
+you must pass this flag to the compiler when running the test.
+```
+$ RUSTFLAGS="--cfg tokio_unstable" cargo test -p tokio --all-features --test rt_metrics
+```
 
 #### Integration tests
 
@@ -197,6 +226,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
@@ -544,7 +598,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.
@@ -621,3 +675,7 @@ When releasing a new version of a crate, follow these steps:
    entry for that release version into your editor and close the window.
 
 [keep-a-changelog]: https://github.com/olivierlacan/keep-a-changelog/blob/master/CHANGELOG.md
+[unit-tests]: https://doc.rust-lang.org/rust-by-example/testing/unit_testing.html
+[integration-tests]: https://doc.rust-lang.org/rust-by-example/testing/integration_testing.html
+[documentation-tests]: https://doc.rust-lang.org/rust-by-example/testing/doc_testing.html
+[conditional-compilation]: https://doc.rust-lang.org/reference/conditional-compilation.html

Cross.toml

@@ -0,0 +1,5 @@
+[build.env]
+passthrough = [
+    "RUSTFLAGS",
+    "RUST_BACKTRACE",
+]

LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2022 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

README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.17.0", features = ["full"] }
+tokio = { version = "1.28.1", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -161,11 +161,46 @@ 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
 
+<!--
+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.
+released at least six months ago. The current MSRV is 1.56.0.
+
+Note that the MSRV is not increased automatically, and only as part of a minor
+release. The MSRV history for past minor releases can be found below:
+
+ * 1.27 to now - Rust 1.56
+ * 1.17 to 1.26 - Rust 1.49
+ * 1.15 to 1.16 - Rust 1.46
+ * 1.0 to 1.14 - Rust 1.45
+
+Note that although we try to avoid the situation where a dependency transitively
+increases the MSRV of Tokio, we do not guarantee that this does not happen.
+However, every minor release will have some set of versions of dependencies that
+works with the MSRV of that minor release.
 
 ## Release schedule
 
@@ -180,18 +215,19 @@ 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. (MSRV 1.49)
+ * `1.20.x` - LTS release until September 2023. (MSRV 1.49)
+ * `1.25.x` - LTS release until March 2024. (MSRV 1.49)
 
-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

SECURITY.md

@@ -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`).

benches/Cargo.toml

@@ -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,11 +30,25 @@ 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"
 harness = false
 
+[[bench]]
+name = "sync_notify"
+path = "sync_notify.rs"
+harness = false
 
 [[bench]]
 name = "sync_rwlock"
@@ -50,3 +69,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

benches/copy.rs

@@ -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);

benches/sync_mpsc_oneshot.rs

@@ -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);

benches/sync_notify.rs

@@ -0,0 +1,90 @@
+use bencher::Bencher;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+use tokio::sync::Notify;
+
+fn rt() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(6)
+        .build()
+        .unwrap()
+}
+
+fn notify_waiters<const N_WAITERS: usize>(b: &mut Bencher) {
+    let rt = rt();
+    let notify = Arc::new(Notify::new());
+    let counter = Arc::new(AtomicUsize::new(0));
+    for _ in 0..N_WAITERS {
+        rt.spawn({
+            let notify = notify.clone();
+            let counter = counter.clone();
+            async move {
+                loop {
+                    notify.notified().await;
+                    counter.fetch_add(1, Ordering::Relaxed);
+                }
+            }
+        });
+    }
+
+    const N_ITERS: usize = 500;
+    b.iter(|| {
+        counter.store(0, Ordering::Relaxed);
+        loop {
+            notify.notify_waiters();
+            if counter.load(Ordering::Relaxed) >= N_ITERS {
+                break;
+            }
+        }
+    });
+}
+
+fn notify_one<const N_WAITERS: usize>(b: &mut Bencher) {
+    let rt = rt();
+    let notify = Arc::new(Notify::new());
+    let counter = Arc::new(AtomicUsize::new(0));
+    for _ in 0..N_WAITERS {
+        rt.spawn({
+            let notify = notify.clone();
+            let counter = counter.clone();
+            async move {
+                loop {
+                    notify.notified().await;
+                    counter.fetch_add(1, Ordering::Relaxed);
+                }
+            }
+        });
+    }
+
+    const N_ITERS: usize = 500;
+    b.iter(|| {
+        counter.store(0, Ordering::Relaxed);
+        loop {
+            notify.notify_one();
+            if counter.load(Ordering::Relaxed) >= N_ITERS {
+                break;
+            }
+        }
+    });
+}
+
+bencher::benchmark_group!(
+    notify_waiters_simple,
+    notify_waiters::<10>,
+    notify_waiters::<50>,
+    notify_waiters::<100>,
+    notify_waiters::<200>,
+    notify_waiters::<500>
+);
+
+bencher::benchmark_group!(
+    notify_one_simple,
+    notify_one::<10>,
+    notify_one::<50>,
+    notify_one::<100>,
+    notify_one::<200>,
+    notify_one::<500>
+);
+
+bencher::benchmark_main!(notify_waiters_simple, notify_one_simple);

benches/sync_rwlock.rs

@@ -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(()));

benches/sync_watch.rs

@@ -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);

benches/time_now.rs

@@ -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);

examples/Cargo.toml

@@ -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.48"
 
 [[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"

examples/hello_world.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(())

examples/named-pipe-multi-client.rs

@@ -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),
                 }
 

examples/tinyhttp.rs

@@ -18,7 +18,7 @@ use futures::SinkExt;
 use http::{header::HeaderValue, Request, Response, StatusCode};
 #[macro_use]
 extern crate serde_derive;
-use std::{env, error::Error, fmt, io};
+use std::{convert::TryFrom, env, error::Error, fmt, io};
 use tokio::net::{TcpListener, TcpStream};
 use tokio_stream::StreamExt;
 use tokio_util::codec::{Decoder, Encoder, Framed};
@@ -180,8 +180,11 @@ impl Decoder for Http {
                 headers[i] = Some((k, v));
             }
 
+            let method = http::Method::try_from(r.method.unwrap())
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+
             (
-                toslice(r.method.unwrap().as_bytes()),
+                method,
                 toslice(r.path.unwrap().as_bytes()),
                 r.version.unwrap(),
                 amt,
@@ -195,7 +198,7 @@ impl Decoder for Http {
         }
         let data = src.split_to(amt).freeze();
         let mut ret = Request::builder();
-        ret = ret.method(&data[method.0..method.1]);
+        ret = ret.method(method);
         let s = data.slice(path.0..path.1);
         let s = unsafe { String::from_utf8_unchecked(Vec::from(s.as_ref())) };
         ret = ret.uri(s);

netlify.toml

@@ -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"

target-specs/i686-unknown-linux-gnu.json

@@ -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"
+}

tests-build/README.md

@@ -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.

tests-build/tests/fail/macros_dead_code.stderr

@@ -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() {}

tests-build/tests/fail/macros_invalid_input.rs

@@ -1,3 +1,5 @@
+#![deny(duplicate_macro_attributes)]
+
 use tests_build::tokio;
 
 #[tokio::main]
@@ -34,13 +36,10 @@ async fn test_worker_threads_not_int() {}
 async fn test_worker_threads_and_current_thread() {}
 
 #[tokio::test(crate = 456)]
-async fn test_crate_not_ident_int() {}
+async fn test_crate_not_path_int() {}
 
 #[tokio::test(crate = "456")]
-async fn test_crate_not_ident_invalid() {}
-
-#[tokio::test(crate = "abc::edf")]
-async fn test_crate_not_ident_path() {}
+async fn test_crate_not_path_invalid() {}
 
 #[tokio::test]
 #[test]

tests-build/tests/fail/macros_invalid_input.stderr

@@ -1,97 +1,95 @@
 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`, `crate`
- --> $DIR/macros_invalid_input.rs:6:15
+ --> $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`, `crate`
-  --> $DIR/macros_invalid_input.rs:15:15
+  --> $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`, `crate`
-  --> $DIR/macros_invalid_input.rs:18:15
+  --> $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: Failed to parse value of `crate` as ident.
-  --> $DIR/macros_invalid_input.rs:36:23
+error: Failed to parse value of `crate` as path.
+  --> $DIR/macros_invalid_input.rs:38:23
    |
-36 | #[tokio::test(crate = 456)]
+38 | #[tokio::test(crate = 456)]
    |                       ^^^
 
-error: Failed to parse value of `crate` as ident: "456"
-  --> $DIR/macros_invalid_input.rs:39:23
+error: Failed to parse value of `crate` as path: "456"
+  --> $DIR/macros_invalid_input.rs:41:23
    |
-39 | #[tokio::test(crate = "456")]
+41 | #[tokio::test(crate = "456")]
    |                       ^^^^^
 
-error: Failed to parse value of `crate` as ident: "abc::edf"
-  --> $DIR/macros_invalid_input.rs:42:23
-   |
-42 | #[tokio::test(crate = "abc::edf")]
-   |                       ^^^^^^^^^^
-
 error: second test attribute is supplied
-  --> $DIR/macros_invalid_input.rs:46:1
+  --> $DIR/macros_invalid_input.rs:45:1
    |
-46 | #[test]
+45 | #[test]
    | ^^^^^^^
 
 error: duplicated attribute
-  --> $DIR/macros_invalid_input.rs:46:1
+  --> $DIR/macros_invalid_input.rs:45:1
    |
-46 | #[test]
+45 | #[test]
    | ^^^^^^^
    |
-   = note: `-D duplicate-macro-attributes` implied by `-D warnings`
+note: the lint level is defined here
+  --> $DIR/macros_invalid_input.rs:1:9
+   |
+1  | #![deny(duplicate_macro_attributes)]
+   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^

tests-build/tests/fail/macros_type_mismatch.rs

@@ -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() {}

tests-build/tests/fail/macros_type_mismatch.stderr

@@ -1,35 +1,33 @@
 error[E0308]: mismatched types
- --> $DIR/macros_type_mismatch.rs:5:5
+ --> tests/fail/macros_type_mismatch.rs:5:5
   |
 4 | async fn missing_semicolon_or_return_type() {
-  |                                             - possibly return type missing here?
+  |                                             - help: a return type might be missing here: `-> _`
 5 |     Ok(())
-  |     ^^^^^^- help: consider using a semicolon here: `;`
-  |     |
-  |     expected `()`, found enum `Result`
+  |     ^^^^^^ expected `()`, found `Result<(), _>`
   |
   = note: expected unit type `()`
                   found enum `Result<(), _>`
 
 error[E0308]: mismatched types
-  --> $DIR/macros_type_mismatch.rs:10:5
+  --> tests/fail/macros_type_mismatch.rs:10:5
    |
 9  | async fn missing_return_type() {
-   |                                - possibly return type missing here?
+   |                                - help: a return type might be missing here: `-> _`
 10 |     return Ok(());
-   |     ^^^^^^^^^^^^^^ expected `()`, found enum `Result`
+   |     ^^^^^^^^^^^^^^ expected `()`, found `Result<(), _>`
    |
    = note: expected unit type `()`
                    found enum `Result<(), _>`
 
 error[E0308]: mismatched types
-  --> $DIR/macros_type_mismatch.rs:23:5
+  --> tests/fail/macros_type_mismatch.rs:23:5
    |
 14 | async fn extra_semicolon() -> Result<(), ()> {
    |                               -------------- expected `Result<(), ()>` because of return type
 ...
 23 |     Ok(());
-   |     ^^^^^^^ expected enum `Result`, found `()`
+   |     ^^^^^^^ expected `Result<(), ()>`, found `()`
    |
    = note:   expected enum `Result<(), ()>`
            found unit type `()`
@@ -38,3 +36,12 @@ help: try adding an expression at the end of the block
 23 ~     Ok(());;
 24 +     Ok(())
    |
+
+error[E0308]: mismatched types
+  --> tests/fail/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

tests-integration/Cargo.toml

@@ -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-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"

tests-integration/tests/macros_main.rs

@@ -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 {

tests-integration/tests/macros_select.rs

@@ -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 {

tests-integration/tests/process_stdio.rs

@@ -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");
+}

tests-integration/tests/rt_yield.rs

@@ -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 });
+}

tokio-macros/CHANGELOG.md

@@ -1,6 +1,55 @@
+# 2.1.0 (April 25th, 2023)
+
+- macros: fix typo in `#[tokio::test]` docs ([#5636])
+- macros: make entrypoints more efficient ([#5621])
+
+[#5621]: https://github.com/tokio-rs/tokio/pull/5621
+[#5636]: https://github.com/tokio-rs/tokio/pull/5636
+
+# 2.0.0 (March 24th, 2023)
+
+This major release updates the dependency on the syn crate to 2.0.0, and
+increases the MSRV to 1.56.
+
+As part of this release, we are adopting a policy of depending on a specific minor
+release of tokio-macros. This prevents Tokio from being able to pull in many different
+versions of tokio-macros.
+
+- macros: update `syn` ([#5572])
+- macros: accept path as crate rename ([#5557])
+
+[#5572]: https://github.com/tokio-rs/tokio/pull/5572
+[#5557]: https://github.com/tokio-rs/tokio/pull/5557
+
+# 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
 

tokio-macros/Cargo.toml

@@ -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 = "2.1.0"
 edition = "2018"
-rust-version = "1.49"
+rust-version = "1.56"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -24,7 +24,7 @@ proc-macro = true
 [dependencies]
 proc-macro2 = "1.0.7"
 quote = "1"
-syn = { version = "1.0.56", features = ["full"] }
+syn = { version = "2.0", features = ["full"] }
 
 [dev-dependencies]
 tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }

tokio-macros/LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2022 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

tokio-macros/src/entry.rs

@@ -1,10 +1,10 @@
-use proc_macro::TokenStream;
-use proc_macro2::{Ident, Span};
+use proc_macro2::{Span, TokenStream, TokenTree};
 use quote::{quote, quote_spanned, ToTokens};
-use syn::parse::Parser;
+use syn::parse::{Parse, ParseStream, Parser};
+use syn::{braced, Attribute, Ident, Path, Signature, Visibility};
 
 // syn::AttributeArgs does not implement syn::Parse
-type AttributeArgs = syn::punctuated::Punctuated<syn::NestedMeta, syn::Token![,]>;
+type AttributeArgs = syn::punctuated::Punctuated<syn::Meta, syn::Token![,]>;
 
 #[derive(Clone, Copy, PartialEq)]
 enum RuntimeFlavor {
@@ -29,7 +29,7 @@ struct FinalConfig {
     flavor: RuntimeFlavor,
     worker_threads: Option<usize>,
     start_paused: Option<bool>,
-    crate_name: Option<String>,
+    crate_name: Option<Path>,
 }
 
 /// Config used in case of the attribute not being able to build a valid config
@@ -47,7 +47,7 @@ struct Configuration {
     worker_threads: Option<(usize, Span)>,
     start_paused: Option<(bool, Span)>,
     is_test: bool,
-    crate_name: Option<String>,
+    crate_name: Option<Path>,
 }
 
 impl Configuration {
@@ -112,8 +112,8 @@ impl Configuration {
         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());
+        let name_path = parse_path(name, span, "crate")?;
+        self.crate_name = Some(name_path);
         Ok(())
     }
 
@@ -199,23 +199,22 @@ 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> {
+fn parse_path(lit: syn::Lit, span: Span, field: &str) -> Result<Path, syn::Error> {
     match lit {
         syn::Lit::Str(s) => {
             let err = syn::Error::new(
                 span,
                 format!(
-                    "Failed to parse value of `{}` as ident: \"{}\"",
+                    "Failed to parse value of `{}` as path: \"{}\"",
                     field,
                     s.value()
                 ),
             );
-            let path = s.parse::<syn::Path>().map_err(|_| err.clone())?;
-            path.get_ident().cloned().ok_or(err)
+            s.parse::<syn::Path>().map_err(|_| err.clone())
         }
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as ident.", field),
+            format!("Failed to parse value of `{}` as path.", field),
         )),
     }
 }
@@ -231,7 +230,7 @@ fn parse_bool(bool: syn::Lit, span: Span, field: &str) -> Result<bool, syn::Erro
 }
 
 fn build_config(
-    input: syn::ItemFn,
+    input: &ItemFn,
     args: AttributeArgs,
     is_test: bool,
     rt_multi_thread: bool,
@@ -246,7 +245,7 @@ fn build_config(
 
     for arg in args {
         match arg {
-            syn::NestedMeta::Meta(syn::Meta::NameValue(namevalue)) => {
+            syn::Meta::NameValue(namevalue) => {
                 let ident = namevalue
                     .path
                     .get_ident()
@@ -255,34 +254,26 @@ fn build_config(
                     })?
                     .to_string()
                     .to_lowercase();
+                let lit = match &namevalue.value {
+                    syn::Expr::Lit(syn::ExprLit { lit, .. }) => lit,
+                    expr => return Err(syn::Error::new_spanned(expr, "Must be a literal")),
+                };
                 match ident.as_str() {
                     "worker_threads" => {
-                        config.set_worker_threads(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
+                        config.set_worker_threads(lit.clone(), syn::spanned::Spanned::span(lit))?;
                     }
                     "flavor" => {
-                        config.set_flavor(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
+                        config.set_flavor(lit.clone(), syn::spanned::Spanned::span(lit))?;
                     }
                     "start_paused" => {
-                        config.set_start_paused(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
+                        config.set_start_paused(lit.clone(), syn::spanned::Spanned::span(lit))?;
                     }
                     "core_threads" => {
                         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),
-                        )?;
+                        config.set_crate_name(lit.clone(), syn::spanned::Spanned::span(lit))?;
                     }
                     name => {
                         let msg = format!(
@@ -293,7 +284,7 @@ fn build_config(
                     }
                 }
             }
-            syn::NestedMeta::Meta(syn::Meta::Path(path)) => {
+            syn::Meta::Path(path) => {
                 let name = path
                     .get_ident()
                     .ok_or_else(|| syn::Error::new_spanned(&path, "Must have specified ident"))?
@@ -333,18 +324,13 @@ fn build_config(
     config.build()
 }
 
-fn parse_knobs(mut input: syn::ItemFn, is_test: bool, config: FinalConfig) -> TokenStream {
+fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenStream {
     input.sig.asyncness = None;
 
     // If type mismatch occurs, the current rustc points to the last statement.
     let (last_stmt_start_span, last_stmt_end_span) = {
-        let mut last_stmt = input
-            .block
-            .stmts
-            .last()
-            .map(ToTokens::into_token_stream)
-            .unwrap_or_default()
-            .into_iter();
+        let mut last_stmt = input.stmts.last().cloned().unwrap_or_default().into_iter();
+
         // `Span` on stable Rust has a limitation that only points to the first
         // token, not the whole tokens. We can work around this limitation by
         // using the first/last span of the tokens like
@@ -354,16 +340,17 @@ 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 crate_path = config
+        .crate_name
+        .map(ToTokens::into_token_stream)
+        .unwrap_or_else(|| Ident::new("tokio", last_stmt_start_span).into_token_stream());
 
     let mut rt = match config.flavor {
         RuntimeFlavor::CurrentThread => quote_spanned! {last_stmt_start_span=>
-            #crate_ident::runtime::Builder::new_current_thread()
+            #crate_path::runtime::Builder::new_current_thread()
         },
         RuntimeFlavor::Threaded => quote_spanned! {last_stmt_start_span=>
-            #crate_ident::runtime::Builder::new_multi_thread()
+            #crate_path::runtime::Builder::new_multi_thread()
         },
     };
     if let Some(v) = config.worker_threads {
@@ -381,46 +368,53 @@ fn parse_knobs(mut input: syn::ItemFn, is_test: bool, config: FinalConfig) -> To
         quote! {}
     };
 
-    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 last_block = 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);
         }
-    })
-    .expect("Parsing failure");
-    input.block.brace_token = brace_token;
-
-    let result = quote! {
-        #header
-        #input
     };
 
-    result.into()
+    let body = input.body();
+
+    // 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_path::pin!(body);
+            let body: ::std::pin::Pin<&mut dyn ::std::future::Future<Output = #output_type>> = body;
+        }
+    } else {
+        quote! {
+            let body = async #body;
+        }
+    };
+
+    input.into_tokens(header, body, last_block)
 }
 
 fn token_stream_with_error(mut tokens: TokenStream, error: syn::Error) -> TokenStream {
-    tokens.extend(TokenStream::from(error.into_compile_error()));
+    tokens.extend(error.into_compile_error());
     tokens
 }
 
@@ -429,7 +423,7 @@ pub(crate) fn main(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
     // If any of the steps for this macro fail, we still want to expand to an item that is as close
     // to the expected output as possible. This helps out IDEs such that completions and other
     // related features keep working.
-    let input: syn::ItemFn = match syn::parse(item.clone()) {
+    let input: ItemFn = match syn::parse2(item.clone()) {
         Ok(it) => it,
         Err(e) => return token_stream_with_error(item, e),
     };
@@ -439,8 +433,8 @@ pub(crate) fn main(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
         Err(syn::Error::new_spanned(&input.sig.ident, msg))
     } else {
         AttributeArgs::parse_terminated
-            .parse(args)
-            .and_then(|args| build_config(input.clone(), args, false, rt_multi_thread))
+            .parse2(args)
+            .and_then(|args| build_config(&input, args, false, rt_multi_thread))
     };
 
     match config {
@@ -453,17 +447,17 @@ pub(crate) fn test(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
     // If any of the steps for this macro fail, we still want to expand to an item that is as close
     // to the expected output as possible. This helps out IDEs such that completions and other
     // related features keep working.
-    let input: syn::ItemFn = match syn::parse(item.clone()) {
+    let input: ItemFn = match syn::parse2(item.clone()) {
         Ok(it) => it,
         Err(e) => return token_stream_with_error(item, e),
     };
-    let config = if let Some(attr) = input.attrs.iter().find(|attr| attr.path.is_ident("test")) {
+    let config = if let Some(attr) = input.attrs().find(|attr| attr.meta.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)
-            .and_then(|args| build_config(input.clone(), args, true, rt_multi_thread))
+            .parse2(args)
+            .and_then(|args| build_config(&input, args, true, rt_multi_thread))
     };
 
     match config {
@@ -471,3 +465,127 @@ pub(crate) fn test(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
         Err(e) => token_stream_with_error(parse_knobs(input, true, DEFAULT_ERROR_CONFIG), e),
     }
 }
+
+struct ItemFn {
+    outer_attrs: Vec<Attribute>,
+    vis: Visibility,
+    sig: Signature,
+    brace_token: syn::token::Brace,
+    inner_attrs: Vec<Attribute>,
+    stmts: Vec<proc_macro2::TokenStream>,
+}
+
+impl ItemFn {
+    /// Access all attributes of the function item.
+    fn attrs(&self) -> impl Iterator<Item = &Attribute> {
+        self.outer_attrs.iter().chain(self.inner_attrs.iter())
+    }
+
+    /// Get the body of the function item in a manner so that it can be
+    /// conveniently used with the `quote!` macro.
+    fn body(&self) -> Body<'_> {
+        Body {
+            brace_token: self.brace_token,
+            stmts: &self.stmts,
+        }
+    }
+
+    /// Convert our local function item into a token stream.
+    fn into_tokens(
+        self,
+        header: proc_macro2::TokenStream,
+        body: proc_macro2::TokenStream,
+        last_block: proc_macro2::TokenStream,
+    ) -> TokenStream {
+        let mut tokens = proc_macro2::TokenStream::new();
+        header.to_tokens(&mut tokens);
+
+        // Outer attributes are simply streamed as-is.
+        for attr in self.outer_attrs {
+            attr.to_tokens(&mut tokens);
+        }
+
+        // Inner attributes require extra care, since they're not supported on
+        // blocks (which is what we're expanded into) we instead lift them
+        // outside of the function. This matches the behaviour of `syn`.
+        for mut attr in self.inner_attrs {
+            attr.style = syn::AttrStyle::Outer;
+            attr.to_tokens(&mut tokens);
+        }
+
+        self.vis.to_tokens(&mut tokens);
+        self.sig.to_tokens(&mut tokens);
+
+        self.brace_token.surround(&mut tokens, |tokens| {
+            body.to_tokens(tokens);
+            last_block.to_tokens(tokens);
+        });
+
+        tokens
+    }
+}
+
+impl Parse for ItemFn {
+    #[inline]
+    fn parse(input: ParseStream<'_>) -> syn::Result<Self> {
+        // This parse implementation has been largely lifted from `syn`, with
+        // the exception of:
+        // * We don't have access to the plumbing necessary to parse inner
+        //   attributes in-place.
+        // * We do our own statements parsing to avoid recursively parsing
+        //   entire statements and only look for the parts we're interested in.
+
+        let outer_attrs = input.call(Attribute::parse_outer)?;
+        let vis: Visibility = input.parse()?;
+        let sig: Signature = input.parse()?;
+
+        let content;
+        let brace_token = braced!(content in input);
+        let inner_attrs = Attribute::parse_inner(&content)?;
+
+        let mut buf = proc_macro2::TokenStream::new();
+        let mut stmts = Vec::new();
+
+        while !content.is_empty() {
+            if let Some(semi) = content.parse::<Option<syn::Token![;]>>()? {
+                semi.to_tokens(&mut buf);
+                stmts.push(buf);
+                buf = proc_macro2::TokenStream::new();
+                continue;
+            }
+
+            // Parse a single token tree and extend our current buffer with it.
+            // This avoids parsing the entire content of the sub-tree.
+            buf.extend([content.parse::<TokenTree>()?]);
+        }
+
+        if !buf.is_empty() {
+            stmts.push(buf);
+        }
+
+        Ok(Self {
+            outer_attrs,
+            vis,
+            sig,
+            brace_token,
+            inner_attrs,
+            stmts,
+        })
+    }
+}
+
+struct Body<'a> {
+    brace_token: syn::token::Brace,
+    // Statements, with terminating `;`.
+    stmts: &'a [TokenStream],
+}
+
+impl ToTokens for Body<'_> {
+    fn to_tokens(&self, tokens: &mut proc_macro2::TokenStream) {
+        self.brace_token.surround(tokens, |tokens| {
+            for stmt in self.stmts {
+                stmt.to_tokens(tokens);
+            }
+        })
+    }
+}

tokio-macros/src/lib.rs

@@ -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
@@ -197,7 +204,7 @@ use proc_macro::TokenStream;
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
-    entry::main(args, item, true)
+    entry::main(args.into(), item.into(), true).into()
 }
 
 /// Marks async function to be executed by selected runtime. This macro helps set up a `Runtime`
@@ -262,14 +269,23 @@ pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
 #[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)
+    entry::main(args.into(), item.into(), false).into()
 }
 
-/// 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)]
@@ -278,9 +294,16 @@ 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.
 ///
-/// 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]
@@ -289,6 +312,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
@@ -298,6 +396,22 @@ 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.
 ///
 /// ### Rename package
@@ -312,7 +426,7 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// ```
 #[proc_macro_attribute]
 pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
-    entry::test(args, item, true)
+    entry::test(args.into(), item.into(), true).into()
 }
 
 /// Marks async function to be executed by runtime, suitable to test environment
@@ -327,7 +441,7 @@ pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
 /// ```
 #[proc_macro_attribute]
 pub fn test_rt(args: TokenStream, item: TokenStream) -> TokenStream {
-    entry::test(args, item, false)
+    entry::test(args.into(), item.into(), false).into()
 }
 
 /// Always fails with the error message below.

tokio-macros/src/select.rs

@@ -1,7 +1,7 @@
 use proc_macro::{TokenStream, TokenTree};
 use proc_macro2::Span;
 use quote::quote;
-use syn::Ident;
+use syn::{parse::Parser, Ident};
 
 pub(crate) fn declare_output_enum(input: TokenStream) -> TokenStream {
     // passed in is: `(_ _ _)` with one `_` per branch
@@ -46,7 +46,7 @@ pub(crate) fn clean_pattern_macro(input: TokenStream) -> TokenStream {
     // If this isn't a pattern, we return the token stream as-is. The select!
     // macro is using it in a location requiring a pattern, so an error will be
     // emitted there.
-    let mut input: syn::Pat = match syn::parse(input.clone()) {
+    let mut input: syn::Pat = match syn::Pat::parse_single.parse(input.clone()) {
         Ok(it) => it,
         Err(_) => return input,
     };
@@ -58,7 +58,6 @@ pub(crate) fn clean_pattern_macro(input: TokenStream) -> TokenStream {
 // Removes any occurrences of ref or mut in the provided pattern.
 fn clean_pattern(pat: &mut syn::Pat) {
     match pat {
-        syn::Pat::Box(_box) => {}
         syn::Pat::Lit(_literal) => {}
         syn::Pat::Macro(_macro) => {}
         syn::Pat::Path(_path) => {}
@@ -94,16 +93,16 @@ fn clean_pattern(pat: &mut syn::Pat) {
             }
         }
         syn::Pat::TupleStruct(tuple) => {
-            for elem in tuple.pat.elems.iter_mut() {
+            for elem in tuple.elems.iter_mut() {
                 clean_pattern(elem);
             }
         }
         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);
         }
         _ => {}
     }

tokio-stream/CHANGELOG.md

@@ -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])

tokio-stream/Cargo.toml

@@ -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"
-edition = "2018"
-rust-version = "1.49"
+version = "0.1.12"
+edition = "2021"
+rust-version = "1.56"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -18,6 +18,16 @@ categories = ["asynchronous"]
 
 [features]
 default = ["time"]
+
+full = [
+    "time",
+    "net",
+    "io-util",
+    "fs",
+    "sync",
+    "signal"
+]
+
 time = ["tokio/time"]
 net = ["tokio/net"]
 io-util = ["tokio/io-util"]
@@ -34,11 +44,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"]

tokio-stream/LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2022 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

tokio-stream/fuzz/.gitignore

@@ -0,0 +1,4 @@
+target
+corpus
+artifacts
+coverage

tokio-stream/fuzz/Cargo.toml

@@ -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

tokio-stream/fuzz/fuzz_targets/fuzz_stream_map.rs

@@ -0,0 +1,81 @@
+#![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: [bool; 64]| {
+    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)
+        }
+    }
+
+    // Try the test with each possible length.
+    for len in 0..data.len() {
+        let mut map = task::spawn(StreamMap::new());
+        let mut expect = 0;
+
+        for (i, is_empty) in data[..len].iter().copied().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);
+            }
+        }
+    }
+});

tokio-stream/src/lib.rs

@@ -96,5 +96,8 @@ pub use pending::{pending, Pending};
 mod stream_map;
 pub use stream_map::StreamMap;
 
+mod stream_close;
+pub use stream_close::StreamNotifyClose;
+
 #[doc(no_inline)]
 pub use futures_core::Stream;

tokio-stream/src/stream_close.rs

@@ -0,0 +1,93 @@
+use crate::Stream;
+use pin_project_lite::pin_project;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A `Stream` that wraps the values in an `Option`.
+    ///
+    /// Whenever the wrapped stream yields an item, this stream yields that item
+    /// wrapped in `Some`. When the inner stream ends, then this stream first
+    /// yields a `None` item, and then this stream will also end.
+    ///
+    /// # Example
+    ///
+    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`.
+    ///
+    /// ```
+    /// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let mut map = StreamMap::new();
+    ///     let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
+    ///     let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
+    ///     map.insert(0, stream);
+    ///     map.insert(1, stream2);
+    ///     while let Some((key, val)) = map.next().await {
+    ///         match val {
+    ///             Some(val) => println!("got {val:?} from stream {key:?}"),
+    ///             None => println!("stream {key:?} closed"),
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    #[must_use = "streams do nothing unless polled"]
+    pub struct StreamNotifyClose<S> {
+        #[pin]
+        inner: Option<S>,
+    }
+}
+
+impl<S> StreamNotifyClose<S> {
+    /// Create a new `StreamNotifyClose`.
+    pub fn new(stream: S) -> Self {
+        Self {
+            inner: Some(stream),
+        }
+    }
+
+    /// Get back the inner `Stream`.
+    ///
+    /// Returns `None` if the stream has reached its end.
+    pub fn into_inner(self) -> Option<S> {
+        self.inner
+    }
+}
+
+impl<S> Stream for StreamNotifyClose<S>
+where
+    S: Stream,
+{
+    type Item = Option<S::Item>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // We can't invoke poll_next after it ended, so we unset the inner stream as a marker.
+        match self
+            .as_mut()
+            .project()
+            .inner
+            .as_pin_mut()
+            .map(|stream| S::poll_next(stream, cx))
+        {
+            Some(Poll::Ready(Some(item))) => Poll::Ready(Some(Some(item))),
+            Some(Poll::Ready(None)) => {
+                self.project().inner.set(None);
+                Poll::Ready(Some(None))
+            }
+            Some(Poll::Pending) => Poll::Pending,
+            None => Poll::Ready(None),
+        }
+    }
+
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if let Some(inner) = &self.inner {
+            // We always return +1 because when there's stream there's atleast one more item.
+            let (l, u) = inner.size_hint();
+            (l.saturating_add(1), u.and_then(|u| u.checked_add(1)))
+        } else {
+            (0, Some(0))
+        }
+    }
+}

tokio-stream/src/stream_ext.rs

@@ -57,10 +57,14 @@ use try_next::TryNext;
 
 cfg_time! {
     pub(crate) mod timeout;
+    pub(crate) mod timeout_repeating;
     use timeout::Timeout;
-    use tokio::time::Duration;
+    use timeout_repeating::TimeoutRepeating;
+    use tokio::time::{Duration, Interval};
     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 +117,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 +159,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
     ///
     /// ```
@@ -910,7 +926,9 @@ pub trait StreamExt: Stream {
     /// If the wrapped stream yields a value before the deadline is reached, the
     /// value is returned. Otherwise, an error is returned. The caller may decide
     /// to continue consuming the stream and will eventually get the next source
-    /// stream value once it becomes available.
+    /// stream value once it becomes available. See
+    /// [`timeout_repeating`](StreamExt::timeout_repeating) for an alternative
+    /// where the timeouts will repeat.
     ///
     /// # Notes
     ///
@@ -957,6 +975,25 @@ pub trait StreamExt: Stream {
     /// assert_eq!(int_stream.try_next().await, Ok(None));
     /// # }
     /// ```
+    ///
+    /// Once a timeout error is received, no further events will be received
+    /// unless the wrapped stream yields a value (timeouts do not repeat).
+    ///
+    /// ```
+    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
+    /// # async fn main() {
+    /// use tokio_stream::{StreamExt, wrappers::IntervalStream};
+    /// use std::time::Duration;
+    /// let interval_stream = IntervalStream::new(tokio::time::interval(Duration::from_millis(100)));
+    /// let timeout_stream = interval_stream.timeout(Duration::from_millis(10));
+    /// tokio::pin!(timeout_stream);
+    ///
+    /// // Only one timeout will be received between values in the source stream.
+    /// assert!(timeout_stream.try_next().await.is_ok());
+    /// assert!(timeout_stream.try_next().await.is_err(), "expected one timeout");
+    /// assert!(timeout_stream.try_next().await.is_ok(), "expected no more timeouts");
+    /// # }
+    /// ```
     #[cfg(all(feature = "time"))]
     #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
     fn timeout(self, duration: Duration) -> Timeout<Self>
@@ -966,8 +1003,99 @@ pub trait StreamExt: Stream {
         Timeout::new(self, duration)
     }
 
+    /// Applies a per-item timeout to the passed stream.
+    ///
+    /// `timeout_repeating()` takes an [`Interval`](tokio::time::Interval) that
+    /// controls the time each element of the stream has to complete before
+    /// timing out.
+    ///
+    /// If the wrapped stream yields a value before the deadline is reached, the
+    /// value is returned. Otherwise, an error is returned. The caller may decide
+    /// to continue consuming the stream and will eventually get the next source
+    /// stream value once it becomes available. Unlike `timeout()`, if no value
+    /// becomes available before the deadline is reached, additional errors are
+    /// returned at the specified interval. See [`timeout`](StreamExt::timeout)
+    /// for an alternative where the timeouts do not repeat.
+    ///
+    /// # Notes
+    ///
+    /// This function consumes the stream passed into it and returns a
+    /// wrapped version of it.
+    ///
+    /// Polling the returned stream will continue to poll the inner stream even
+    /// if one or more items time out.
+    ///
+    /// # Examples
+    ///
+    /// Suppose we have a stream `int_stream` that yields 3 numbers (1, 2, 3):
+    ///
+    /// ```
+    /// # #[tokio::main]
+    /// # async fn main() {
+    /// use tokio_stream::{self as stream, StreamExt};
+    /// use std::time::Duration;
+    /// # let int_stream = stream::iter(1..=3);
+    ///
+    /// let int_stream = int_stream.timeout_repeating(tokio::time::interval(Duration::from_secs(1)));
+    /// tokio::pin!(int_stream);
+    ///
+    /// // When no items time out, we get the 3 elements in succession:
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(2)));
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(3)));
+    /// assert_eq!(int_stream.try_next().await, Ok(None));
+    ///
+    /// // If the second item times out, we get an error and continue polling the stream:
+    /// # let mut int_stream = stream::iter(vec![Ok(1), Err(()), Ok(2), Ok(3)]);
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
+    /// assert!(int_stream.try_next().await.is_err());
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(2)));
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(3)));
+    /// assert_eq!(int_stream.try_next().await, Ok(None));
+    ///
+    /// // If we want to stop consuming the source stream the first time an
+    /// // element times out, we can use the `take_while` operator:
+    /// # let int_stream = stream::iter(vec![Ok(1), Err(()), Ok(2), Ok(3)]);
+    /// let mut int_stream = int_stream.take_while(Result::is_ok);
+    ///
+    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
+    /// assert_eq!(int_stream.try_next().await, Ok(None));
+    /// # }
+    /// ```
+    ///
+    /// Timeout errors will be continuously produced at the specified interval
+    /// until the wrapped stream yields a value.
+    ///
+    /// ```
+    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
+    /// # async fn main() {
+    /// use tokio_stream::{StreamExt, wrappers::IntervalStream};
+    /// use std::time::Duration;
+    /// let interval_stream = IntervalStream::new(tokio::time::interval(Duration::from_millis(23)));
+    /// let timeout_stream = interval_stream.timeout_repeating(tokio::time::interval(Duration::from_millis(9)));
+    /// tokio::pin!(timeout_stream);
+    ///
+    /// // Multiple timeouts will be received between values in the source stream.
+    /// assert!(timeout_stream.try_next().await.is_ok());
+    /// assert!(timeout_stream.try_next().await.is_err(), "expected one timeout");
+    /// assert!(timeout_stream.try_next().await.is_err(), "expected a second timeout");
+    /// // Will eventually receive another value from the source stream...
+    /// assert!(timeout_stream.try_next().await.is_ok(), "expected non-timeout");
+    /// # }
+    /// ```
+    #[cfg(all(feature = "time"))]
+    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
+    fn timeout_repeating(self, interval: Interval) -> TimeoutRepeating<Self>
+    where
+        Self: Sized,
+    {
+        TimeoutRepeating::new(self, interval)
+    }
+
     /// 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 +1121,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 +1185,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,
     };

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -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)
+    }
+}

tokio-stream/src/stream_ext/collect.rs

@@ -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())
         }
     }
 }

tokio-stream/src/stream_ext/next.rs

@@ -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> {

tokio-stream/src/stream_ext/then.rs

@@ -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);

tokio-stream/src/stream_ext/throttle.rs

@@ -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 {

tokio-stream/src/stream_ext/timeout.rs

@@ -23,8 +23,8 @@ pin_project! {
     }
 }
 
-/// Error returned by `Timeout`.
-#[derive(Debug, PartialEq)]
+/// Error returned by `Timeout` and `TimeoutRepeating`.
+#[derive(Debug, PartialEq, Eq)]
 pub struct Elapsed(());
 
 impl<S: Stream> Timeout<S> {

tokio-stream/src/stream_ext/timeout_repeating.rs

@@ -0,0 +1,56 @@
+use crate::stream_ext::Fuse;
+use crate::{Elapsed, Stream};
+use tokio::time::Interval;
+
+use core::pin::Pin;
+use core::task::{Context, Poll};
+use pin_project_lite::pin_project;
+
+pin_project! {
+    /// Stream returned by the [`timeout_repeating`](super::StreamExt::timeout_repeating) method.
+    #[must_use = "streams do nothing unless polled"]
+    #[derive(Debug)]
+    pub struct TimeoutRepeating<S> {
+        #[pin]
+        stream: Fuse<S>,
+        #[pin]
+        interval: Interval,
+    }
+}
+
+impl<S: Stream> TimeoutRepeating<S> {
+    pub(super) fn new(stream: S, interval: Interval) -> Self {
+        TimeoutRepeating {
+            stream: Fuse::new(stream),
+            interval,
+        }
+    }
+}
+
+impl<S: Stream> Stream for TimeoutRepeating<S> {
+    type Item = Result<S::Item, Elapsed>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut me = self.project();
+
+        match me.stream.poll_next(cx) {
+            Poll::Ready(v) => {
+                if v.is_some() {
+                    me.interval.reset();
+                }
+                return Poll::Ready(v.map(Ok));
+            }
+            Poll::Pending => {}
+        };
+
+        ready!(me.interval.poll_tick(cx));
+        Poll::Ready(Some(Err(Elapsed::new())))
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let (lower, _) = self.stream.size_hint();
+
+        // The timeout stream may insert an error an infinite number of times.
+        (lower, None)
+    }
+}

tokio-stream/src/stream_ext/try_next.rs

@@ -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> {

tokio-stream/src/stream_map.rs

@@ -42,10 +42,18 @@ use std::task::{Context, Poll};
 /// to be merged, it may be advisable to use tasks sending values on a shared
 /// [`mpsc`] channel.
 ///
+/// # Notes
+///
+/// `StreamMap` removes finished streams automatically, without alerting the user.
+/// In some scenarios, the caller would want to know on closed streams.
+/// To do this, use [`StreamNotifyClose`] as a wrapper to your stream.
+/// It will return None when the stream is closed.
+///
 /// [`StreamExt::merge`]: crate::StreamExt::merge
 /// [`mpsc`]: https://docs.rs/tokio/1.0/tokio/sync/mpsc/index.html
 /// [`pin!`]: https://docs.rs/tokio/1.0/tokio/macro.pin.html
 /// [`Box::pin`]: std::boxed::Box::pin
+/// [`StreamNotifyClose`]: crate::StreamNotifyClose
 ///
 /// # Examples
 ///
@@ -170,6 +178,28 @@ use std::task::{Context, Poll};
 ///     }
 /// }
 /// ```
+///
+/// Using `StreamNotifyClose` to handle closed streams with `StreamMap`.
+///
+/// ```
+/// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
+///
+/// #[tokio::main]
+/// async fn main() {
+///     let mut map = StreamMap::new();
+///     let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
+///     let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
+///     map.insert(0, stream);
+///     map.insert(1, stream2);
+///     while let Some((key, val)) = map.next().await {
+///         match val {
+///             Some(val) => println!("got {val:?} from stream {key:?}"),
+///             None => println!("stream {key:?} closed"),
+///         }
+///     }
+/// }
+/// ```
+
 #[derive(Debug)]
 pub struct StreamMap<K, V> {
     /// Streams stored in the map
@@ -568,7 +598,7 @@ where
     }
 }
 
-impl<K, V> std::iter::FromIterator<(K, V)> for StreamMap<K, V>
+impl<K, V> FromIterator<(K, V)> for StreamMap<K, V>
 where
     K: Hash + Eq,
 {

tokio-stream/src/wrappers/broadcast.rs

@@ -18,7 +18,7 @@ pub struct BroadcastStream<T> {
 }
 
 /// An error returned from the inner stream of a [`BroadcastStream`].
-#[derive(Debug, PartialEq, Clone)]
+#[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.

tokio-stream/src/wrappers/watch.rs

@@ -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"));
 /// # }
@@ -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> {

tokio-stream/tests/chunks_timeout.rs

@@ -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]));
+}

tokio-stream/tests/stream_close.rs

@@ -0,0 +1,11 @@
+use tokio_stream::{StreamExt, StreamNotifyClose};
+
+#[tokio::test]
+async fn basic_usage() {
+    let mut stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
+
+    assert_eq!(stream.next().await, Some(Some(0)));
+    assert_eq!(stream.next().await, Some(Some(1)));
+    assert_eq!(stream.next().await, Some(None));
+    assert_eq!(stream.next().await, None);
+}

tokio-stream/tests/stream_panic.rs

@@ -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(())
+}

tokio-stream/tests/stream_stream_map.rs

@@ -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)
 }

tokio-stream/tests/watch.rs

@@ -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");
+}

tokio-test/Cargo.toml

@@ -5,8 +5,8 @@ name = "tokio-test"
 # - Update CHANGELOG.md.
 # - Create "tokio-test-0.4.x" git tag.
 version = "0.4.2"
-edition = "2018"
-rust-version = "1.49"
+edition = "2021"
+rust-version = "1.56"
 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"

tokio-test/LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2022 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

tokio-test/src/io.rs

@@ -310,6 +310,8 @@ impl Inner {
 
                         if now < until {
                             break;
+                        } else {
+                            self.waiting = None;
                         }
                     } else {
                         self.waiting = Some(Instant::now() + *dur);

tokio-test/src/macros.rs

@@ -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};

tokio-test/src/task.rs

@@ -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))

tokio-test/tests/io.rs

@@ -2,6 +2,7 @@
 
 use std::io;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::time::{Duration, Instant};
 use tokio_test::io::Builder;
 
 #[tokio::test]
@@ -84,3 +85,65 @@ async fn mock_panics_write_data_left() {
     use tokio_test::io::Builder;
     Builder::new().write(b"write").build();
 }
+
+#[tokio::test(start_paused = true)]
+async fn wait() {
+    const FIRST_WAIT: Duration = Duration::from_secs(1);
+
+    let mut mock = Builder::new()
+        .wait(FIRST_WAIT)
+        .read(b"hello ")
+        .read(b"world!")
+        .build();
+
+    let mut buf = [0; 256];
+
+    let start = Instant::now(); // record the time the read call takes
+                                //
+    let n = mock.read(&mut buf).await.expect("read 1");
+    assert_eq!(&buf[..n], b"hello ");
+    println!("time elapsed after first read {:?}", start.elapsed());
+
+    let n = mock.read(&mut buf).await.expect("read 2");
+    assert_eq!(&buf[..n], b"world!");
+    println!("time elapsed after second read {:?}", start.elapsed());
+
+    // make sure the .wait() instruction worked
+    assert!(
+        start.elapsed() >= FIRST_WAIT,
+        "consuming the whole mock only took {}ms",
+        start.elapsed().as_millis()
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn multiple_wait() {
+    const FIRST_WAIT: Duration = Duration::from_secs(1);
+    const SECOND_WAIT: Duration = Duration::from_secs(1);
+
+    let mut mock = Builder::new()
+        .wait(FIRST_WAIT)
+        .read(b"hello ")
+        .wait(SECOND_WAIT)
+        .read(b"world!")
+        .build();
+
+    let mut buf = [0; 256];
+
+    let start = Instant::now(); // record the time it takes to consume the mock
+
+    let n = mock.read(&mut buf).await.expect("read 1");
+    assert_eq!(&buf[..n], b"hello ");
+    println!("time elapsed after first read {:?}", start.elapsed());
+
+    let n = mock.read(&mut buf).await.expect("read 2");
+    assert_eq!(&buf[..n], b"world!");
+    println!("time elapsed after second read {:?}", start.elapsed());
+
+    // make sure the .wait() instruction worked
+    assert!(
+        start.elapsed() >= FIRST_WAIT + SECOND_WAIT,
+        "consuming the whole mock only took {}ms",
+        start.elapsed().as_millis()
+    );
+}

tokio-util/CHANGELOG.md

@@ -1,3 +1,96 @@
+# 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
@@ -49,6 +142,12 @@
 [#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

tokio-util/Cargo.toml

@@ -4,9 +4,9 @@ name = "tokio-util"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.1"
-edition = "2018"
-rust-version = "1.49"
+version = "0.7.7"
+edition = "2021"
+rust-version = "1.56"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -29,13 +29,12 @@ codec = ["tracing"]
 time = ["tokio/time","slab"]
 io = []
 io-util = ["io", "tokio/rt", "tokio/io-util"]
-rt = ["tokio/rt", "tokio/sync", "futures-util"]
+rt = ["tokio/rt", "tokio/sync", "futures-util", "hashbrown"]
 
 __docs_rs = ["futures-util"]
 
 [dependencies]
-tokio = { version = "1.6.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"
@@ -45,6 +44,9 @@ pin-project-lite = "0.2.0"
 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"] }
 tokio-test = { version = "0.4.0", path = "../tokio-test" }
@@ -53,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"]

tokio-util/LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2022 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

tokio-util/src/codec/decoder.rs

@@ -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.

tokio-util/src/codec/framed.rs

@@ -253,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

tokio-util/src/codec/framed_impl.rs

@@ -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,7 +281,7 @@ where
         let mut pinned = self.project();
 
         while !pinned.state.borrow_mut().buffer.is_empty() {
-            let WriteFrame { buffer } = pinned.state.borrow_mut();
+            let WriteFrame { buffer, .. } = pinned.state.borrow_mut();
             trace!(remaining = buffer.len(), "writing;");
 
             let n = ready!(poll_write_buf(pinned.inner.as_mut(), cx, buffer))?;

tokio-util/src/codec/framed_write.rs

@@ -123,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

tokio-util/src/codec/length_delimited.rs

@@ -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))
             }

tokio-util/src/io/copy_to_bytes.rs

@@ -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)
+    }
+}

tokio-util/src/io/inspect.rs

@@ -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()
+    }
+}

tokio-util/src/io/mod.rs

@@ -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};

tokio-util/src/io/sink_writer.rs

@@ -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)
+    }
+}

tokio-util/src/io/stream_reader.rs

@@ -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>
@@ -93,7 +191,7 @@ where
     }
 
     /// Consumes this `StreamReader`, returning a Tuple consisting
-    /// of the underlying stream and an Option of the interal buffer,
+    /// 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() {
@@ -201,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,
+        }
+    }
+}

tokio-util/src/io/sync_bridge.rs

@@ -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())
     }

tokio-util/src/lib.rs

@@ -29,6 +29,7 @@ cfg_codec! {
 }
 
 cfg_net! {
+    #[cfg(not(target_arch = "wasm32"))]
     pub mod udp;
     pub mod net;
 }
@@ -115,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();

tokio-util/src/sync/cancellation_token.rs

@@ -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.
@@ -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 =====
 
@@ -91,44 +97,19 @@ impl core::fmt::Debug for CancellationToken {
 }
 
 impl Clone for CancellationToken {
+    /// Creates a clone of the `CancellationToken` which will get cancelled
+    /// whenever the current token gets cancelled, and vice versa.
     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,31 +122,14 @@ 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.
+    /// current token gets cancelled. Unlike a cloned `CancellationToken`,
+    /// cancelling a child token does not cancel the parent token.
     ///
     /// If the current token is already cancelled, the child token will get
     /// returned in cancelled state.
@@ -203,56 +167,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 +176,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 +228,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 +242,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;
     }
 }

tokio-util/src/sync/cancellation_token/tree_node.rs

@@ -0,0 +1,365 @@
+//! 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,
+{
+    use std::sync::TryLockError;
+
+    let mut locked_node = node.inner.lock().unwrap();
+
+    // Every time this fails, the number of ancestors of the node decreases,
+    // so the loop must succeed after a finite number of iterations.
+    loop {
+        // Look up the parent of the currently locked node.
+        let potential_parent = match locked_node.parent.as_ref() {
+            Some(potential_parent) => potential_parent.clone(),
+            None => return func(locked_node, None),
+        };
+
+        // Lock the parent. This may require unlocking the child first.
+        let locked_parent = match potential_parent.inner.try_lock() {
+            Ok(locked_parent) => locked_parent,
+            Err(TryLockError::WouldBlock) => {
+                drop(locked_node);
+                // Deadlock safety:
+                //
+                // Due to invariant #2, the potential parent must come before
+                // the child in the creation order. Therefore, we can safely
+                // lock the child while holding the parent lock.
+                let locked_parent = potential_parent.inner.lock().unwrap();
+                locked_node = node.inner.lock().unwrap();
+                locked_parent
+            }
+            Err(TryLockError::Poisoned(err)) => Err(err).unwrap(),
+        };
+
+        // If we unlocked the child, then the parent may have changed. Check
+        // that we still have the right parent.
+        if let Some(actual_parent) = locked_node.parent.as_ref() {
+            if Arc::ptr_eq(actual_parent, &potential_parent) {
+                return func(locked_node, Some(locked_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() {
+        parent.children.shrink_to(2 * len);
+    }
+}
+
+/// 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();
+}

tokio-util/src/sync/intrusive_double_linked_list.rs

@@ -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));
-            }
-        }
-    }
-}

tokio-util/src/sync/mod.rs

@@ -1,9 +1,9 @@
 //! 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::{PollSendError, PollSender};

tokio-util/src/sync/mpsc.rs

@@ -136,6 +136,7 @@ impl<T: Send + 'static> PollSender<T> {
     ///
     /// 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 => {

tokio-util/src/sync/poll_semaphore.rs

@@ -12,7 +12,10 @@ use super::ReusableBoxFuture;
 /// [`Semaphore`]: tokio::sync::Semaphore
 pub struct PollSemaphore {
     semaphore: Arc<Semaphore>,
-    permit_fut: Option<ReusableBoxFuture<'static, 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
     }
 }

tokio-util/src/sync/reusable_box.rs

@@ -1,17 +1,18 @@
 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>>`.
 ///
 /// This type lets you replace the future stored in the box without
 /// reallocating when the size and alignment permits this.
 pub struct ReusableBoxFuture<'a, T> {
-    boxed: NonNull<dyn Future<Output = T> + Send + 'a>,
+    boxed: Pin<Box<dyn Future<Output = T> + Send + 'a>>,
 }
 
 impl<'a, T> ReusableBoxFuture<'a, T> {
@@ -20,11 +21,9 @@ impl<'a, T> ReusableBoxFuture<'a, T> {
     where
         F: Future<Output = T> + Send + 'a,
     {
-        let boxed: Box<dyn Future<Output = T> + Send + 'a> = Box::new(future);
-
-        let boxed = NonNull::from(Box::leak(boxed));
-
-        Self { boxed }
+        Self {
+            boxed: Box::pin(future),
+        }
     }
 
     /// Replace the future currently stored in this box.
@@ -49,62 +48,29 @@ impl<'a, T> ReusableBoxFuture<'a, T> {
     where
         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 + 'a,
-    {
-        // 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.
@@ -122,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> {}
 
-// 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> {
     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
+    }
+}