chore: prepare tokio-util v0.7.13 (#7012)

In #6112, tests for the tracing instrumentation were introduced. They
had to live in their own test crate under `tokio/tests` because the
`tracing-mock` crate that the tests use had not yet been published to
crates.io.

Now `tracing-mock` has been published to crates.io and so the separate
test crate and separate job to run it are no longer necessary. The
tracing instrumentation tests can be placed in with the other
integration tests in the `tokio` crate.

The tests themselves have also been updated to match the changes in the
`tracing-mock` API since the version which was being used.

.cirrus.yml

@@ -1,10 +1,10 @@
 only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
 auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
 freebsd_instance:
-  image_family: freebsd-14-0
+  image_family: freebsd-14-1
 env:
   RUST_STABLE: stable
-  RUST_NIGHTLY: nightly-2023-10-21
+  RUST_NIGHTLY: nightly-2024-05-05
   RUSTFLAGS: -D warnings
 
 # Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the

.github/buildomat/README.md

@@ -0,0 +1,20 @@
+# Buildomat illumos CI
+
+This directory contains CI configurations for the [illumos] operating system.
+Tokio's illumos CI jobs are run using [Buildomat], a CI system developed by
+Oxide Computer, which supports illumos. See [the Buildomat README] for more
+details.
+
+## illumos-Specific CI Failures
+
+If your pull request's CI build fails on illumos, and you aren't able to easily
+reproduce the failure on other operating systems, don't worry! The
+[tokio-rs/illumos] team is responsible for maintaining Tokio's illumos support,
+and can be called on to assist contributors with illumos-specific issues. Please
+feel free to tag @tokio-rs/illumos to ask for help resolving build failures on
+illumos 
+
+[illumos]: https://www.illumos.org/
+[Buildomat]: https://github.com/oxidecomputer/buildomat
+[the Buildomat README]: https://github.com/oxidecomputer/buildomat
+[tokio-rs/illumos]: https://github.com/orgs/tokio-rs/teams/illumos

.github/buildomat/config.toml

@@ -0,0 +1,8 @@
+# Repository-level Buildomat configuration.
+# See: https://github.com/oxidecomputer/buildomat#per-repository-configuration
+
+# Enable buildomat. This one should be self-explanatory.
+enable = true
+# Allow CI runs for PRs from users outside the `tokio-rs` organization. Our
+# buildomat jobs don't touch any secrets/keys, so this should be fine.
+org_only = false

.github/workflows/audit.yml

@@ -13,18 +13,12 @@ permissions:
   contents: read
 
 jobs:
-  security-audit:
+  cargo-deny:
     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
+      checks: write
+      contents: read
+      issues: write
     runs-on: ubuntu-latest
-    if: "!contains(github.event.head_commit.message, 'ci skip')"
     steps:
-      - uses: actions/checkout@v4
-
-      - name: Audit Check
-        # https://github.com/rustsec/audit-check/issues/2
-        uses: rustsec/audit-check@master
-        with:
-          token: ${{ secrets.GITHUB_TOKEN }}
+    - uses: actions/checkout@v4
+    - uses: EmbarkStudios/cargo-deny-action@v2

.github/workflows/ci.yml

@@ -13,10 +13,13 @@ concurrency:
 env:
   RUSTFLAGS: -Dwarnings
   RUST_BACKTRACE: 1
+  RUSTUP_WINDOWS_PATH_ADD_BIN: 1
   # Change to specific Rust release to pin
   rust_stable: stable
-  rust_nightly: nightly-2023-10-21
-  rust_clippy: '1.76'
+  rust_nightly: nightly-2024-05-05
+  # Pin a specific miri version
+  rust_miri_nightly: nightly-2024-10-21
+  rust_clippy: '1.77'
   # When updating this, also update:
   # - README.md
   # - tokio/README.md
@@ -25,7 +28,7 @@ env:
   # - tokio-util/Cargo.toml
   # - tokio-test/Cargo.toml
   # - tokio-stream/Cargo.toml
-  rust_min: '1.63'
+  rust_min: '1.70'
 
 defaults:
   run:
@@ -44,7 +47,6 @@ jobs:
       - test-workspace-all-features
       - test-integration-tests-per-feature
       - test-parking_lot
-      - test-tracing-instrumentation
       - valgrind
       - test-unstable
       - miri
@@ -64,10 +66,11 @@ jobs:
       - loom-compile
       - check-readme
       - test-hyper
+      - test-quinn
       - x86_64-fortanix-unknown-sgx
       - check-redox
       - wasm32-unknown-unknown
-      - wasm32-wasi
+      - wasm32-wasip1
       - check-external-types
       - check-fuzzing
       - check-unstable-mt-counters
@@ -176,7 +179,7 @@ jobs:
         run: |
           set -euxo pipefail
           RUSTFLAGS="$RUSTFLAGS -C panic=abort -Zpanic-abort-tests" cargo nextest run --workspace --exclude tokio-macros --exclude tests-build --all-features --tests
-  
+
   test-integration-tests-per-feature:
     needs: basics
     name: Run integration tests for each feature
@@ -242,34 +245,6 @@ jobs:
       - name: Check tests with all features enabled
         run: cargo check --workspace --all-features --tests
 
-  test-tracing-instrumentation:
-    # These tests use the as-yet unpublished `tracing-mock` crate to test the
-    # tracing instrumentation present in Tokio. As such they are placed in
-    # their own test crate outside of the workspace.
-    needs: basics
-    name: test tokio instrumentation
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@stable
-        with:
-          toolchain: ${{ env.rust_stable }}
-      - name: Install cargo-nextest
-        uses: taiki-e/install-action@v2
-        with:
-          tool: cargo-nextest
-
-      - uses: Swatinem/rust-cache@v2
-
-      - name: test tracing-instrumentation
-        run: |
-          set -euxo pipefail
-          cargo nextest run
-        working-directory: tokio/tests/tracing-instrumentation
-        env:
-          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
-
   valgrind:
     name: valgrind
     needs: basics
@@ -279,7 +254,7 @@ jobs:
       - name: Install Rust ${{ env.rust_stable }}
         uses: dtolnay/rust-toolchain@stable
         with:
-            toolchain: ${{ env.rust_stable }}
+            toolchain: 1.82
 
       - name: Install Valgrind
         uses: taiki-e/install-action@valgrind
@@ -411,17 +386,19 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_nightly }}
+      - name: Install Rust ${{ env.rust_miri_nightly }}
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: ${{ env.rust_nightly }}
+          toolchain: ${{ env.rust_miri_nightly }}
           components: miri
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-nextest
       - uses: Swatinem/rust-cache@v2
       - name: miri
-        # 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
+          cargo miri nextest run --features full --lib --tests --no-fail-fast
         working-directory: tokio
         env:
           MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
@@ -762,7 +739,15 @@ jobs:
 
   docs:
     name: docs
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.run.os }}
+    strategy:
+      matrix:
+        run:
+          - os: windows-latest
+          - os: ubuntu-latest
+            RUSTFLAGS: --cfg tokio_taskdump
+            RUSTDOCFLAGS: --cfg tokio_taskdump
+
     steps:
       - uses: actions/checkout@v4
       - name: Install Rust ${{ env.rust_nightly }}
@@ -774,8 +759,8 @@ jobs:
         run: |
           cargo doc --lib --no-deps --all-features --document-private-items
         env:
-          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable --cfg tokio_taskdump
-          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable --cfg tokio_taskdump -Dwarnings
+          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable ${{ matrix.run.RUSTFLAGS }}
+          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings ${{ matrix.run.RUSTDOCFLAGS }}
 
   loom-compile:
     name: build loom tests
@@ -858,6 +843,56 @@ jobs:
         run: cargo test --features full
         working-directory: hyper
 
+  test-quinn:
+    name: Test Quinn
+    needs: basics
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os:
+          - windows-latest
+          - ubuntu-latest
+          - macos-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: ${{ env.rust_stable }}
+
+      - name: Clone Quinn
+        run: git clone https://github.com/quinn-rs/quinn.git
+
+      - name: Checkout the latest release because HEAD maybe contains breakage.
+        run: |
+          set -x
+          tag=$(git describe --abbrev=0 --tags)
+          git checkout "${tag}"
+        working-directory: quinn
+
+      - name: Patch Quinn to use tokio from this repository
+        run: |
+          set -x
+          echo '[patch.crates-io]' >>Cargo.toml
+          echo 'tokio = { path = "../tokio" }' >>Cargo.toml
+          git diff
+        working-directory: quinn
+
+      - uses: Swatinem/rust-cache@v2
+        with:
+          # The cargo workspaces and target directory configuration.
+          # These entries are separated by newlines and have the form
+          # `$workspace -> $target`. The `$target` part is treated as a directory
+          # relative to the `$workspace` and defaults to "target" if not explicitly given.
+          # default: ". -> target"
+          workspaces: "./quinn"
+
+      - name: Test Quinn
+        working-directory: quinn
+        env:
+          RUSTFLAGS: ""
+        run: cargo test
+
   x86_64-fortanix-unknown-sgx:
     name: build tokio for x86_64-fortanix-unknown-sgx
     needs: basics
@@ -896,10 +931,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_stable }}
+      - name: Install Rust 1.81
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: ${{ env.rust_stable }}
+          toolchain: 1.81
       - name: Install wasm-pack
         uses: taiki-e/install-action@wasm-pack
 
@@ -908,17 +943,22 @@ jobs:
         run: wasm-pack test --node -- --features "macros sync"
         working-directory: tokio
 
-  wasm32-wasi:
-    name: wasm32-wasi
+  wasm32-wasip1:
+    name: ${{ matrix.target }}
     needs: basics
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        target:
+          - wasm32-wasip1
+          - wasm32-wasip1-threads
     steps:
       - uses: actions/checkout@v4
       - name: Install Rust ${{ env.rust_stable }}
         uses: dtolnay/rust-toolchain@stable
         with:
           toolchain: ${{ env.rust_stable }}
-          targets: wasm32-wasi
+          targets: ${{ matrix.target }}
 
       # Install dependencies
       - name: Install cargo-hack, wasmtime, and cargo-wasi
@@ -928,28 +968,40 @@ jobs:
 
       - uses: Swatinem/rust-cache@v2
       - name: WASI test tokio full
-        run: cargo test -p tokio --target wasm32-wasi --features full
+        run: cargo test -p tokio --target ${{ matrix.target }} --features full
         env:
-          CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
-          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+          CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: WASI test tokio-util full
-        run: cargo test -p tokio-util --target wasm32-wasi --features full
+        run: cargo test -p tokio-util --target ${{ matrix.target }} --features full
         env:
-          CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
-          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+          CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: WASI test tokio-stream
-        run: cargo test -p tokio-stream --target wasm32-wasi --features time,net,io-util,sync
+        run: cargo test -p tokio-stream --target ${{ matrix.target }} --features time,net,io-util,sync
         env:
-          CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
-          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
+          CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - 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
+        if: matrix.target == 'wasm32-wasip1'
         working-directory: tests-integration
 
+      - name: test tests-integration --features wasi-threads-rt
+        run: cargo test --target ${{ matrix.target }} --features wasi-threads-rt
+        if: matrix.target == 'wasm32-wasip1-threads'
+        working-directory: tests-integration
+        env:
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
+
   check-external-types:
     name: check-external-types (${{ matrix.os }})
     needs: basics
@@ -962,7 +1014,7 @@ jobs:
         rust:
           # `check-external-types` requires a specific Rust nightly version. See
           # the README for details: https://github.com/awslabs/cargo-check-external-types
-          - nightly-2023-10-21
+          - nightly-2024-06-30
     steps:
       - uses: actions/checkout@v4
       - name: Install Rust ${{ matrix.rust }}
@@ -973,11 +1025,28 @@ jobs:
       - name: Install cargo-check-external-types
         uses: taiki-e/cache-cargo-install-action@v1
         with:
-          tool: [email protected]0
+          tool: [email protected]3
       - name: check-external-types
         run: cargo check-external-types --all-features
         working-directory: tokio
 
+  check-unexpected-lints-cfgs:
+    name: check unexpected lints and cfgs
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust ${{ env.rust_nightly }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_nightly }}
+      - name: don't allow warnings
+        run: sed -i '/#!\[allow(unknown_lints, unexpected_cfgs)\]/d' */src/lib.rs */tests/*.rs
+      - name: check for unknown lints and cfgs
+        run: cargo check --all-features --tests
+        env:
+          RUSTFLAGS: -Dwarnings --check-cfg=cfg(loom,tokio_unstable,tokio_taskdump,fuzzing,mio_unsupported_force_poll_poll,tokio_internal_mt_counters,fs,tokio_no_parking_lot,tokio_no_tuning_tests) -Funexpected_cfgs -Funknown_lints
+
   check-fuzzing:
     name: check-fuzzing
     needs: basics
@@ -1016,16 +1085,22 @@ jobs:
       - name: Make sure dictionary words are sorted and unique
         run: |
           # `sed` removes the first line (number of words) and
-          # the last line (new line). 
-          # 
+          # the last line (new line).
+          #
           # `sort` makes sure everything in between is sorted
           # and contains no duplicates.
-          # 
+          #
           # Since `sort` is sensitive to locale, we set it
           # using LC_ALL to en_US.UTF8 to be consistent in different
           # environments.
 
-          sed '1d; $d' spellcheck.dic | LC_ALL=en_US.UTF8 sort -uc
+          (
+            sed '1d; $d' spellcheck.dic | LC_ALL=en_US.UTF8 sort -uc
+          ) || {
+            echo "Dictionary is not in sorted order. Correct order is:"
+            LC_ALL=en_US.UTF8 sort -u <(sed '1d; $d' spellcheck.dic)
+            false
+          }
       - name: Run cargo-spellcheck
         run: cargo spellcheck --code 1
 

.github/workflows/pr-audit.yml

@@ -16,17 +16,8 @@ permissions:
   contents: read
 
 jobs:
-  security-audit:
+  cargo-deny:
     runs-on: ubuntu-latest
-    if: "!contains(github.event.head_commit.message, 'ci skip')"
     steps:
-      - uses: actions/checkout@v4
-
-      - name: Install cargo-audit
-        run: cargo install cargo-audit
-
-      - name: Generate lockfile
-        run: cargo generate-lockfile
-
-      - name: Audit dependencies
-        run: cargo audit
+    - uses: actions/checkout@v4
+    - uses: EmbarkStudios/cargo-deny-action@v2

.github/workflows/stress-test.yml

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

CONTRIBUTING.md

@@ -149,23 +149,31 @@ When updating this, also update:
 -->
 
 ```
-cargo +1.76 clippy --all --tests --all-features
+cargo +1.77 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:
+When building documentation, a simple `cargo doc` is not sufficient. To produce
+documentation equivalent to what will be produced in docs.rs's builds of Tokio's
+docs, please use:
 
 ```
-RUSTDOCFLAGS="--cfg docsrs" RUSTFLAGS="--cfg docsrs" cargo +nightly doc --all-features
+RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg docsrs --cfg tokio_unstable" cargo +nightly doc --all-features [--open]
 ```
 
-To build documentation including Tokio's unstable features, it is necessary to
-pass `--cfg tokio_unstable` to both RustDoc *and* rustc. To build the
-documentation for unstable features, use this command:
+This turns on indicators to display the Cargo features required for
+conditionally compiled APIs in Tokio, and it enables documentation of unstable
+Tokio features. Notice that it is necessary to pass cfg flags to both RustDoc
+*and* rustc.
+
+There is a more concise way to build docs.rs-equivalent docs by using [`cargo
+docs-rs`], which reads the above documentation flags out of Tokio's Cargo.toml
+as docs.rs itself does.
+
+[`cargo docs-rs`]: https://github.com/dtolnay/cargo-docs-rs
 
 ```
-RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg docsrs --cfg tokio_unstable" cargo +nightly doc --all-features
+cargo install --locked cargo-docs-rs
+cargo +nightly docs-rs [--open]
 ```
 
 The `cargo fmt` command does not work on the Tokio codebase. You can use the
@@ -192,8 +200,8 @@ run loom tests that test unstable features.
 
 You can run miri tests with
 ```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" \
-    cargo +nightly miri test --features full --lib
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields" \
+    cargo +nightly miri test --features full --lib --tests
 ```
 
 ### Performing spellcheck on tokio codebase
@@ -253,7 +261,7 @@ 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`
+`cargo install --locked cargo-fuzz`
 
 To list the available fuzzing harnesses you can run;
 
@@ -261,7 +269,7 @@ To list the available fuzzing harnesses you can run;
 $ cd tokio
 $ cargo fuzz list
 fuzz_linked_list
-````
+```
 
 Running a fuzz test is as simple as;
 

Cargo.toml

@@ -17,4 +17,3 @@ members = [
 
 [workspace.metadata.spellcheck]
 config = "spellcheck.toml"
-

LICENSE

@@ -1,25 +1,21 @@
-Copyright (c) 2023 Tokio Contributors
+MIT License
 
-Permission is hereby granted, free of charge, to any
-person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the
-Software without restriction, including without
-limitation the rights to use, copy, modify, merge,
-publish, distribute, sublicense, and/or sell copies of
-the Software, and to permit persons to whom the Software
-is furnished to do so, subject to the following
-conditions:
+Copyright (c) Tokio Contributors
 
-The above copyright notice and this permission notice
-shall be included in all copies or substantial portions
-of the Software.
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
 
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
-ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
-PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
-IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-DEALINGS IN THE SOFTWARE.
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

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.37.0", features = ["full"] }
+tokio = { version = "1.42.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -186,12 +186,13 @@ When updating this, also update:
 
 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.63.
+released at least six months ago. The current MSRV is 1.70.
 
 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.30 to now - Rust 1.63
+ * 1.39 to now  - Rust 1.70
+ * 1.30 to 1.38 - Rust 1.63
  * 1.27 to 1.29 - Rust 1.56
  * 1.17 to 1.26 - Rust 1.49
  * 1.15 to 1.16 - Rust 1.46
@@ -215,8 +216,8 @@ warrants a patch release with a fix for the bug, it will be backported and
 released as a new patch release for each LTS minor version. Our current LTS
 releases are:
 
- * `1.32.x` - LTS release until September 2024. (MSRV 1.63)
  * `1.36.x` - LTS release until March 2025. (MSRV 1.63)
+ * `1.38.x` - LTS release until July 2025. (MSRV 1.63)
 
 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
@@ -236,6 +237,7 @@ tokio = { version = "~1.32", features = [...] }
  * `1.18.x` - LTS release until June 2023.
  * `1.20.x` - LTS release until September 2023.
  * `1.25.x` - LTS release until March 2024.
+ * `1.32.x` - LTS release until September 2024.
 
 ## License
 

benches/Cargo.toml

@@ -3,6 +3,7 @@ name = "benches"
 version = "0.0.0"
 publish = false
 edition = "2021"
+license = "MIT"
 
 [features]
 test-util = ["tokio/test-util"]
@@ -12,11 +13,10 @@ tokio = { version = "1.5.0", path = "../tokio", features = ["full"] }
 criterion = "0.5.1"
 rand = "0.8"
 rand_chacha = "0.3"
-num_cpus = "1.16.0"
 
 [dev-dependencies]
 tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
-tokio-stream = { path = "../tokio-stream" }
+tokio-stream = { version = "0.1", path = "../tokio-stream" }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.42"
@@ -90,3 +90,8 @@ harness = false
 name = "time_now"
 path = "time_now.rs"
 harness = false
+
+[[bench]]
+name = "time_timeout"
+path = "time_timeout.rs"
+harness = false

benches/sync_mpsc.rs

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

benches/time_timeout.rs

@@ -0,0 +1,109 @@
+use std::time::{Duration, Instant};
+
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use tokio::{
+    runtime::Runtime,
+    time::{sleep, timeout},
+};
+
+// a very quick async task, but might timeout
+async fn quick_job() -> usize {
+    1
+}
+
+fn build_run_time(workers: usize) -> Runtime {
+    if workers == 1 {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+    } else {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .worker_threads(workers)
+            .build()
+            .unwrap()
+    }
+}
+
+fn single_thread_scheduler_timeout(c: &mut Criterion) {
+    do_timeout_test(c, 1, "single_thread_timeout");
+}
+
+fn multi_thread_scheduler_timeout(c: &mut Criterion) {
+    do_timeout_test(c, 8, "multi_thread_timeout-8");
+}
+
+fn do_timeout_test(c: &mut Criterion, workers: usize, name: &str) {
+    let runtime = build_run_time(workers);
+    c.bench_function(name, |b| {
+        b.iter_custom(|iters| {
+            let start = Instant::now();
+            runtime.block_on(async {
+                black_box(spawn_timeout_job(iters as usize, workers)).await;
+            });
+            start.elapsed()
+        })
+    });
+}
+
+async fn spawn_timeout_job(iters: usize, procs: usize) {
+    let mut handles = Vec::with_capacity(procs);
+    for _ in 0..procs {
+        handles.push(tokio::spawn(async move {
+            for _ in 0..iters / procs {
+                let h = timeout(Duration::from_secs(1), quick_job());
+                assert_eq!(black_box(h.await.unwrap()), 1);
+            }
+        }));
+    }
+    for handle in handles {
+        handle.await.unwrap();
+    }
+}
+
+fn single_thread_scheduler_sleep(c: &mut Criterion) {
+    do_sleep_test(c, 1, "single_thread_sleep");
+}
+
+fn multi_thread_scheduler_sleep(c: &mut Criterion) {
+    do_sleep_test(c, 8, "multi_thread_sleep-8");
+}
+
+fn do_sleep_test(c: &mut Criterion, workers: usize, name: &str) {
+    let runtime = build_run_time(workers);
+
+    c.bench_function(name, |b| {
+        b.iter_custom(|iters| {
+            let start = Instant::now();
+            runtime.block_on(async {
+                black_box(spawn_sleep_job(iters as usize, workers)).await;
+            });
+            start.elapsed()
+        })
+    });
+}
+
+async fn spawn_sleep_job(iters: usize, procs: usize) {
+    let mut handles = Vec::with_capacity(procs);
+    for _ in 0..procs {
+        handles.push(tokio::spawn(async move {
+            for _ in 0..iters / procs {
+                let _h = black_box(sleep(Duration::from_secs(1)));
+            }
+        }));
+    }
+    for handle in handles {
+        handle.await.unwrap();
+    }
+}
+
+criterion_group!(
+    timeout_benchmark,
+    single_thread_scheduler_timeout,
+    multi_thread_scheduler_timeout,
+    single_thread_scheduler_sleep,
+    multi_thread_scheduler_sleep
+);
+
+criterion_main!(timeout_benchmark);

deny.toml

@@ -0,0 +1,21 @@
+# https://embarkstudios.github.io/cargo-deny/cli/init.html
+
+[graph]
+all-features = true
+
+[licenses]
+allow = [
+    "MIT",
+    "Apache-2.0",
+]
+exceptions = [
+    { allow = ["Unicode-3.0", "Unicode-DFS-2016"], crate = "unicode-ident" },
+]
+
+[bans]
+multiple-versions = "allow"
+wildcards = "deny"
+
+[sources]
+unknown-registry = "deny"
+unknown-git = "deny"

examples/Cargo.toml

@@ -3,6 +3,7 @@ name = "examples"
 version = "0.0.0"
 publish = false
 edition = "2021"
+license = "MIT"
 
 # If you copy one of the examples into a new project, you should be using
 # [dependencies] instead, and delete the **path**.
@@ -25,7 +26,7 @@ once_cell = "1.5.2"
 rand = "0.8.3"
 
 [target.'cfg(windows)'.dev-dependencies.windows-sys]
-version = "0.48"
+version = "0.52"
 
 [[example]]
 name = "chat"

examples/chat.rs

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

examples/connect.rs

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

examples/dump.rs

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

examples/echo-udp.rs

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

examples/echo.rs

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

examples/print_each_packet.rs

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

examples/proxy.rs

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

examples/tinydb.rs

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

examples/tinyhttp.rs

@@ -31,13 +31,13 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .nth(1)
         .unwrap_or_else(|| "127.0.0.1:8080".to_string());
     let server = TcpListener::bind(&addr).await?;
-    println!("Listening on: {}", addr);
+    println!("Listening on: {addr}");
 
     loop {
         let (stream, _) = server.accept().await?;
         tokio::spawn(async move {
             if let Err(e) = process(stream).await {
-                println!("failed to process connection; error = {}", e);
+                println!("failed to process connection; error = {e}");
             }
         });
     }
@@ -159,7 +159,7 @@ impl Decoder for Http {
             let mut parsed_headers = [httparse::EMPTY_HEADER; 16];
             let mut r = httparse::Request::new(&mut parsed_headers);
             let status = r.parse(src).map_err(|e| {
-                let msg = format!("failed to parse http request: {:?}", e);
+                let msg = format!("failed to parse http request: {e:?}");
                 io::Error::new(io::ErrorKind::Other, msg)
             })?;
 
@@ -259,11 +259,11 @@ mod date {
         unix_date: u64,
     }
 
-    thread_local!(static LAST: RefCell<LastRenderedNow> = RefCell::new(LastRenderedNow {
+    thread_local!(static LAST: RefCell<LastRenderedNow> = const { RefCell::new(LastRenderedNow {
         bytes: [0; 128],
         amt: 0,
         unix_date: 0,
-    }));
+    }) });
 
     impl fmt::Display for Now {
         fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

examples/udp-codec.rs

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

spellcheck.dic

@@ -1,4 +1,4 @@
-283
+292
 &
 +
 <
@@ -12,12 +12,17 @@
 0xA
 0xD
 100ms
+100ns
 10ms
+10μs
 ~12
+120s
+12.5%
 ±1m
 ±1ms
 1ms
 1s
+25%
 250ms
 2x
 ~4
@@ -34,6 +39,7 @@ amongst
 api
 APIs
 async
+atomics
 awaitable
 backend
 backpressure
@@ -98,6 +104,7 @@ errored
 EWMA
 expirations
 fcntl
+fd
 fd's
 FIFOs
 filename
@@ -114,9 +121,11 @@ GID
 goroutines
 Growable
 gzip
+H2
 hashmaps
 HashMaps
 hashsets
+HdrHistogram
 ie
 Illumos
 impl
@@ -131,6 +140,7 @@ io
 IOCP
 iOS
 IOs
+io_uring
 IP
 IPv4
 IPv6
@@ -149,6 +159,7 @@ metadata
 mio
 Mio
 mio's
+miri
 misconfigured
 mock's
 mpmc

stress-test/Cargo.toml

@@ -3,12 +3,13 @@ name = "stress-test"
 version = "0.1.0"
 authors = ["Tokio Contributors <[email protected]>"]
 edition = "2021"
+license = "MIT"
 publish = false
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-tokio = { path = "../tokio/", features = ["full"] }
+tokio = { version = "1.0.0", path = "../tokio/", features = ["full"] }
 
 [dev-dependencies]
 rand = "0.8"

target-specs/README.md

@@ -0,0 +1,9 @@
+This is used for the `no-atomic-u64-test` ci check that verifies that Tokio
+works even if the `AtomicU64` type is missing.
+
+When increasing the nightly compiler version, you may need to regenerate this
+target using the following command:
+```
+rustc +nightly -Z unstable-options --print target-spec-json --target i686-unknown-linux-gnu | grep -v 'is-builtin' | sed 's/"max-atomic-width": 64/"max-atomic-width": 32/' > target-specs/i686-unknown-linux-gnu.json
+```
+

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

@@ -1,29 +1,35 @@
 {
   "arch": "x86",
   "cpu": "pentium4",
+  "crt-objects-fallback": "false",
   "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",
+  "data-layout": "e-m:e-p:32:32-p270:32:32-p271:32:32-p272:64:64-i128:128-f64:32:64-f80:32-n8:16:32-S128",
   "dynamic-linking": true,
   "env": "gnu",
   "has-rpath": true,
   "has-thread-local": true,
+  "linker-flavor": "gnu-cc",
   "llvm-target": "i686-unknown-linux-gnu",
   "max-atomic-width": 32,
+  "metadata": {
+    "description": null,
+    "host_tools": null,
+    "std": null,
+    "tier": null
+  },
   "os": "linux",
   "position-independent-executables": true,
   "pre-link-args": {
-    "gcc": [
+    "gnu-cc": [
+      "-m32"
+    ],
+    "gnu-lld-cc": [
       "-m32"
     ]
   },
   "relro-level": "full",
   "stack-probes": {
-    "kind": "inline-or-call",
-    "min-llvm-version-for-inline": [
-      16,
-      0,
-      0
-    ]
+    "kind": "inline"
   },
   "supported-sanitizers": [
     "address"

tests-build/Cargo.toml

@@ -3,6 +3,7 @@ name = "tests-build"
 version = "0.1.0"
 authors = ["Tokio Contributors <[email protected]>"]
 edition = "2021"
+license = "MIT"
 publish = false
 
 [features]
@@ -10,7 +11,7 @@ full = ["tokio/full"]
 rt = ["tokio/rt", "tokio/macros"]
 
 [dependencies]
-tokio = { path = "../tokio", optional = true }
+tokio = { version = "1.0.0", path = "../tokio", optional = true }
 
 [dev-dependencies]
 trybuild = "1.0"

tests-build/README.md

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

tests-build/tests/fail/macros_invalid_input.rs

@@ -41,8 +41,31 @@ async fn test_crate_not_path_int() {}
 #[tokio::test(crate = "456")]
 async fn test_crate_not_path_invalid() {}
 
+#[tokio::test(flavor = "multi_thread", unhandled_panic = "shutdown_runtime")]
+async fn test_multi_thread_with_unhandled_panic() {}
+
 #[tokio::test]
 #[test]
 async fn test_has_second_test_attr() {}
 
+#[tokio::test]
+#[::core::prelude::v1::test]
+async fn test_has_second_test_attr_v1() {}
+
+#[tokio::test]
+#[core::prelude::rust_2015::test]
+async fn test_has_second_test_attr_rust_2015() {}
+
+#[tokio::test]
+#[::std::prelude::rust_2018::test]
+async fn test_has_second_test_attr_rust_2018() {}
+
+#[tokio::test]
+#[std::prelude::rust_2021::test]
+async fn test_has_second_test_attr_rust_2021() {}
+
+#[tokio::test]
+#[tokio::test]
+async fn test_has_generated_second_test_attr() {}
+
 fn main() {}

tests-build/tests/fail/macros_invalid_input.stderr

@@ -1,95 +1,121 @@
 error: the `async` keyword is missing from the function declaration
- --> $DIR/macros_invalid_input.rs:6:1
+ --> tests/fail/macros_invalid_input.rs:6:1
   |
 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:8:15
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.
+ --> tests/fail/macros_invalid_input.rs:8:15
   |
 8 | #[tokio::main(foo)]
   |               ^^^
 
 error: Must have specified ident
-  --> $DIR/macros_invalid_input.rs:11:15
+  --> tests/fail/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:15:1
+  --> tests/fail/macros_invalid_input.rs:15:1
    |
 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:17:15
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.
+  --> tests/fail/macros_invalid_input.rs:17:15
    |
 17 | #[tokio::test(foo)]
    |               ^^^
 
-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
-  --> $DIR/macros_invalid_input.rs:20:15
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`
+  --> tests/fail/macros_invalid_input.rs:20:15
    |
 20 | #[tokio::test(foo = 123)]
    |               ^^^^^^^^^
 
 error: Failed to parse value of `flavor` as string.
-  --> $DIR/macros_invalid_input.rs:23:24
+  --> tests/fail/macros_invalid_input.rs:23:24
    |
 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:26:24
+  --> tests/fail/macros_invalid_input.rs:26:24
    |
 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:29:55
+  --> tests/fail/macros_invalid_input.rs:29:55
    |
 29 | #[tokio::test(flavor = "multi_thread", start_paused = false)]
    |                                                       ^^^^^
 
 error: Failed to parse value of `worker_threads` as integer.
-  --> $DIR/macros_invalid_input.rs:32:57
+  --> tests/fail/macros_invalid_input.rs:32:57
    |
 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:35:59
+  --> tests/fail/macros_invalid_input.rs:35:59
    |
 35 | #[tokio::test(flavor = "current_thread", worker_threads = 4)]
    |                                                           ^
 
 error: Failed to parse value of `crate` as path.
-  --> $DIR/macros_invalid_input.rs:38:23
+  --> tests/fail/macros_invalid_input.rs:38:23
    |
 38 | #[tokio::test(crate = 456)]
    |                       ^^^
 
 error: Failed to parse value of `crate` as path: "456"
-  --> $DIR/macros_invalid_input.rs:41:23
+  --> tests/fail/macros_invalid_input.rs:41:23
    |
 41 | #[tokio::test(crate = "456")]
    |                       ^^^^^
 
-error: second test attribute is supplied
-  --> $DIR/macros_invalid_input.rs:45:1
+error: The `unhandled_panic` option requires the `current_thread` runtime flavor. Use `#[tokio::test(flavor = "current_thread")]`
+  --> tests/fail/macros_invalid_input.rs:44:58
    |
-45 | #[test]
+44 | #[tokio::test(flavor = "multi_thread", unhandled_panic = "shutdown_runtime")]
+   |                                                          ^^^^^^^^^^^^^^^^^^
+
+error: second test attribute is supplied, consider removing or changing the order of your test attributes
+  --> tests/fail/macros_invalid_input.rs:48:1
+   |
+48 | #[test]
    | ^^^^^^^
 
-error: duplicated attribute
-  --> $DIR/macros_invalid_input.rs:45:1
+error: second test attribute is supplied, consider removing or changing the order of your test attributes
+  --> tests/fail/macros_invalid_input.rs:52:1
    |
-45 | #[test]
-   | ^^^^^^^
+52 | #[::core::prelude::v1::test]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error: second test attribute is supplied, consider removing or changing the order of your test attributes
+  --> tests/fail/macros_invalid_input.rs:56:1
    |
-note: the lint level is defined here
-  --> $DIR/macros_invalid_input.rs:1:9
+56 | #[core::prelude::rust_2015::test]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error: second test attribute is supplied, consider removing or changing the order of your test attributes
+  --> tests/fail/macros_invalid_input.rs:60:1
    |
-1  | #![deny(duplicate_macro_attributes)]
-   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^
+60 | #[::std::prelude::rust_2018::test]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error: second test attribute is supplied, consider removing or changing the order of your test attributes
+  --> tests/fail/macros_invalid_input.rs:64:1
+   |
+64 | #[std::prelude::rust_2021::test]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error: second test attribute is supplied, consider removing or changing the order of your test attributes
+  --> tests/fail/macros_invalid_input.rs:67:1
+   |
+67 | #[tokio::test]
+   | ^^^^^^^^^^^^^^
+   |
+   = note: this error originates in the attribute macro `tokio::test` (in Nightly builds, run with -Z macro-backtrace for more info)

tests-build/tests/macros.rs

@@ -1,4 +1,5 @@
 #[test]
+#[cfg_attr(miri, ignore)]
 fn compile_fail_full() {
     let t = trybuild::TestCases::new();
 

tests-integration/Cargo.toml

@@ -3,6 +3,7 @@ name = "tests-integration"
 version = "0.1.0"
 authors = ["Tokio Contributors <[email protected]>"]
 edition = "2021"
+license = "MIT"
 publish = false
 
 [[bin]]
@@ -39,6 +40,7 @@ rt-process-signal = ["rt-net", "tokio/process", "tokio/signal"]
 # This is an explicit feature so we can use `cargo hack` testing single features
 # instead of all possible permutations.
 wasi-rt = ["rt", "macros", "sync"]
+wasi-threads-rt = ["wasi-rt", "rt-multi-thread"]
 
 full = [
     "macros",
@@ -54,8 +56,8 @@ rt = ["tokio/rt"]
 rt-multi-thread = ["rt", "tokio/rt-multi-thread"]
 
 [dependencies]
-tokio = { path = "../tokio" }
-tokio-test = { path = "../tokio-test", optional = true }
+tokio = { version = "1.0.0", path = "../tokio" }
+tokio-test = { version = "0.4", path = "../tokio-test", optional = true }
 doc-comment = "0.3.1"
 futures = { version = "0.3.0", features = ["async-await"] }
 bytes = "1.0.0"

tests-integration/src/bin/test-mem.rs

@@ -1,4 +1,4 @@
-use futures::future::poll_fn;
+use std::future::poll_fn;
 
 fn main() {
     let rt = tokio::runtime::Builder::new_multi_thread()

tests-integration/tests/macros_main.rs

@@ -1,8 +1,4 @@
-#![cfg(all(
-    feature = "macros",
-    feature = "rt-multi-thread",
-    not(target_os = "wasi")
-))]
+#![cfg(all(feature = "macros", feature = "rt-multi-thread"))]
 
 #[tokio::main]
 async fn basic_main() -> usize {

tests-integration/tests/macros_select.rs

@@ -4,7 +4,10 @@ use futures::channel::oneshot;
 use futures::executor::block_on;
 use std::thread;
 
-#[cfg_attr(target_os = "wasi", ignore = "WASI: std::thread::spawn not supported")]
+#[cfg_attr(
+    not(feature = "rt-multi-thread"),
+    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(all(feature = "full", not(target_os = "wasi")))]
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
 
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::join;
@@ -10,6 +10,7 @@ use futures::future::{self, FutureExt};
 use std::env;
 use std::io;
 use std::process::{ExitStatus, Stdio};
+use std::task::ready;
 
 fn cat() -> Command {
     let mut cmd = Command::new(env!("CARGO_BIN_EXE_test-cat"));
@@ -205,13 +206,13 @@ async fn vectored_writes() {
         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 {
+        std::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))?;
+            let n = ready!(Pin::new(&mut stdin).poll_write_vectored(cx, &slices))?;
             writes_completed += 1;
             input.advance(n);
         })

tokio-macros/CHANGELOG.md

@@ -1,3 +1,17 @@
+# 2.4.0 (July 22nd, 2024)
+
+- msrv: increase MSRV to 1.70 ([#6645])
+- macros: allow `unhandled_panic` behavior for `#[tokio::main]` and `#[tokio::test]` ([#6593])
+
+[#6593]: https://github.com/tokio-rs/tokio/pull/6593
+[#6645]: https://github.com/tokio-rs/tokio/pull/6645
+
+# 2.3.0 (May 30th, 2024)
+
+- macros: make `#[tokio::test]` append `#[test]` at the end of the attribute list ([#6497])
+
+[#6497]: https://github.com/tokio-rs/tokio/pull/6497
+
 # 2.2.0 (November 19th, 2023)
 
 ### Changed

tokio-macros/Cargo.toml

@@ -4,9 +4,9 @@ name = "tokio-macros"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-macros-1.x.y" git tag.
-version = "2.2.0"
+version = "2.4.0"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"

tokio-macros/LICENSE

@@ -1,32 +1,7 @@
-Copyright (c) 2023 Tokio Contributors
-
-Permission is hereby granted, free of charge, to any
-person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the
-Software without restriction, including without
-limitation the rights to use, copy, modify, merge,
-publish, distribute, sublicense, and/or sell copies of
-the Software, and to permit persons to whom the Software
-is furnished to do so, subject to the following
-conditions:
-
-The above copyright notice and this permission notice
-shall be included in all copies or substantial portions
-of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
-ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
-PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
-IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-DEALINGS IN THE SOFTWARE.
-
-The MIT License (MIT)
+MIT License
 
 Copyright (c) 2019 Yoshua Wuyts
+Copyright (c) Tokio Contributors
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

tokio-macros/src/entry.rs

@@ -20,7 +20,32 @@ impl RuntimeFlavor {
             "single_thread" => Err("The single threaded runtime flavor is called `current_thread`.".to_string()),
             "basic_scheduler" => Err("The `basic_scheduler` runtime flavor has been renamed to `current_thread`.".to_string()),
             "threaded_scheduler" => Err("The `threaded_scheduler` runtime flavor has been renamed to `multi_thread`.".to_string()),
-            _ => Err(format!("No such runtime flavor `{}`. The runtime flavors are `current_thread` and `multi_thread`.", s)),
+            _ => Err(format!("No such runtime flavor `{s}`. The runtime flavors are `current_thread` and `multi_thread`.")),
+        }
+    }
+}
+
+#[derive(Clone, Copy, PartialEq)]
+enum UnhandledPanic {
+    Ignore,
+    ShutdownRuntime,
+}
+
+impl UnhandledPanic {
+    fn from_str(s: &str) -> Result<UnhandledPanic, String> {
+        match s {
+            "ignore" => Ok(UnhandledPanic::Ignore),
+            "shutdown_runtime" => Ok(UnhandledPanic::ShutdownRuntime),
+            _ => Err(format!("No such unhandled panic behavior `{s}`. The unhandled panic behaviors are `ignore` and `shutdown_runtime`.")),
+        }
+    }
+
+    fn into_tokens(self, crate_path: &TokenStream) -> TokenStream {
+        match self {
+            UnhandledPanic::Ignore => quote! { #crate_path::runtime::UnhandledPanic::Ignore },
+            UnhandledPanic::ShutdownRuntime => {
+                quote! { #crate_path::runtime::UnhandledPanic::ShutdownRuntime }
+            }
         }
     }
 }
@@ -30,6 +55,7 @@ struct FinalConfig {
     worker_threads: Option<usize>,
     start_paused: Option<bool>,
     crate_name: Option<Path>,
+    unhandled_panic: Option<UnhandledPanic>,
 }
 
 /// Config used in case of the attribute not being able to build a valid config
@@ -38,6 +64,7 @@ const DEFAULT_ERROR_CONFIG: FinalConfig = FinalConfig {
     worker_threads: None,
     start_paused: None,
     crate_name: None,
+    unhandled_panic: None,
 };
 
 struct Configuration {
@@ -48,6 +75,7 @@ struct Configuration {
     start_paused: Option<(bool, Span)>,
     is_test: bool,
     crate_name: Option<Path>,
+    unhandled_panic: Option<(UnhandledPanic, Span)>,
 }
 
 impl Configuration {
@@ -63,6 +91,7 @@ impl Configuration {
             start_paused: None,
             is_test,
             crate_name: None,
+            unhandled_panic: None,
         }
     }
 
@@ -117,6 +146,25 @@ impl Configuration {
         Ok(())
     }
 
+    fn set_unhandled_panic(
+        &mut self,
+        unhandled_panic: syn::Lit,
+        span: Span,
+    ) -> Result<(), syn::Error> {
+        if self.unhandled_panic.is_some() {
+            return Err(syn::Error::new(
+                span,
+                "`unhandled_panic` set multiple times.",
+            ));
+        }
+
+        let unhandled_panic = parse_string(unhandled_panic, span, "unhandled_panic")?;
+        let unhandled_panic =
+            UnhandledPanic::from_str(&unhandled_panic).map_err(|err| syn::Error::new(span, err))?;
+        self.unhandled_panic = Some((unhandled_panic, span));
+        Ok(())
+    }
+
     fn macro_name(&self) -> &'static str {
         if self.is_test {
             "tokio::test"
@@ -163,11 +211,24 @@ impl Configuration {
             (_, None) => None,
         };
 
+        let unhandled_panic = match (flavor, self.unhandled_panic) {
+            (F::Threaded, Some((_, unhandled_panic_span))) => {
+                let msg = format!(
+                    "The `unhandled_panic` option requires the `current_thread` runtime flavor. Use `#[{}(flavor = \"current_thread\")]`",
+                    self.macro_name(),
+                );
+                return Err(syn::Error::new(unhandled_panic_span, msg));
+            }
+            (F::CurrentThread, Some((unhandled_panic, _))) => Some(unhandled_panic),
+            (_, None) => None,
+        };
+
         Ok(FinalConfig {
             crate_name: self.crate_name.clone(),
             flavor,
             worker_threads,
             start_paused,
+            unhandled_panic,
         })
     }
 }
@@ -178,12 +239,12 @@ fn parse_int(int: syn::Lit, span: Span, field: &str) -> Result<usize, syn::Error
             Ok(value) => Ok(value),
             Err(e) => Err(syn::Error::new(
                 span,
-                format!("Failed to parse value of `{}` as integer: {}", field, e),
+                format!("Failed to parse value of `{field}` as integer: {e}"),
             )),
         },
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as integer.", field),
+            format!("Failed to parse value of `{field}` as integer."),
         )),
     }
 }
@@ -194,7 +255,7 @@ fn parse_string(int: syn::Lit, span: Span, field: &str) -> Result<String, syn::E
         syn::Lit::Verbatim(s) => Ok(s.to_string()),
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as string.", field),
+            format!("Failed to parse value of `{field}` as string."),
         )),
     }
 }
@@ -214,7 +275,7 @@ fn parse_path(lit: syn::Lit, span: Span, field: &str) -> Result<Path, syn::Error
         }
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as path.", field),
+            format!("Failed to parse value of `{field}` as path."),
         )),
     }
 }
@@ -224,7 +285,7 @@ fn parse_bool(bool: syn::Lit, span: Span, field: &str) -> Result<bool, syn::Erro
         syn::Lit::Bool(b) => Ok(b.value),
         _ => Err(syn::Error::new(
             span,
-            format!("Failed to parse value of `{}` as bool.", field),
+            format!("Failed to parse value of `{field}` as bool."),
         )),
     }
 }
@@ -275,10 +336,13 @@ fn build_config(
                     "crate" => {
                         config.set_crate_name(lit.clone(), syn::spanned::Spanned::span(lit))?;
                     }
+                    "unhandled_panic" => {
+                        config
+                            .set_unhandled_panic(lit.clone(), syn::spanned::Spanned::span(lit))?;
+                    }
                     name => {
                         let msg = format!(
-                            "Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`",
-                            name,
+                            "Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`",
                         );
                         return Err(syn::Error::new_spanned(namevalue, msg));
                     }
@@ -293,21 +357,19 @@ fn build_config(
                 let msg = match name.as_str() {
                     "threaded_scheduler" | "multi_thread" => {
                         format!(
-                            "Set the runtime flavor with #[{}(flavor = \"multi_thread\")].",
-                            macro_name
+                            "Set the runtime flavor with #[{macro_name}(flavor = \"multi_thread\")]."
                         )
                     }
                     "basic_scheduler" | "current_thread" | "single_threaded" => {
                         format!(
-                            "Set the runtime flavor with #[{}(flavor = \"current_thread\")].",
-                            macro_name
+                            "Set the runtime flavor with #[{macro_name}(flavor = \"current_thread\")]."
                         )
                     }
-                    "flavor" | "worker_threads" | "start_paused" => {
-                        format!("The `{}` attribute requires an argument.", name)
+                    "flavor" | "worker_threads" | "start_paused" | "crate" | "unhandled_panic" => {
+                        format!("The `{name}` attribute requires an argument.")
                     }
                     name => {
-                        format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`", name)
+                        format!("Unknown attribute {name} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`, `unhandled_panic`.")
                     }
                 };
                 return Err(syn::Error::new_spanned(path, msg));
@@ -359,8 +421,12 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
     if let Some(v) = config.start_paused {
         rt = quote_spanned! {last_stmt_start_span=> #rt.start_paused(#v) };
     }
+    if let Some(v) = config.unhandled_panic {
+        let unhandled_panic = v.into_tokens(&crate_path);
+        rt = quote_spanned! {last_stmt_start_span=> #rt.unhandled_panic(#unhandled_panic) };
+    }
 
-    let header = if is_test {
+    let generated_attrs = if is_test {
         quote! {
             #[::core::prelude::v1::test]
         }
@@ -369,8 +435,9 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
     };
 
     let body_ident = quote! { body };
+    // This explicit `return` is intentional. See tokio-rs/tokio#4636
     let last_block = quote_spanned! {last_stmt_end_span=>
-        #[allow(clippy::expect_used, clippy::diverging_sub_expression)]
+        #[allow(clippy::expect_used, clippy::diverging_sub_expression, clippy::needless_return)]
         {
             return #rt
                 .enable_all()
@@ -410,7 +477,7 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
         }
     };
 
-    input.into_tokens(header, body, last_block)
+    input.into_tokens(generated_attrs, body, last_block)
 }
 
 fn token_stream_with_error(mut tokens: TokenStream, error: syn::Error) -> TokenStream {
@@ -418,7 +485,6 @@ fn token_stream_with_error(mut tokens: TokenStream, error: syn::Error) -> TokenS
     tokens
 }
 
-#[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub(crate) fn main(args: TokenStream, item: TokenStream, rt_multi_thread: bool) -> TokenStream {
     // 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
@@ -443,6 +509,35 @@ pub(crate) fn main(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
     }
 }
 
+// Check whether given attribute is a test attribute of forms:
+// * `#[test]`
+// * `#[core::prelude::*::test]` or `#[::core::prelude::*::test]`
+// * `#[std::prelude::*::test]` or `#[::std::prelude::*::test]`
+fn is_test_attribute(attr: &Attribute) -> bool {
+    let path = match &attr.meta {
+        syn::Meta::Path(path) => path,
+        _ => return false,
+    };
+    let candidates = [
+        ["core", "prelude", "*", "test"],
+        ["std", "prelude", "*", "test"],
+    ];
+    if path.leading_colon.is_none()
+        && path.segments.len() == 1
+        && path.segments[0].arguments.is_none()
+        && path.segments[0].ident == "test"
+    {
+        return true;
+    } else if path.segments.len() != candidates[0].len() {
+        return false;
+    }
+    candidates.into_iter().any(|segments| {
+        path.segments.iter().zip(segments).all(|(segment, path)| {
+            segment.arguments.is_none() && (path == "*" || segment.ident == path)
+        })
+    })
+}
+
 pub(crate) fn test(args: TokenStream, item: TokenStream, rt_multi_thread: bool) -> TokenStream {
     // 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
@@ -451,8 +546,8 @@ pub(crate) fn test(args: TokenStream, item: TokenStream, rt_multi_thread: bool)
         Ok(it) => it,
         Err(e) => return token_stream_with_error(item, e),
     };
-    let config = if let Some(attr) = input.attrs().find(|attr| attr.meta.path().is_ident("test")) {
-        let msg = "second test attribute is supplied";
+    let config = if let Some(attr) = input.attrs().find(|attr| is_test_attribute(attr)) {
+        let msg = "second test attribute is supplied, consider removing or changing the order of your test attributes";
         Err(syn::Error::new_spanned(attr, msg))
     } else {
         AttributeArgs::parse_terminated
@@ -493,13 +588,11 @@ impl ItemFn {
     /// Convert our local function item into a token stream.
     fn into_tokens(
         self,
-        header: proc_macro2::TokenStream,
+        generated_attrs: 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);
@@ -513,6 +606,9 @@ impl ItemFn {
             attr.to_tokens(&mut tokens);
         }
 
+        // Add generated macros at the end, so macros processed later are aware of them.
+        generated_attrs.to_tokens(&mut tokens);
+
         self.vis.to_tokens(&mut tokens);
         self.sig.to_tokens(&mut tokens);
 

tokio-macros/src/lib.rs

@@ -1,3 +1,4 @@
+#![allow(unknown_lints, unexpected_cfgs)]
 #![allow(clippy::needless_doctest_main)]
 #![warn(
     missing_debug_implementations,
@@ -201,8 +202,55 @@ use proc_macro::TokenStream;
 ///         })
 /// }
 /// ```
+///
+/// ### Configure unhandled panic behavior
+///
+/// Available options are `shutdown_runtime` and `ignore`. For more details, see
+/// [`Builder::unhandled_panic`].
+///
+/// This option is only compatible with the `current_thread` runtime.
+///
+/// ```no_run
+/// # #![allow(unknown_lints, unexpected_cfgs)]
+/// #[cfg(tokio_unstable)]
+/// #[tokio::main(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
+/// async fn main() {
+///     let _ = tokio::spawn(async {
+///         panic!("This panic will shutdown the runtime.");
+///     }).await;
+/// }
+/// # #[cfg(not(tokio_unstable))]
+/// # fn main() { }
+/// ```
+///
+/// Equivalent code not using `#[tokio::main]`
+///
+/// ```no_run
+/// # #![allow(unknown_lints, unexpected_cfgs)]
+/// #[cfg(tokio_unstable)]
+/// fn main() {
+///     tokio::runtime::Builder::new_current_thread()
+///         .enable_all()
+///         .unhandled_panic(UnhandledPanic::ShutdownRuntime)
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             let _ = tokio::spawn(async {
+///                 panic!("This panic will shutdown the runtime.");
+///             }).await;
+///         })
+/// }
+/// # #[cfg(not(tokio_unstable))]
+/// # fn main() { }
+/// ```
+///
+/// **Note**: This option depends on Tokio's [unstable API][unstable]. See [the
+/// documentation on unstable features][unstable] for details on how to enable
+/// Tokio's unstable features.
+///
+/// [`Builder::unhandled_panic`]: ../tokio/runtime/struct.Builder.html#method.unhandled_panic
+/// [unstable]: ../tokio/index.html#unstable-features
 #[proc_macro_attribute]
-#[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
     entry::main(args.into(), item.into(), true).into()
 }
@@ -267,7 +315,6 @@ 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.into(), item.into(), false).into()
 }
@@ -365,7 +412,7 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// ### Set number of worker threads
 ///
 /// ```no_run
-/// #[tokio::test(flavor ="multi_thread", worker_threads = 2)]
+/// #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 /// async fn my_test() {
 ///     assert!(true);
 /// }
@@ -424,6 +471,55 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 ///     println!("Hello world");
 /// }
 /// ```
+///
+/// ### Configure unhandled panic behavior
+///
+/// Available options are `shutdown_runtime` and `ignore`. For more details, see
+/// [`Builder::unhandled_panic`].
+///
+/// This option is only compatible with the `current_thread` runtime.
+///
+/// ```no_run
+/// # #![allow(unknown_lints, unexpected_cfgs)]
+/// #[cfg(tokio_unstable)]
+/// #[tokio::test(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
+/// async fn my_test() {
+///     let _ = tokio::spawn(async {
+///         panic!("This panic will shutdown the runtime.");
+///     }).await;
+/// }
+/// # #[cfg(not(tokio_unstable))]
+/// # fn main() { }
+/// ```
+///
+/// Equivalent code not using `#[tokio::test]`
+///
+/// ```no_run
+/// # #![allow(unknown_lints, unexpected_cfgs)]
+/// #[cfg(tokio_unstable)]
+/// #[test]
+/// fn my_test() {
+///     tokio::runtime::Builder::new_current_thread()
+///         .enable_all()
+///         .unhandled_panic(UnhandledPanic::ShutdownRuntime)
+///         .build()
+///         .unwrap()
+///         .block_on(async {
+///             let _ = tokio::spawn(async {
+///                 panic!("This panic will shutdown the runtime.");
+///             }).await;
+///         })
+/// }
+/// # #[cfg(not(tokio_unstable))]
+/// # fn main() { }
+/// ```
+///
+/// **Note**: This option depends on Tokio's [unstable API][unstable]. See [the
+/// documentation on unstable features][unstable] for details on how to enable
+/// Tokio's unstable features.
+///
+/// [`Builder::unhandled_panic`]: ../tokio/runtime/struct.Builder.html#method.unhandled_panic
+/// [unstable]: ../tokio/index.html#unstable-features
 #[proc_macro_attribute]
 pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
     entry::test(args.into(), item.into(), true).into()

tokio-macros/src/select.rs

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

tokio-stream/CHANGELOG.md

@@ -1,3 +1,15 @@
+# 0.1.16 (September 5th, 2024)
+
+This release bumps the MSRV of tokio-stream to 1.70.
+
+- stream: add `next_many` and `poll_next_many` to `StreamMap` ([#6409])
+- stream: make stream adapters public ([#6658])
+- readme: add readme for tokio-stream ([#6456])
+
+[#6409]: https://github.com/tokio-rs/tokio/pull/6409
+[#6658]: https://github.com/tokio-rs/tokio/pull/6658
+[#6456]: https://github.com/tokio-rs/tokio/pull/6456
+
 # 0.1.15 (March 14th, 2024)
 
 This release bumps the MSRV of tokio-stream to 1.63.

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.15"
+version = "0.1.16"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -45,7 +45,7 @@ tokio-util = { version = "0.7.0", path = "../tokio-util", optional = true }
 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" }
+tokio-test = { version = "0.4", path = "../tokio-test" }
 futures = { version = "0.3", default-features = false }
 
 [package.metadata.docs.rs]

tokio-stream/LICENSE

@@ -1,25 +1,21 @@
-Copyright (c) 2023 Tokio Contributors
+MIT License
 
-Permission is hereby granted, free of charge, to any
-person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the
-Software without restriction, including without
-limitation the rights to use, copy, modify, merge,
-publish, distribute, sublicense, and/or sell copies of
-the Software, and to permit persons to whom the Software
-is furnished to do so, subject to the following
-conditions:
+Copyright (c) Tokio Contributors
 
-The above copyright notice and this permission notice
-shall be included in all copies or substantial portions
-of the Software.
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
 
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
-ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
-PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
-IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-DEALINGS IN THE SOFTWARE.
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

tokio-stream/README.md

@@ -0,0 +1,13 @@
+# tokio-stream
+
+Utilities to work with `Stream` and `tokio`.
+
+## License
+
+This project is licensed under the [MIT license](LICENSE).
+
+### Contribution
+
+Unless you explicitly state otherwise, any contribution intentionally submitted
+for inclusion in Tokio by you, shall be licensed as MIT, without any additional
+terms or conditions.

tokio-stream/src/lib.rs

@@ -1,3 +1,4 @@
+#![allow(unknown_lints, unexpected_cfgs)]
 #![allow(
     clippy::cognitive_complexity,
     clippy::large_enum_variant,
@@ -73,15 +74,26 @@
 #[macro_use]
 mod macros;
 
-mod poll_fn;
-pub(crate) use poll_fn::poll_fn;
-
 pub mod wrappers;
 
 mod stream_ext;
 pub use stream_ext::{collect::FromStream, StreamExt};
+/// Adapters for [`Stream`]s created by methods in [`StreamExt`].
+pub mod adapters {
+    pub use crate::stream_ext::{
+        Chain, Filter, FilterMap, Fuse, Map, MapWhile, Merge, Peekable, Skip, SkipWhile, Take,
+        TakeWhile, Then,
+    };
+    cfg_time! {
+        pub use crate::stream_ext::{ChunksTimeout, Timeout, TimeoutRepeating};
+    }
+}
+
 cfg_time! {
-    pub use stream_ext::timeout::{Elapsed, Timeout};
+    #[deprecated = "Import those symbols from adapters instead"]
+    #[doc(hidden)]
+    pub use stream_ext::timeout::Timeout;
+    pub use stream_ext::timeout::Elapsed;
 }
 
 mod empty;

tokio-stream/src/macros.rs

@@ -57,12 +57,3 @@ macro_rules! cfg_signal {
         )*
     }
 }
-
-macro_rules! ready {
-    ($e:expr $(,)?) => {
-        match $e {
-            std::task::Poll::Ready(t) => t,
-            std::task::Poll::Pending => return std::task::Poll::Pending,
-        }
-    };
-}

tokio-stream/src/poll_fn.rs

@@ -1,35 +0,0 @@
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-pub(crate) struct PollFn<F> {
-    f: F,
-}
-
-pub(crate) fn poll_fn<T, F>(f: F) -> PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    PollFn { f }
-}
-
-impl<T, F> Future for PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    type Output = T;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
-        // Safety: We never construct a `Pin<&mut F>` anywhere, so accessing `f`
-        // mutably in an unpinned way is sound.
-        //
-        // This use of unsafe cannot be replaced with the pin-project macro
-        // because:
-        //  * If we put `#[pin]` on the field, then it gives us a `Pin<&mut F>`,
-        //    which we can't use to call the closure.
-        //  * If we don't put `#[pin]` on the field, then it makes `PollFn` be
-        //    unconditionally `Unpin`, which we also don't want.
-        let me = unsafe { Pin::into_inner_unchecked(self) };
-        (me.f)(cx)
-    }
-}

tokio-stream/src/stream_ext.rs

@@ -8,66 +8,66 @@ mod any;
 use any::AnyFuture;
 
 mod chain;
-use chain::Chain;
+pub use chain::Chain;
 
 pub(crate) mod collect;
 use collect::{Collect, FromStream};
 
 mod filter;
-use filter::Filter;
+pub use filter::Filter;
 
 mod filter_map;
-use filter_map::FilterMap;
+pub use filter_map::FilterMap;
 
 mod fold;
 use fold::FoldFuture;
 
 mod fuse;
-use fuse::Fuse;
+pub use fuse::Fuse;
 
 mod map;
-use map::Map;
+pub use map::Map;
 
 mod map_while;
-use map_while::MapWhile;
+pub use map_while::MapWhile;
 
 mod merge;
-use merge::Merge;
+pub use merge::Merge;
 
 mod next;
 use next::Next;
 
 mod skip;
-use skip::Skip;
+pub use skip::Skip;
 
 mod skip_while;
-use skip_while::SkipWhile;
+pub use skip_while::SkipWhile;
 
 mod take;
-use take::Take;
+pub use take::Take;
 
 mod take_while;
-use take_while::TakeWhile;
+pub use take_while::TakeWhile;
 
 mod then;
-use then::Then;
+pub use then::Then;
 
 mod try_next;
 use try_next::TryNext;
 
 mod peekable;
-use peekable::Peekable;
+pub use peekable::Peekable;
 
 cfg_time! {
     pub(crate) mod timeout;
     pub(crate) mod timeout_repeating;
-    use timeout::Timeout;
-    use timeout_repeating::TimeoutRepeating;
+    pub use timeout::Timeout;
+    pub use timeout_repeating::TimeoutRepeating;
     use tokio::time::{Duration, Interval};
     mod throttle;
     use throttle::{throttle, Throttle};
     mod chunks_timeout;
-    use chunks_timeout::ChunksTimeout;
+    pub use chunks_timeout::ChunksTimeout;
 }
 
 /// An extension trait for the [`Stream`] trait that provides a variety of

tokio-stream/src/stream_ext/all.rs

@@ -3,7 +3,7 @@ use crate::Stream;
 use core::future::Future;
 use core::marker::PhantomPinned;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {
@@ -42,7 +42,7 @@ where
 
         // Take a maximum of 32 items from the stream before yielding.
         for _ in 0..32 {
-            match futures_core::ready!(stream.as_mut().poll_next(cx)) {
+            match ready!(stream.as_mut().poll_next(cx)) {
                 Some(v) => {
                     if !(me.f)(v) {
                         return Poll::Ready(false);

tokio-stream/src/stream_ext/any.rs

@@ -3,7 +3,7 @@ use crate::Stream;
 use core::future::Future;
 use core::marker::PhantomPinned;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {
@@ -42,7 +42,7 @@ where
 
         // Take a maximum of 32 items from the stream before yielding.
         for _ in 0..32 {
-            match futures_core::ready!(stream.as_mut().poll_next(cx)) {
+            match ready!(stream.as_mut().poll_next(cx)) {
                 Some(v) => {
                     if (me.f)(v) {
                         return Poll::Ready(true);

tokio-stream/src/stream_ext/chain.rs

@@ -2,7 +2,7 @@ use crate::stream_ext::Fuse;
 use crate::Stream;
 
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -4,7 +4,7 @@ use tokio::time::{sleep, Sleep};
 
 use core::future::Future;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 use std::time::Duration;
 

tokio-stream/src/stream_ext/collect.rs

@@ -4,7 +4,7 @@ use core::future::Future;
 use core::marker::PhantomPinned;
 use core::mem;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 // Do not export this struct until `FromStream` can be unsealed.

tokio-stream/src/stream_ext/filter.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/filter_map.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/fold.rs

@@ -3,7 +3,7 @@ use crate::Stream;
 use core::future::Future;
 use core::marker::PhantomPinned;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/fuse.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use pin_project_lite::pin_project;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 pin_project! {
     /// Stream returned by [`fuse()`][super::StreamExt::fuse].

tokio-stream/src/stream_ext/peekable.rs

@@ -8,7 +8,7 @@ use crate::stream_ext::Fuse;
 use crate::StreamExt;
 
 pin_project! {
-    /// Stream returned by the [`chain`](super::StreamExt::peekable) method.
+    /// Stream returned by the [`peekable`](super::StreamExt::peekable) method.
     pub struct Peekable<T: Stream> {
         peek: Option<T::Item>,
         #[pin]

tokio-stream/src/stream_ext/skip.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/skip_while.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/throttle.rs

@@ -5,7 +5,7 @@ use tokio::time::{Duration, Instant, Sleep};
 
 use std::future::Future;
 use std::pin::Pin;
-use std::task::{self, Poll};
+use std::task::{self, ready, Poll};
 
 use pin_project_lite::pin_project;
 

tokio-stream/src/stream_ext/timeout.rs

@@ -4,7 +4,7 @@ use tokio::time::{Instant, Sleep};
 
 use core::future::Future;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 use std::fmt;
 use std::time::Duration;

tokio-stream/src/stream_ext/timeout_repeating.rs

@@ -3,7 +3,7 @@ use crate::{Elapsed, Stream};
 use tokio::time::Interval;
 
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_map.rs

@@ -1,9 +1,10 @@
-use crate::{poll_fn, Stream};
+use crate::Stream;
 
 use std::borrow::Borrow;
+use std::future::poll_fn;
 use std::hash::Hash;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 /// Combine many streams into one, indexing each source stream with a unique
 /// key.
@@ -467,10 +468,10 @@ impl<K, V> StreamMap<K, V> {
     /// assert!(map.remove(&1).is_some());
     /// assert!(map.remove(&1).is_none());
     /// ```
-    pub fn remove<Q: ?Sized>(&mut self, k: &Q) -> Option<V>
+    pub fn remove<Q>(&mut self, k: &Q) -> Option<V>
     where
         K: Borrow<Q>,
-        Q: Hash + Eq,
+        Q: Hash + Eq + ?Sized,
     {
         for i in 0..self.entries.len() {
             if self.entries[i].0.borrow() == k {
@@ -496,10 +497,10 @@ impl<K, V> StreamMap<K, V> {
     /// assert_eq!(map.contains_key(&1), true);
     /// assert_eq!(map.contains_key(&2), false);
     /// ```
-    pub fn contains_key<Q: ?Sized>(&self, k: &Q) -> bool
+    pub fn contains_key<Q>(&self, k: &Q) -> bool
     where
         K: Borrow<Q>,
-        Q: Hash + Eq,
+        Q: Hash + Eq + ?Sized,
     {
         for i in 0..self.entries.len() {
             if self.entries[i].0.borrow() == k {

tokio-stream/src/wrappers/broadcast.rs

@@ -6,12 +6,12 @@ use futures_core::Stream;
 use tokio_util::sync::ReusableBoxFuture;
 
 use std::fmt;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 /// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
 ///
 /// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
-/// [`Stream`]: trait@crate::Stream
+/// [`Stream`]: trait@futures_core::Stream
 #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
 pub struct BroadcastStream<T> {
     inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,

tokio-stream/src/wrappers/interval.rs

@@ -33,7 +33,7 @@ impl Stream for IntervalStream {
     }
 
     fn size_hint(&self) -> (usize, Option<usize>) {
-        (std::usize::MAX, None)
+        (usize::MAX, None)
     }
 }
 

tokio-stream/src/wrappers/watch.rs

@@ -5,7 +5,7 @@ use futures_core::Stream;
 use tokio_util::sync::ReusableBoxFuture;
 
 use std::fmt;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use tokio::sync::watch::error::RecvError;
 
 /// A wrapper around [`tokio::sync::watch::Receiver`] that implements [`Stream`].

tokio-stream/tests/async_send_sync.rs

@@ -15,18 +15,21 @@ fn require_unpin<T: Unpin>(_t: &T) {}
 #[allow(dead_code)]
 struct Invalid;
 
+#[allow(unused)]
 trait AmbiguousIfSend<A> {
     fn some_item(&self) {}
 }
 impl<T: ?Sized> AmbiguousIfSend<()> for T {}
 impl<T: ?Sized + Send> AmbiguousIfSend<Invalid> for T {}
 
+#[allow(unused)]
 trait AmbiguousIfSync<A> {
     fn some_item(&self) {}
 }
 impl<T: ?Sized> AmbiguousIfSync<()> for T {}
 impl<T: ?Sized + Sync> AmbiguousIfSync<Invalid> for T {}
 
+#[allow(unused)]
 trait AmbiguousIfUnpin<A> {
     fn some_item(&self) {}
 }

tokio-stream/tests/stream_chain.rs

@@ -6,13 +6,14 @@ mod support {
 }
 
 use support::mpsc;
+use tokio_stream::adapters::Chain;
 
 #[tokio::test]
 async fn basic_usage() {
     let one = stream::iter(vec![1, 2, 3]);
     let two = stream::iter(vec![4, 5, 6]);
 
-    let mut stream = one.chain(two);
+    let mut stream = visibility_test(one, two);
 
     assert_eq!(stream.size_hint(), (6, Some(6)));
     assert_eq!(stream.next().await, Some(1));
@@ -39,7 +40,16 @@ async fn basic_usage() {
     assert_eq!(stream.next().await, None);
 }
 
+fn visibility_test<I, S1, S2>(s1: S1, s2: S2) -> Chain<S1, S2>
+where
+    S1: Stream<Item = I>,
+    S2: Stream<Item = I>,
+{
+    s1.chain(s2)
+}
+
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // Block on https://github.com/tokio-rs/tokio/issues/6860
 async fn pending_first() {
     let (tx1, rx1) = mpsc::unbounded_channel_stream();
     let (tx2, rx2) = mpsc::unbounded_channel_stream();

tokio-stream/tests/stream_timeout.rs

@@ -1,7 +1,7 @@
 #![cfg(all(feature = "time", feature = "sync", feature = "io-util"))]
 
 use tokio::time::{self, sleep, Duration};
-use tokio_stream::{self, StreamExt};
+use tokio_stream::StreamExt;
 use tokio_test::*;
 
 use futures::stream;

tokio-test/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio-test"
 # - Create "tokio-test-0.4.x" git tag.
 version = "0.4.4"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"

tokio-test/LICENSE

@@ -1,25 +1,21 @@
-Copyright (c) 2023 Tokio Contributors
+MIT License
 
-Permission is hereby granted, free of charge, to any
-person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the
-Software without restriction, including without
-limitation the rights to use, copy, modify, merge,
-publish, distribute, sublicense, and/or sell copies of
-the Software, and to permit persons to whom the Software
-is furnished to do so, subject to the following
-conditions:
+Copyright (c) Tokio Contributors
 
-The above copyright notice and this permission notice
-shall be included in all copies or substantial portions
-of the Software.
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
 
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
-ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
-PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
-IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-DEALINGS IN THE SOFTWARE.
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

tokio-test/src/io.rs

@@ -23,13 +23,13 @@ use tokio::sync::mpsc;
 use tokio::time::{self, Duration, Instant, Sleep};
 use tokio_stream::wrappers::UnboundedReceiverStream;
 
-use futures_core::{ready, Stream};
+use futures_core::Stream;
 use std::collections::VecDeque;
 use std::fmt;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
-use std::task::{self, Poll, Waker};
+use std::task::{self, ready, Poll, Waker};
 use std::{cmp, io};
 
 /// An I/O object that follows a predefined script.

tokio-test/src/lib.rs

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

tokio-test/src/stream_mock.rs

@@ -36,10 +36,10 @@
 
 use std::collections::VecDeque;
 use std::pin::Pin;
-use std::task::Poll;
+use std::task::{ready, Poll};
 use std::time::Duration;
 
-use futures_core::{ready, Stream};
+use futures_core::Stream;
 use std::future::Future;
 use tokio::time::{sleep_until, Instant, Sleep};
 
@@ -161,8 +161,7 @@ impl<T: Unpin> Drop for StreamMock<T> {
 
         assert!(
             undropped_count == 0,
-            "StreamMock was dropped before all actions were consumed, {} actions were not consumed",
-            undropped_count
+            "StreamMock was dropped before all actions were consumed, {undropped_count} actions were not consumed"
         );
     }
 }

tokio-test/src/task.rs

@@ -148,6 +148,10 @@ impl<T: Stream> Stream for Spawn<T> {
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.future.as_mut().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.future.size_hint()
+    }
 }
 
 impl MockTask {

tokio-test/tests/io.rs

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

tokio-test/tests/task.rs

@@ -0,0 +1,25 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio_stream::Stream;
+use tokio_test::task;
+
+/// A [`Stream`] that has a stub size hint.
+struct SizedStream;
+
+impl Stream for SizedStream {
+    type Item = ();
+
+    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Poll::Pending
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (100, Some(200))
+    }
+}
+
+#[test]
+fn test_spawn_stream_size_hint() {
+    let spawn = task::spawn(SizedStream);
+    assert_eq!(spawn.size_hint(), (100, Some(200)));
+}

tokio-util/CHANGELOG.md

@@ -1,3 +1,68 @@
+# 0.7.13 (December 4th, 2024)
+
+### Fixed
+
+- codec: fix incorrect handling of invalid utf-8 in `LinesCodec::decode_eof` ([#7011])
+
+[#7011]: https://github.com/tokio-rs/tokio/pull/7011
+
+# 0.7.12 (September 5th, 2024)
+
+This release bumps the MSRV to 1.70. ([#6645])
+
+### Added
+- sync: Add `run_until_cancelled` to `tokio_util::sync::CancellationToken` ([#6618])
+- task: add `AbortOnDropHandle` type ([#6786])
+
+### Changed
+- deps: no default features for hashbrown ([#6541])
+- time: wake `DelayQueue` when removing last item ([#6752])
+- deps: enable the full feature when compiled for the playground ([#6818])
+
+### Documented
+- task: fix typo in `TaskTracker` docs ([#6792])
+
+[#6645]: https://github.com/tokio-rs/tokio/pull/6645
+[#6541]: https://github.com/tokio-rs/tokio/pull/6541
+[#6618]: https://github.com/tokio-rs/tokio/pull/6618
+[#6752]: https://github.com/tokio-rs/tokio/pull/6752
+[#6786]: https://github.com/tokio-rs/tokio/pull/6786
+[#6792]: https://github.com/tokio-rs/tokio/pull/6792
+[#6818]: https://github.com/tokio-rs/tokio/pull/6818
+
+# 0.7.11 (May 4th, 2024)
+
+This release updates the MSRV to 1.63. ([#6126])
+
+### Added
+
+- either: implement `Sink` for `Either` ([#6239])
+- time: add `DelayQueue::deadline` ([#6163])
+- time: add `FutureExt::timeout` ([#6276])
+
+### Changed
+
+- codec: assert compatibility between `LengthDelimitedCodec` options ([#6414])
+- codec: make tracing feature optional for codecs ([#6434])
+- io: add `T: ?Sized` to `tokio_util::io::poll_read_buf` ([#6441])
+- sync: remove `'static` bound on `impl Sink for PollSender` ([#6397])
+
+### Documented
+
+- codec: add examples for `FramedRead` and `FramedWrite` ([#6310])
+- codec: document cancel safety of `SinkExt::send` and `StreamExt::next` ([#6417])
+
+[#6126]: https://github.com/tokio-rs/tokio/pull/6126
+[#6163]: https://github.com/tokio-rs/tokio/pull/6163
+[#6239]: https://github.com/tokio-rs/tokio/pull/6239
+[#6276]: https://github.com/tokio-rs/tokio/pull/6276
+[#6310]: https://github.com/tokio-rs/tokio/pull/6310
+[#6397]: https://github.com/tokio-rs/tokio/pull/6397
+[#6414]: https://github.com/tokio-rs/tokio/pull/6414
+[#6417]: https://github.com/tokio-rs/tokio/pull/6417
+[#6434]: https://github.com/tokio-rs/tokio/pull/6434
+[#6441]: https://github.com/tokio-rs/tokio/pull/6441
+
 # 0.7.10 (October 24th, 2023)
 
 ### 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.10"
+version = "0.7.13"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
@@ -25,7 +25,7 @@ full = ["codec", "compat", "io-util", "time", "net", "rt"]
 
 net = ["tokio/net"]
 compat = ["futures-io",]
-codec = ["tracing"]
+codec = []
 time = ["tokio/time","slab"]
 io = []
 io-util = ["io", "tokio/rt", "tokio/io-util"]
@@ -42,10 +42,10 @@ futures-io = { version = "0.3.0", optional = true }
 futures-util = { version = "0.3.0", optional = true }
 pin-project-lite = "0.2.11"
 slab = { version = "0.4.4", optional = true } # Backs `DelayQueue`
-tracing = { version = "0.1.25", default-features = false, features = ["std"], optional = true }
+tracing = { version = "0.1.29", default-features = false, features = ["std"], optional = true }
 
 [target.'cfg(tokio_unstable)'.dependencies]
-hashbrown = { version = "0.14.0", optional = true }
+hashbrown = { version = "0.14.0", default-features = false, optional = true }
 
 [dev-dependencies]
 tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
@@ -65,3 +65,6 @@ 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"]
+
+[package.metadata.playground]
+features = ["full"]

tokio-util/LICENSE

@@ -1,25 +1,21 @@
-Copyright (c) 2023 Tokio Contributors
+MIT License
 
-Permission is hereby granted, free of charge, to any
-person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the
-Software without restriction, including without
-limitation the rights to use, copy, modify, merge,
-publish, distribute, sublicense, and/or sell copies of
-the Software, and to permit persons to whom the Software
-is furnished to do so, subject to the following
-conditions:
+Copyright (c) Tokio Contributors
 
-The above copyright notice and this permission notice
-shall be included in all copies or substantial portions
-of the Software.
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
 
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
-ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
-PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
-IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-DEALINGS IN THE SOFTWARE.
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

tokio-util/src/codec/any_delimiter_codec.rs

@@ -2,7 +2,7 @@ use crate::codec::decoder::Decoder;
 use crate::codec::encoder::Encoder;
 
 use bytes::{Buf, BufMut, Bytes, BytesMut};
-use std::{cmp, fmt, io, str, usize};
+use std::{cmp, fmt, io, str};
 
 const DEFAULT_SEEK_DELIMITERS: &[u8] = b",;\n\r";
 const DEFAULT_SEQUENCE_WRITER: &[u8] = b",";
@@ -249,7 +249,7 @@ impl fmt::Display for AnyDelimiterCodecError {
             AnyDelimiterCodecError::MaxChunkLengthExceeded => {
                 write!(f, "max chunk length exceeded")
             }
-            AnyDelimiterCodecError::Io(e) => write!(f, "{}", e),
+            AnyDelimiterCodecError::Io(e) => write!(f, "{e}"),
         }
     }
 }

tokio-util/src/codec/encoder.rs

@@ -9,7 +9,7 @@ pub trait Encoder<Item> {
     /// The type of encoding errors.
     ///
     /// [`FramedWrite`] requires `Encoder`s errors to implement `From<io::Error>`
-    /// in the interest letting it return `Error`s directly.
+    /// in the interest of letting it return `Error`s directly.
     ///
     /// [`FramedWrite`]: crate::codec::FramedWrite
     type Error: From<io::Error>;

tokio-util/src/codec/framed_impl.rs

@@ -5,14 +5,12 @@ use futures_core::Stream;
 use tokio::io::{AsyncRead, AsyncWrite};
 
 use bytes::BytesMut;
-use futures_core::ready;
 use futures_sink::Sink;
 use pin_project_lite::pin_project;
 use std::borrow::{Borrow, BorrowMut};
 use std::io;
 use std::pin::Pin;
-use std::task::{Context, Poll};
-use tracing::trace;
+use std::task::{ready, Context, Poll};
 
 pin_project! {
     #[derive(Debug)]

tokio-util/src/codec/length_delimited.rs

@@ -75,55 +75,23 @@
 //!
 //! ## Example 1
 //!
-//! The following will parse a `u16` length field at offset 0, including the
-//! frame head in the yielded `BytesMut`.
-//!
-//! ```
-//! # use tokio::io::AsyncRead;
-//! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
-//!     .length_field_offset(0) // default value
-//!     .length_field_type::<u16>()
-//!     .length_adjustment(0)   // default value
-//!     .num_skip(0) // Do not strip frame header
-//!     .new_read(io);
-//! # }
-//! # pub fn main() {}
-//! ```
-//!
-//! The following frame will be decoded as such:
-//!
-//! ```text
-//!          INPUT                           DECODED
-//! +-- len ---+--- Payload ---+     +-- len ---+--- Payload ---+
-//! | \x00\x0B |  Hello world  | --> | \x00\x0B |  Hello world  |
-//! +----------+---------------+     +----------+---------------+
-//! ```
-//!
-//! The value of the length field is 11 (`\x0B`) which represents the length
-//! of the payload, `hello world`. By default, [`FramedRead`] assumes that
-//! the length field represents the number of bytes that **follows** the
-//! length field. Thus, the entire frame has a length of 13: 2 bytes for the
-//! frame head + 11 bytes for the payload.
-//!
-//! ## Example 2
-//!
 //! The following will parse a `u16` length field at offset 0, omitting the
 //! frame head in the yielded `BytesMut`.
 //!
 //! ```
-//! # use tokio::io::AsyncRead;
+//! # use tokio_stream::StreamExt;
 //! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\x00\x0BHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
 //!     .length_field_type::<u16>()
 //!     .length_adjustment(0)   // default value
-//!     // `num_skip` is not needed, the default is to skip
 //!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"Hello world");
 //! # }
-//! # pub fn main() {}
 //! ```
 //!
 //! The following frame will be decoded as such:
@@ -135,27 +103,32 @@
 //! +----------+---------------+     +---------------+
 //! ```
 //!
-//! This is similar to the first example, the only difference is that the
-//! frame head is **not** included in the yielded `BytesMut` value.
+//! The value of the length field is 11 (`\x0B`) which represents the length
+//! of the payload, `hello world`. By default, [`FramedRead`] assumes that
+//! the length field represents the number of bytes that **follows** the
+//! length field. Thus, the entire frame has a length of 13: 2 bytes for the
+//! frame head + 11 bytes for the payload.
 //!
-//! ## Example 3
+//! ## Example 2
 //!
 //! The following will parse a `u16` length field at offset 0, including the
-//! frame head in the yielded `BytesMut`. In this case, the length field
-//! **includes** the frame head length.
+//! frame head in the yielded `BytesMut`.
 //!
 //! ```
-//! # use tokio::io::AsyncRead;
+//! # use tokio_stream::StreamExt;
 //! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\x00\x0BHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
 //!     .length_field_type::<u16>()
-//!     .length_adjustment(-2)  // size of head
-//!     .num_skip(0)
+//!     .length_adjustment(2)   // Add head size to length
+//!     .num_skip(0)            // Do NOT skip the head
 //!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"\x00\x0BHello world");
 //! # }
-//! # pub fn main() {}
 //! ```
 //!
 //! The following frame will be decoded as such:
@@ -163,10 +136,46 @@
 //! ```text
 //!          INPUT                           DECODED
 //! +-- len ---+--- Payload ---+     +-- len ---+--- Payload ---+
-//! | \x00\x0D |  Hello world  | --> | \x00\x0D |  Hello world  |
+//! | \x00\x0B |  Hello world  | --> | \x00\x0B |  Hello world  |
 //! +----------+---------------+     +----------+---------------+
 //! ```
 //!
+//! This is similar to the first example, the only difference is that the
+//! frame head is **included** in the yielded `BytesMut` value. To achieve
+//! this, we need to add the header size to the length with `length_adjustment`,
+//! and set `num_skip` to `0` to prevent skipping the head.
+//!
+//! ## Example 3
+//!
+//! The following will parse a `u16` length field at offset 0, omitting the
+//! frame head in the yielded `BytesMut`. In this case, the length field
+//! **includes** the frame head length.
+//!
+//! ```
+//! # use tokio_stream::StreamExt;
+//! # use tokio_util::codec::LengthDelimitedCodec;
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\x00\x0DHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
+//!     .length_field_offset(0) // default value
+//!     .length_field_type::<u16>()
+//!     .length_adjustment(-2)  // size of head
+//!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"Hello world");
+//! # }
+//! ```
+//!
+//! The following frame will be decoded as such:
+//!
+//! ```text
+//!          INPUT                           DECODED
+//! +-- len ---+--- Payload ---+     +--- Payload ---+
+//! | \x00\x0D |  Hello world  | --> |  Hello world  |
+//! +----------+---------------+     +---------------+
+//! ```
+//!
 //! In most cases, the length field represents the length of the payload
 //! only, as shown in the previous examples. However, in some protocols the
 //! length field represents the length of the whole frame, including the
@@ -179,17 +188,20 @@
 //! frame head, including the frame head in the yielded `BytesMut`.
 //!
 //! ```
-//! # use tokio::io::AsyncRead;
+//! # use tokio_stream::StreamExt;
 //! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\x00\x00\x0B\xCA\xFEHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
 //!     .length_field_length(3)
-//!     .length_adjustment(2)  // remaining head
+//!     .length_adjustment(3 + 2)  // len field and remaining head
 //!     .num_skip(0)
 //!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"\x00\x00\x0B\xCA\xFEHello world");
 //! # }
-//! # pub fn main() {}
 //! ```
 //!
 //! The following frame will be decoded as such:
@@ -223,17 +235,20 @@
 //! included.
 //!
 //! ```
-//! # use tokio::io::AsyncRead;
+//! # use tokio_stream::StreamExt;
 //! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\xCA\x00\x0B\xFEHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
 //!     .length_field_offset(1) // length of hdr1
 //!     .length_field_type::<u16>()
 //!     .length_adjustment(1)  // length of hdr2
 //!     .num_skip(3) // length of hdr1 + LEN
 //!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"\xFEHello world");
 //! # }
-//! # pub fn main() {}
 //! ```
 //!
 //! The following frame will be decoded as such:
@@ -269,15 +284,19 @@
 //! length.
 //!
 //! ```
-//! # use tokio::io::AsyncRead;
+//! # use tokio_stream::StreamExt;
 //! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\xCA\x00\x0F\xFEHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
 //!     .length_field_offset(1) // length of hdr1
 //!     .length_field_type::<u16>()
 //!     .length_adjustment(-3)  // length of hdr1 + LEN, negative
 //!     .num_skip(3)
 //!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"\xFEHello world");
 //! # }
 //! ```
 //!
@@ -308,17 +327,20 @@
 //! frame head, excluding the 4th byte from the yielded `BytesMut`.
 //!
 //! ```
-//! # use tokio::io::AsyncRead;
+//! # use tokio_stream::StreamExt;
 //! # use tokio_util::codec::LengthDelimitedCodec;
-//! # fn bind_read<T: AsyncRead>(io: T) {
-//! LengthDelimitedCodec::builder()
+//! # #[tokio::main]
+//! # async fn main() {
+//! # let io: &[u8] = b"\x00\x00\x0B\xFFHello world";
+//! let mut reader = LengthDelimitedCodec::builder()
 //!     .length_field_offset(0) // default value
 //!     .length_field_length(3)
 //!     .length_adjustment(0)  // default value
 //!     .num_skip(4) // skip the first 4 bytes
 //!     .new_read(io);
+//! # let res = reader.next().await.unwrap().unwrap().to_vec();
+//! # assert_eq!(res, b"Hello world");
 //! # }
-//! # pub fn main() {}
 //! ```
 //!
 //! The following frame will be decoded as such:
@@ -639,7 +661,6 @@ mod builder {
     impl LengthFieldType for u64 {}
 
     #[cfg(any(
-        target_pointer_width = "8",
         target_pointer_width = "16",
         target_pointer_width = "32",
         target_pointer_width = "64",
@@ -1028,28 +1049,13 @@ impl Builder {
     }
 
     fn adjust_max_frame_len(&mut self) {
-        // This function is basically `std::u64::saturating_add_signed`. Since it
-        // requires MSRV 1.66, its implementation is copied here.
-        //
-        // TODO: use the method from std when MSRV becomes >= 1.66
-        fn saturating_add_signed(num: u64, rhs: i64) -> u64 {
-            let (res, overflow) = num.overflowing_add(rhs as u64);
-            if overflow == (rhs < 0) {
-                res
-            } else if overflow {
-                u64::MAX
-            } else {
-                0
-            }
-        }
-
         // Calculate the maximum number that can be represented using `length_field_len` bytes.
         let max_number = match 1u64.checked_shl((8 * self.length_field_len) as u32) {
             Some(shl) => shl - 1,
             None => u64::MAX,
         };
 
-        let max_allowed_len = saturating_add_signed(max_number, self.length_adjustment as i64);
+        let max_allowed_len = max_number.saturating_add_signed(self.length_adjustment as i64);
 
         if self.max_frame_len as u64 > max_allowed_len {
             self.max_frame_len = usize::try_from(max_allowed_len).unwrap_or(usize::MAX);

tokio-util/src/codec/lines_codec.rs

@@ -2,7 +2,7 @@ use crate::codec::decoder::Decoder;
 use crate::codec::encoder::Encoder;
 
 use bytes::{Buf, BufMut, BytesMut};
-use std::{cmp, fmt, io, str, usize};
+use std::{cmp, fmt, io, str};
 
 /// A simple [`Decoder`] and [`Encoder`] implementation that splits up data into lines.
 ///
@@ -169,6 +169,7 @@ impl Decoder for LinesCodec {
         Ok(match self.decode(buf)? {
             Some(frame) => Some(frame),
             None => {
+                self.next_index = 0;
                 // No terminating newline - return remaining data, if any
                 if buf.is_empty() || buf == &b"\r"[..] {
                     None
@@ -176,7 +177,6 @@ impl Decoder for LinesCodec {
                     let line = buf.split_to(buf.len());
                     let line = without_carriage_return(&line);
                     let line = utf8(line)?;
-                    self.next_index = 0;
                     Some(line.to_string())
                 }
             }
@@ -218,7 +218,7 @@ impl fmt::Display for LinesCodecError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             LinesCodecError::MaxLineLengthExceeded => write!(f, "max line length exceeded"),
-            LinesCodecError::Io(e) => write!(f, "{}", e),
+            LinesCodecError::Io(e) => write!(f, "{e}"),
         }
     }
 }

tokio-util/src/compat.rs

@@ -1,10 +1,9 @@
 //! Compatibility between the `tokio::io` and `futures-io` versions of the
 //! `AsyncRead` and `AsyncWrite` traits.
-use futures_core::ready;
 use pin_project_lite::pin_project;
 use std::io;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 pin_project! {
     /// A compatibility layer that allows conversion between the

tokio-util/src/io/inspect.rs

@@ -1,8 +1,7 @@
-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 std::task::{ready, Context, Poll};
 
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 

tokio-util/src/io/sink_writer.rs

@@ -1,11 +1,10 @@
-use futures_core::ready;
 use futures_sink::Sink;
 
 use futures_core::stream::Stream;
 use pin_project_lite::pin_project;
 use std::io;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use tokio::io::{AsyncRead, AsyncWrite};
 
 pin_project! {

tokio-util/src/io/stream_reader.rs

@@ -326,7 +326,7 @@ impl<S, B> StreamReader<S, B> {
     }
 }
 
-impl<S: Sink<T, Error = E>, E, T> Sink<T> for StreamReader<S, E> {
+impl<S: Sink<T, Error = E>, B, E, T> Sink<T> for StreamReader<S, B> {
     type Error = E;
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.project().inner.poll_ready(cx)

tokio-util/src/lib.rs

@@ -1,3 +1,4 @@
+#![allow(unknown_lints, unexpected_cfgs)]
 #![allow(clippy::needless_doctest_main)]
 #![warn(
     missing_debug_implementations,
@@ -25,6 +26,9 @@ mod cfg;
 mod loom;
 
 cfg_codec! {
+    #[macro_use]
+    mod tracing;
+
     pub mod codec;
 }
 

tokio-util/src/sync/cancellation_token.rs

@@ -241,6 +241,52 @@ impl CancellationToken {
     pub fn drop_guard(self) -> DropGuard {
         DropGuard { inner: Some(self) }
     }
+
+    /// Runs a future to completion and returns its result wrapped inside of an `Option`
+    /// unless the `CancellationToken` is cancelled. In that case the function returns
+    /// `None` and the future gets dropped.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is only cancel safe if `fut` is cancel safe.
+    pub async fn run_until_cancelled<F>(&self, fut: F) -> Option<F::Output>
+    where
+        F: Future,
+    {
+        pin_project! {
+            /// A Future that is resolved once the corresponding [`CancellationToken`]
+            /// is cancelled or a given Future gets resolved. It is biased towards the
+            /// Future completion.
+            #[must_use = "futures do nothing unless polled"]
+            struct RunUntilCancelledFuture<'a, F: Future> {
+                #[pin]
+                cancellation: WaitForCancellationFuture<'a>,
+                #[pin]
+                future: F,
+            }
+        }
+
+        impl<'a, F: Future> Future for RunUntilCancelledFuture<'a, F> {
+            type Output = Option<F::Output>;
+
+            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                let this = self.project();
+                if let Poll::Ready(res) = this.future.poll(cx) {
+                    Poll::Ready(Some(res))
+                } else if this.cancellation.poll(cx).is_ready() {
+                    Poll::Ready(None)
+                } else {
+                    Poll::Pending
+                }
+            }
+        }
+
+        RunUntilCancelledFuture {
+            cancellation: self.cancelled(),
+            future: fut,
+        }
+        .await
+    }
 }
 
 // ===== impl WaitForCancellationFuture =====

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

@@ -206,7 +206,7 @@ fn move_children_to_parent(node: &mut Inner, parent: &mut Inner) {
     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.clone_from(&node.parent);
             child_locked.parent_idx = parent.children.len();
         }
         parent.children.push(child);

tokio-util/src/sync/poll_semaphore.rs

@@ -1,8 +1,8 @@
-use futures_core::{ready, Stream};
+use futures_core::Stream;
 use std::fmt;
 use std::pin::Pin;
 use std::sync::Arc;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use tokio::sync::{AcquireError, OwnedSemaphorePermit, Semaphore, TryAcquireError};
 
 use super::ReusableBoxFuture;