chore: prepare tokio-stream v0.1.17 (#7020)

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.

.circleci/config.yml

@@ -1,4 +1,4 @@
-version: 2.1        
+version: 2.1
 jobs:
   test-arm:
     machine:

.github/buildomat/README.md

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

.github/workflows/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

@@ -17,6 +17,8 @@ env:
   # Change to specific Rust release to pin
   rust_stable: stable
   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
@@ -45,7 +47,6 @@ jobs:
       - test-workspace-all-features
       - test-integration-tests-per-feature
       - test-parking_lot
-      - test-tracing-instrumentation
       - valgrind
       - test-unstable
       - miri
@@ -178,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
@@ -244,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
@@ -281,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
@@ -413,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
@@ -764,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 }}
@@ -776,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
@@ -948,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
 
@@ -1031,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 }}
@@ -1042,7 +1025,7 @@ 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
@@ -1102,16 +1085,39 @@ 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
-
+        run: |
+          if ! cargo spellcheck --code 1
+          then
+              echo ''
+              echo ''
+              echo 'If this is a Rust method/type/variable name, then you should'
+              echo 'enclose it in backticks like this: `MyRustType`.'
+              echo ''
+              echo 'If this is a real word, then you can add it to spellcheck.dic'
+              exit 1
+          fi
+      - name: Detect trailing whitespace
+        run: |
+          if grep --exclude-dir=.git --exclude-dir=target -rne '\s$' .
+          then
+              echo ''
+              echo 'Please remove trailing whitespace from these lines.'
+              exit 1
+          fi

.github/workflows/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

@@ -196,12 +196,12 @@ LOOM_MAX_PREEMPTIONS=1 LOOM_MAX_BRANCHES=10000 RUSTFLAGS="--cfg loom -C debug_as
     cargo test --lib --release --features full -- --test-threads=1 --nocapture
 ```
 Additionally, you can also add `--cfg tokio_unstable` to the `RUSTFLAGS` environment variable to
-run loom tests that test unstable features. 
+run loom tests that test unstable features.
 
 You can run miri tests with
 ```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-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
@@ -216,8 +216,8 @@ cargo install --locked cargo-spellcheck
 cargo spellcheck check
 ```
 
-if the command rejects a word, you should backtick the rejected word if it's code related. If not, the 
-rejected word should be put into `spellcheck.dic` file. 
+if the command rejects a word, you should backtick the rejected word if it's code related. If not, the
+rejected word should be put into `spellcheck.dic` file.
 
 Note that when you add a word into the file, you should also update the first line which tells the spellcheck tool
 the total number of words included in the file
@@ -269,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;
 

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.40.0", features = ["full"] }
+tokio = { version = "1.42.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -216,7 +216,6 @@ 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)
 
@@ -238,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"]
@@ -15,7 +16,7 @@ rand_chacha = "0.3"
 
 [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"

benches/sync_mpsc.rs

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

deny.toml

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

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/echo-udp.rs

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

examples/echo.rs

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

examples/print_each_packet.rs

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

examples/proxy.rs

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

examples/tinydb.rs

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

examples/tinyhttp.rs

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

examples/udp-codec.rs

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

spellcheck.dic

@@ -1,4 +1,4 @@
-285
+292
 &
 +
 <
@@ -12,12 +12,17 @@
 0xA
 0xD
 100ms
+100ns
 10ms
+10μs
 ~12
+120s
+12.5%
 ±1m
 ±1ms
 1ms
 1s
+25%
 250ms
 2x
 ~4
@@ -99,6 +104,7 @@ errored
 EWMA
 expirations
 fcntl
+fd
 fd's
 FIFOs
 filename
@@ -115,9 +121,11 @@ GID
 goroutines
 Growable
 gzip
+H2
 hashmaps
 HashMaps
 hashsets
+HdrHistogram
 ie
 Illumos
 impl
@@ -151,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"

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/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]]
@@ -55,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/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;
@@ -206,7 +206,7 @@ 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 {

tokio-macros/src/entry.rs

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

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,26 @@
+# 0.1.17 (December 6th, 2024)
+
+- deps: fix dev-dependency on tokio-test ([#6931], [#7019])
+- stream: fix link on `Peekable` ([#6861])
+- sync: fix `Stream` link in broadcast docs ([#6873])
+
+[#6861]: https://github.com/tokio-rs/tokio/pull/6861
+[#6873]: https://github.com/tokio-rs/tokio/pull/6873
+[#6931]: https://github.com/tokio-rs/tokio/pull/6931
+[#7019]: https://github.com/tokio-rs/tokio/pull/7019
+
+# 0.1.16 (September 5th, 2024)
+
+This release bumps the MSRV of tokio-stream to 1.70.
+
+- 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,7 +4,7 @@ name = "tokio-stream"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.15"
+version = "0.1.17"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -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/src/lib.rs

@@ -74,9 +74,6 @@
 #[macro_use]
 mod macros;
 
-mod poll_fn;
-pub(crate) use poll_fn::poll_fn;
-
 pub mod wrappers;
 
 mod stream_ext;

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/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_map.rs

@@ -1,6 +1,7 @@
-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::{ready, Context, Poll};

tokio-stream/src/wrappers/broadcast.rs

@@ -11,7 +11,7 @@ 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/tests/stream_chain.rs

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

tokio-test/src/stream_mock.rs

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

tokio-test/tests/io.rs

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

tokio-util/CHANGELOG.md

@@ -1,3 +1,35 @@
+# 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])

tokio-util/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-util"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.11"
+version = "0.7.13"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -42,7 +42,7 @@ 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", default-features = false, optional = true }
@@ -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/src/codec/any_delimiter_codec.rs

@@ -249,7 +249,7 @@ impl fmt::Display for AnyDelimiterCodecError {
             AnyDelimiterCodecError::MaxChunkLengthExceeded => {
                 write!(f, "max chunk length exceeded")
             }
-            AnyDelimiterCodecError::Io(e) => write!(f, "{}", e),
+            AnyDelimiterCodecError::Io(e) => write!(f, "{e}"),
         }
     }
 }

tokio-util/src/codec/lines_codec.rs

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

tokio-util/src/task/spawn_pinned.rs

@@ -14,7 +14,7 @@ use tokio::task::{spawn_local, JoinHandle, LocalSet};
 /// Internally the local pool uses a [`tokio::task::LocalSet`] for each worker thread
 /// in the pool. Consequently you can also use [`tokio::task::spawn_local`] (which will
 /// execute on the same thread) inside the Future you supply to the various spawn methods
-/// of `LocalPoolHandle`,
+/// of `LocalPoolHandle`.
 ///
 /// [`tokio::task::LocalSet`]: tokio::task::LocalSet
 /// [`tokio::task::spawn_local`]: tokio::task::spawn_local
@@ -39,9 +39,9 @@ use tokio::task::{spawn_local, JoinHandle, LocalSet};
 ///             task::spawn_local(async move {
 ///                 println!("{}", data_clone);
 ///             });
-///     
+///
 ///             data.to_string()
-///         }   
+///         }
 ///     }).await.unwrap();
 ///     println!("output: {}", output);
 /// }
@@ -194,7 +194,7 @@ enum WorkerChoice {
 }
 
 struct LocalPool {
-    workers: Vec<LocalWorkerHandle>,
+    workers: Box<[LocalWorkerHandle]>,
 }
 
 impl LocalPool {
@@ -249,7 +249,7 @@ impl LocalPool {
             // Send the callback to the LocalSet task
             if let Err(e) = worker_spawner.send(spawn_task) {
                 // Propagate the error as a panic in the join handle.
-                panic!("Failed to send job to worker: {}", e);
+                panic!("Failed to send job to worker: {e}");
             }
 
             // Wait for the task's join handle
@@ -260,7 +260,7 @@ impl LocalPool {
                     // join handle... We assume something happened to the worker
                     // and the task was not spawned. Propagate the error as a
                     // panic in the join handle.
-                    panic!("Worker failed to send join handle: {}", e);
+                    panic!("Worker failed to send join handle: {e}");
                 }
             };
 
@@ -284,12 +284,12 @@ impl LocalPool {
                         // No one else should have the join handle, so this is
                         // unexpected. Forward this error as a panic in the join
                         // handle.
-                        panic!("spawn_pinned task was canceled: {}", e);
+                        panic!("spawn_pinned task was canceled: {e}");
                     } else {
                         // Something unknown happened (not a panic or
                         // cancellation). Forward this error as a panic in the
                         // join handle.
-                        panic!("spawn_pinned task failed: {}", e);
+                        panic!("spawn_pinned task failed: {e}");
                     }
                 }
             }

tokio-util/src/time/delay_queue.rs

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

tokio-util/src/time/mod.rs

@@ -8,7 +8,7 @@
 //!
 //! This type must be used from within the context of the `Runtime`.
 
-use futures_core::Future;
+use std::future::Future;
 use std::time::Duration;
 use tokio::time::Timeout;
 

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

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

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

@@ -34,7 +34,7 @@ pub(crate) struct Wheel<T> {
     /// * ~ 4 min slots / ~ 4 hr range
     /// * ~ 4 hr slots / ~ 12 day range
     /// * ~ 12 day slots / ~ 2 yr range
-    levels: Vec<Level<T>>,
+    levels: Box<[Level<T>]>,
 }
 
 /// Number of levels. Each level has 64 slots. By using 6 levels with 64 slots
@@ -280,13 +280,7 @@ mod test {
     #[test]
     fn test_level_for() {
         for pos in 0..64 {
-            assert_eq!(
-                0,
-                level_for(0, pos),
-                "level_for({}) -- binary = {:b}",
-                pos,
-                pos
-            );
+            assert_eq!(0, level_for(0, pos), "level_for({pos}) -- binary = {pos:b}");
         }
 
         for level in 1..5 {
@@ -295,9 +289,7 @@ mod test {
                 assert_eq!(
                     level,
                     level_for(0, a as u64),
-                    "level_for({}) -- binary = {:b}",
-                    a,
-                    a
+                    "level_for({a}) -- binary = {a:b}"
                 );
 
                 if pos > level {
@@ -305,9 +297,7 @@ mod test {
                     assert_eq!(
                         level,
                         level_for(0, a as u64),
-                        "level_for({}) -- binary = {:b}",
-                        a,
-                        a
+                        "level_for({a}) -- binary = {a:b}"
                     );
                 }
 
@@ -316,9 +306,7 @@ mod test {
                     assert_eq!(
                         level,
                         level_for(0, a as u64),
-                        "level_for({}) -- binary = {:b}",
-                        a,
-                        a
+                        "level_for({a}) -- binary = {a:b}"
                     );
                 }
             }

tokio-util/src/util/poll_buf.rs

@@ -17,7 +17,7 @@ use std::task::{ready, Context, Poll};
 /// use tokio_stream as stream;
 /// use tokio::io::Result;
 /// use tokio_util::io::{StreamReader, poll_read_buf};
-/// use futures::future::poll_fn;
+/// use std::future::poll_fn;
 /// use std::pin::Pin;
 /// # #[tokio::main]
 /// # async fn main() -> std::io::Result<()> {
@@ -95,9 +95,9 @@ pub fn poll_read_buf<T: AsyncRead + ?Sized, B: BufMut>(
 /// use tokio::fs::File;
 ///
 /// use bytes::Buf;
+/// use std::future::poll_fn;
 /// use std::io::Cursor;
 /// use std::pin::Pin;
-/// use futures::future::poll_fn;
 ///
 /// #[tokio::main]
 /// async fn main() -> io::Result<()> {

tokio-util/tests/codecs.rs

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

tokio-util/tests/framed_write.rs

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

tokio-util/tests/io_inspect.rs

@@ -1,5 +1,5 @@
-use futures::future::poll_fn;
 use std::{
+    future::poll_fn,
     io::IoSlice,
     pin::Pin,
     task::{Context, Poll},

tokio-util/tests/io_reader_stream.rs

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

tokio-util/tests/length_delimited.rs

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

tokio-util/tests/mpsc.rs

@@ -1,5 +1,5 @@
-use futures::future::poll_fn;
 use futures::sink::SinkExt;
+use std::future::poll_fn;
 use tokio::sync::mpsc::channel;
 use tokio_test::task::spawn;
 use tokio_test::{

tokio-util/tests/poll_semaphore.rs

@@ -9,7 +9,7 @@ type SemRet = Option<OwnedSemaphorePermit>;
 fn semaphore_poll(
     sem: &mut PollSemaphore,
 ) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + '_> {
-    let fut = futures::future::poll_fn(move |cx| sem.poll_acquire(cx));
+    let fut = std::future::poll_fn(move |cx| sem.poll_acquire(cx));
     tokio_test::task::spawn(fut)
 }
 
@@ -17,7 +17,7 @@ fn semaphore_poll_many(
     sem: &mut PollSemaphore,
     permits: u32,
 ) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + '_> {
-    let fut = futures::future::poll_fn(move |cx| sem.poll_acquire_many(cx, permits));
+    let fut = std::future::poll_fn(move |cx| sem.poll_acquire_many(cx, permits));
     tokio_test::task::spawn(fut)
 }
 

tokio-util/tests/spawn_pinned.rs

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

tokio-util/tests/time_delay_queue.rs

@@ -92,6 +92,7 @@ async fn single_short_delay() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // Too slow on miri.
 async fn multi_delay_at_start() {
     time::pause();
 
@@ -110,7 +111,7 @@ async fn multi_delay_at_start() {
 
     let start = Instant::now();
     for elapsed in 0..1200 {
-        println!("elapsed: {:?}", elapsed);
+        println!("elapsed: {elapsed:?}");
         let elapsed = elapsed + 1;
         tokio::time::sleep_until(start + ms(elapsed)).await;
 
@@ -327,7 +328,7 @@ async fn remove_at_timer_wheel_threshold() {
             let entry = queue.remove(&key1).into_inner();
             assert_eq!(entry, "foo");
         }
-        other => panic!("other: {:?}", other),
+        other => panic!("other: {other:?}"),
     }
 }
 

tokio-util/tests/udp.rs

@@ -1,5 +1,6 @@
 #![warn(rust_2018_idioms)]
 #![cfg(not(target_os = "wasi"))] // Wasi doesn't support UDP
+#![cfg(not(miri))] // No `socket` in Miri.
 
 use tokio::net::UdpSocket;
 use tokio_stream::StreamExt;

tokio/CHANGELOG.md

@@ -1,3 +1,104 @@
+# 1.42.0 (Dec 3rd, 2024)
+
+### Added
+
+- io: add `AsyncFd::{try_io, try_io_mut}` ([#6967])
+
+### Fixed
+
+- io: avoid `ptr->ref->ptr` roundtrip in RegistrationSet ([#6929])
+- runtime: do not defer `yield_now` inside `block_in_place` ([#6999])
+
+### Changes
+
+- io: simplify io readiness logic ([#6966])
+
+### Documented
+
+- net: fix docs for `tokio::net::unix::{pid_t, gid_t, uid_t}` ([#6791])
+- time: fix a typo in `Instant` docs ([#6982])
+
+[#6791]: https://github.com/tokio-rs/tokio/pull/6791
+[#6929]: https://github.com/tokio-rs/tokio/pull/6929
+[#6966]: https://github.com/tokio-rs/tokio/pull/6966
+[#6967]: https://github.com/tokio-rs/tokio/pull/6967
+[#6982]: https://github.com/tokio-rs/tokio/pull/6982
+[#6999]: https://github.com/tokio-rs/tokio/pull/6999
+
+# 1.41.1 (Nov 7th, 2024)
+
+### Fixed
+
+- metrics: fix bug with wrong number of buckets for the histogram ([#6957])
+- net: display `net` requirement for `net::UdpSocket` in docs ([#6938])
+- net: fix typo in `TcpStream` internal comment ([#6944])
+
+[#6957]: https://github.com/tokio-rs/tokio/pull/6957
+[#6938]: https://github.com/tokio-rs/tokio/pull/6938
+[#6944]: https://github.com/tokio-rs/tokio/pull/6944
+
+# 1.41.0 (Oct 22th, 2024)
+
+### Added
+
+- metrics: stabilize `global_queue_depth` ([#6854], [#6918])
+- net: add conversions for unix `SocketAddr` ([#6868])
+- sync: add `watch::Sender::sender_count` ([#6836])
+- sync: add `mpsc::Receiver::blocking_recv_many` ([#6867])
+- task: stabilize `Id` apis ([#6793], [#6891])
+
+### Added (unstable)
+
+- metrics: add H2 Histogram option to improve histogram granularity ([#6897])
+- metrics: rename some histogram apis ([#6924])
+- runtime: add `LocalRuntime` ([#6808])
+
+### Changed
+
+- runtime: box futures larger than 16k on release mode ([#6826])
+- sync: add `#[must_use]` to `Notified` ([#6828])
+- sync: make `watch` cooperative ([#6846])
+- sync: make `broadcast::Receiver` cooperative ([#6870])
+- task: add task size to tracing instrumentation ([#6881])
+- wasm: enable `cfg_fs` for `wasi` target ([#6822])
+
+### Fixed
+
+- net: fix regression of abstract socket path in unix socket ([#6838])
+
+### Documented
+
+- io: recommend `OwnedFd` with `AsyncFd` ([#6821])
+- io: document cancel safety of `AsyncFd` methods ([#6890])
+- macros: render more comprehensible documentation for `join` and `try_join` ([#6814], [#6841])
+- net: fix swapped examples for `TcpSocket::set_nodelay` and `TcpSocket::nodelay` ([#6840])
+- sync: document runtime compatibility ([#6833])
+
+[#6793]: https://github.com/tokio-rs/tokio/pull/6793
+[#6808]: https://github.com/tokio-rs/tokio/pull/6808
+[#6810]: https://github.com/tokio-rs/tokio/pull/6810
+[#6814]: https://github.com/tokio-rs/tokio/pull/6814
+[#6821]: https://github.com/tokio-rs/tokio/pull/6821
+[#6822]: https://github.com/tokio-rs/tokio/pull/6822
+[#6826]: https://github.com/tokio-rs/tokio/pull/6826
+[#6828]: https://github.com/tokio-rs/tokio/pull/6828
+[#6833]: https://github.com/tokio-rs/tokio/pull/6833
+[#6836]: https://github.com/tokio-rs/tokio/pull/6836
+[#6838]: https://github.com/tokio-rs/tokio/pull/6838
+[#6840]: https://github.com/tokio-rs/tokio/pull/6840
+[#6841]: https://github.com/tokio-rs/tokio/pull/6841
+[#6846]: https://github.com/tokio-rs/tokio/pull/6846
+[#6854]: https://github.com/tokio-rs/tokio/pull/6854
+[#6867]: https://github.com/tokio-rs/tokio/pull/6867
+[#6868]: https://github.com/tokio-rs/tokio/pull/6868
+[#6870]: https://github.com/tokio-rs/tokio/pull/6870
+[#6881]: https://github.com/tokio-rs/tokio/pull/6881
+[#6890]: https://github.com/tokio-rs/tokio/pull/6890
+[#6891]: https://github.com/tokio-rs/tokio/pull/6891
+[#6897]: https://github.com/tokio-rs/tokio/pull/6897
+[#6918]: https://github.com/tokio-rs/tokio/pull/6918
+[#6924]: https://github.com/tokio-rs/tokio/pull/6924
+
 # 1.40.0 (August 30th, 2024)
 
 ### Added

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.40.0"
+version = "1.42.0"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -96,12 +96,12 @@ mio = { version = "1.0.1", optional = true, default-features = false }
 parking_lot = { version = "0.12.0", optional = true }
 
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
-socket2 = { version = "0.5.5", optional = true, features = [ "all" ] }
+socket2 = { version = "0.5.5", optional = true, features = ["all"] }
 
 # Currently unstable. The API exposed by these features may be broken at any time.
 # Requires `--cfg tokio_unstable` to enable.
 [target.'cfg(tokio_unstable)'.dependencies]
-tracing = { version = "0.1.25", default-features = false, features = ["std"], optional = true } # Not in full
+tracing = { version = "0.1.29", default-features = false, features = ["std"], optional = true } # Not in full
 
 # Currently unstable. The API exposed by these features may be broken at any time.
 # Requires `--cfg tokio_unstable` to enable.
@@ -123,8 +123,8 @@ optional = true
 [target.'cfg(windows)'.dev-dependencies.windows-sys]
 version = "0.52"
 features = [
-    "Win32_Foundation",
-    "Win32_Security_Authorization",
+  "Win32_Foundation",
+  "Win32_Security_Authorization",
 ]
 
 [dev-dependencies]
@@ -137,6 +137,7 @@ async-stream = "0.3"
 [target.'cfg(not(target_family = "wasm"))'.dev-dependencies]
 socket2 = "0.5.5"
 tempfile = "3.1.0"
+proptest = "1"
 
 [target.'cfg(not(all(target_family = "wasm", target_os = "unknown")))'.dev-dependencies]
 rand = "0.8.0"
@@ -150,6 +151,9 @@ mio-aio = { version = "0.9.0", features = ["tokio"] }
 [target.'cfg(loom)'.dev-dependencies]
 loom = { version = "0.7", features = ["futures", "checkpoint"] }
 
+[target.'cfg(all(tokio_unstable, target_has_atomic = "64"))'.dev-dependencies]
+tracing-mock = "= 0.1.0-beta.1"
+
 [package.metadata.docs.rs]
 all-features = true
 # enable unstable features in the documentation
@@ -165,8 +169,7 @@ features = ["full", "test-util"]
 # The following are types that are allowed to be exposed in Tokio's public API.
 # The standard library is allowed by default.
 allowed_external_types = [
-   "bytes::buf::buf_impl::Buf",
-   "bytes::buf::buf_mut::BufMut",
-
-   "tokio_macros::*",
+  "bytes::buf::buf_impl::Buf",
+  "bytes::buf::buf_mut::BufMut",
+  "tokio_macros::*",
 ]

tokio/README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.40.0", features = ["full"] }
+tokio = { version = "1.42.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -216,7 +216,6 @@ 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)
 
@@ -238,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
 

tokio/src/fs/file.rs

@@ -936,7 +936,7 @@ cfg_windows! {
 
 impl Inner {
     async fn complete_inflight(&mut self) {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| self.poll_complete_inflight(cx)).await;
     }

tokio/src/fs/mocks.rs

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

tokio/src/fs/open_options.rs

@@ -394,6 +394,7 @@ impl OpenOptions {
     }
 
     /// Returns a mutable reference to the underlying `std::fs::OpenOptions`
+    #[cfg(any(windows, unix))]
     pub(super) fn as_inner_mut(&mut self) -> &mut StdOpenOptions {
         &mut self.0
     }

tokio/src/fs/read_dir.rs

@@ -76,7 +76,7 @@ impl ReadDir {
     ///
     /// This method is cancellation safe.
     pub async fn next_entry(&mut self) -> io::Result<Option<DirEntry>> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
         poll_fn(|cx| self.poll_next_entry(cx)).await
     }
 

tokio/src/future/mod.rs

@@ -5,10 +5,6 @@
 #[cfg(any(feature = "macros", feature = "process"))]
 pub(crate) mod maybe_done;
 
-mod poll_fn;
-#[allow(unused_imports)]
-pub use poll_fn::poll_fn;
-
 cfg_process! {
     mod try_join;
     pub(crate) use try_join::try_join3;

tokio/src/future/poll_fn.rs

@@ -1,60 +0,0 @@
-#![allow(dead_code)]
-
-//! Definition of the `PollFn` adapter combinator.
-
-use std::fmt;
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-// This struct is intentionally `!Unpin` when `F` is `!Unpin`. This is to
-// mitigate the issue where rust puts noalias on mutable references to the
-// `PollFn` type if it is `Unpin`. If the closure has ownership of a future,
-// then this "leaks" and the future is affected by noalias too, which we don't
-// want.
-//
-// See this thread for more information:
-// <https://internals.rust-lang.org/t/surprising-soundness-trouble-around-pollfn/17484>
-//
-// The fact that `PollFn` is not `Unpin` when it shouldn't be is tested in
-// `tests/async_send_sync.rs`.
-
-/// Future for the [`poll_fn`] function.
-pub struct PollFn<F> {
-    f: F,
-}
-
-/// Creates a new future wrapping around a function returning [`Poll`].
-pub fn poll_fn<T, F>(f: F) -> PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    PollFn { f }
-}
-
-impl<F> fmt::Debug for PollFn<F> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("PollFn").finish()
-    }
-}
-
-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/src/io/async_fd.rs

@@ -21,11 +21,13 @@ use std::task::{ready, Context, Poll};
 /// the [`AsyncFd`] is dropped.
 ///
 /// The [`AsyncFd`] takes ownership of an arbitrary object to represent the IO
-/// object. It is intended that this object will handle closing the file
+/// object. It is intended that the inner object will handle closing the file
 /// descriptor when it is dropped, avoiding resource leaks and ensuring that the
 /// [`AsyncFd`] can clean up the registration before closing the file descriptor.
 /// The [`AsyncFd::into_inner`] function can be used to extract the inner object
-/// to retake control from the tokio IO reactor.
+/// to retake control from the tokio IO reactor. The [`OwnedFd`] type is often
+/// used as the inner object, as it is the simplest type that closes the fd on
+/// drop.
 ///
 /// The inner object is required to implement [`AsRawFd`]. This file descriptor
 /// must not change while [`AsyncFd`] owns the inner object, i.e. the
@@ -175,6 +177,7 @@ use std::task::{ready, Context, Poll};
 /// [`TcpStream::poll_read_ready`]: struct@crate::net::TcpStream
 /// [`AsyncRead`]: trait@crate::io::AsyncRead
 /// [`AsyncWrite`]: trait@crate::io::AsyncWrite
+/// [`OwnedFd`]: struct@std::os::fd::OwnedFd
 pub struct AsyncFd<T: AsRawFd> {
     registration: Registration,
     // The inner value is always present. the Option is required for `drop` and `into_inner`.
@@ -699,6 +702,13 @@ impl<T: AsRawFd> AsyncFd<T> {
     /// concurrently with other methods on this struct. This method only
     /// provides shared access to the inner IO resource when handling the
     /// [`AsyncFdReadyGuard`].
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. Once a readiness event occurs, the method
+    /// will continue to return immediately until the readiness event is
+    /// consumed by an attempt to read or write that fails with `WouldBlock` or
+    /// `Poll::Pending`.
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn readable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {
         self.ready(Interest::READABLE).await
@@ -710,6 +720,13 @@ impl<T: AsRawFd> AsyncFd<T> {
     ///
     /// This method takes `&mut self`, so it is possible to access the inner IO
     /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. Once a readiness event occurs, the method
+    /// will continue to return immediately until the readiness event is
+    /// consumed by an attempt to read or write that fails with `WouldBlock` or
+    /// `Poll::Pending`.
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn readable_mut<'a>(&'a mut self) -> io::Result<AsyncFdReadyMutGuard<'a, T>> {
         self.ready_mut(Interest::READABLE).await
@@ -723,6 +740,13 @@ impl<T: AsRawFd> AsyncFd<T> {
     /// concurrently with other methods on this struct. This method only
     /// provides shared access to the inner IO resource when handling the
     /// [`AsyncFdReadyGuard`].
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. Once a readiness event occurs, the method
+    /// will continue to return immediately until the readiness event is
+    /// consumed by an attempt to read or write that fails with `WouldBlock` or
+    /// `Poll::Pending`.
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn writable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {
         self.ready(Interest::WRITABLE).await
@@ -734,6 +758,13 @@ impl<T: AsRawFd> AsyncFd<T> {
     ///
     /// This method takes `&mut self`, so it is possible to access the inner IO
     /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancel safe. Once a readiness event occurs, the method
+    /// will continue to return immediately until the readiness event is
+    /// consumed by an attempt to read or write that fails with `WouldBlock` or
+    /// `Poll::Pending`.
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn writable_mut<'a>(&'a mut self) -> io::Result<AsyncFdReadyMutGuard<'a, T>> {
         self.ready_mut(Interest::WRITABLE).await
@@ -841,6 +872,56 @@ impl<T: AsRawFd> AsyncFd<T> {
             .async_io(interest, || f(self.inner.as_mut().unwrap()))
             .await
     }
+
+    /// Tries to read or write from the file descriptor using a user-provided IO operation.
+    ///
+    /// If the file descriptor is ready, the provided closure is called. The closure
+    /// should attempt to perform IO operation on the file descriptor by manually
+    /// calling the appropriate syscall. If the operation fails because the
+    /// file descriptor is not actually ready, then the closure should return a
+    /// `WouldBlock` error and the readiness flag is cleared. The return value
+    /// of the closure is then returned by `try_io`.
+    ///
+    /// If the file descriptor is not ready, then the closure is not called
+    /// and a `WouldBlock` error is returned.
+    ///
+    /// The closure should only return a `WouldBlock` error if it has performed
+    /// an IO operation on the file descriptor that failed due to the file descriptor not being
+    /// ready. Returning a `WouldBlock` error in any other situation will
+    /// incorrectly clear the readiness flag, which can cause the file descriptor to
+    /// behave incorrectly.
+    ///
+    /// The closure should not perform the IO operation using any of the methods
+    /// defined on the Tokio `AsyncFd` type, as this will mess with the
+    /// readiness flag and can cause the file descriptor to behave incorrectly.
+    ///
+    /// This method is not intended to be used with combined interests.
+    /// The closure should perform only one type of IO operation, so it should not
+    /// require more than one ready state. This method may panic or sleep forever
+    /// if it is called with a combined interest.
+    pub fn try_io<R>(
+        &self,
+        interest: Interest,
+        f: impl FnOnce(&T) -> io::Result<R>,
+    ) -> io::Result<R> {
+        self.registration
+            .try_io(interest, || f(self.inner.as_ref().unwrap()))
+    }
+
+    /// Tries to read or write from the file descriptor using a user-provided IO operation.
+    ///
+    /// The behavior is the same as [`try_io`], except that the closure can mutate the inner
+    /// value of the [`AsyncFd`].
+    ///
+    /// [`try_io`]: AsyncFd::try_io
+    pub fn try_io_mut<R>(
+        &mut self,
+        interest: Interest,
+        f: impl FnOnce(&mut T) -> io::Result<R>,
+    ) -> io::Result<R> {
+        self.registration
+            .try_io(interest, || f(self.inner.as_mut().unwrap()))
+    }
 }
 
 impl<T: AsRawFd> AsRawFd for AsyncFd<T> {

tokio/src/io/mod.rs

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

tokio/src/io/util/copy_bidirectional.rs

@@ -1,8 +1,8 @@
 use super::copy::CopyBuffer;
 
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite};
 
+use std::future::poll_fn;
 use std::io;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};

tokio/src/io/util/lines.rs

@@ -67,7 +67,7 @@ where
     /// # }
     /// ```
     pub async fn next_line(&mut self) -> io::Result<Option<String>> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| Pin::new(&mut *self).poll_next_line(cx)).await
     }

tokio/src/io/util/split.rs

@@ -59,7 +59,7 @@ where
     /// # }
     /// ```
     pub async fn next_segment(&mut self) -> io::Result<Option<Vec<u8>>> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| Pin::new(&mut *self).poll_next_segment(cx)).await
     }

tokio/src/lib.rs

@@ -19,6 +19,7 @@
 #![cfg_attr(docsrs, feature(doc_cfg))]
 #![cfg_attr(docsrs, allow(unused_attributes))]
 #![cfg_attr(loom, allow(dead_code, unreachable_pub))]
+#![cfg_attr(windows, allow(rustdoc::broken_intra_doc_links))]
 
 //! A runtime for writing reliable network applications without compromising speed.
 //!
@@ -351,8 +352,10 @@
 //! - [`task::Builder`]
 //! - Some methods on [`task::JoinSet`]
 //! - [`runtime::RuntimeMetrics`]
+//! - [`runtime::Builder::on_task_spawn`]
+//! - [`runtime::Builder::on_task_terminate`]
 //! - [`runtime::Builder::unhandled_panic`]
-//! - [`task::Id`]
+//! - [`runtime::TaskMeta`]
 //!
 //! This flag enables **unstable** features. The public API of these features
 //! may break in 1.x releases. To enable these features, the `--cfg
@@ -631,15 +634,15 @@ pub mod stream {}
 // local re-exports of platform specific things, allowing for decent
 // documentation to be shimmed in on docs.rs
 
-#[cfg(docsrs)]
+#[cfg(all(docsrs, unix))]
 pub mod doc;
 
 #[cfg(any(feature = "net", feature = "fs"))]
-#[cfg(docsrs)]
+#[cfg(all(docsrs, unix))]
 #[allow(unused)]
 pub(crate) use self::doc::os;
 
-#[cfg(not(docsrs))]
+#[cfg(not(all(docsrs, unix)))]
 #[allow(unused)]
 pub(crate) use std::os;
 

tokio/src/loom/mocked.rs

@@ -2,7 +2,7 @@ pub(crate) use loom::*;
 
 pub(crate) mod sync {
 
-    pub(crate) use loom::sync::MutexGuard;
+    pub(crate) use loom::sync::{MutexGuard, RwLockReadGuard, RwLockWriteGuard};
 
     #[derive(Debug)]
     pub(crate) struct Mutex<T>(loom::sync::Mutex<T>);
@@ -30,6 +30,38 @@ pub(crate) mod sync {
             self.0.get_mut().unwrap()
         }
     }
+
+    #[derive(Debug)]
+    pub(crate) struct RwLock<T>(loom::sync::RwLock<T>);
+
+    #[allow(dead_code)]
+    impl<T> RwLock<T> {
+        #[inline]
+        pub(crate) fn new(t: T) -> Self {
+            Self(loom::sync::RwLock::new(t))
+        }
+
+        #[inline]
+        pub(crate) fn read(&self) -> RwLockReadGuard<'_, T> {
+            self.0.read().unwrap()
+        }
+
+        #[inline]
+        pub(crate) fn try_read(&self) -> Option<RwLockReadGuard<'_, T>> {
+            self.0.try_read().ok()
+        }
+
+        #[inline]
+        pub(crate) fn write(&self) -> RwLockWriteGuard<'_, T> {
+            self.0.write().unwrap()
+        }
+
+        #[inline]
+        pub(crate) fn try_write(&self) -> Option<RwLockWriteGuard<'_, T>> {
+            self.0.try_write().ok()
+        }
+    }
+
     pub(crate) use loom::sync::*;
 
     pub(crate) mod atomic {

tokio/src/loom/std/mod.rs

@@ -8,6 +8,7 @@ mod barrier;
 mod mutex;
 #[cfg(all(feature = "parking_lot", not(miri)))]
 mod parking_lot;
+mod rwlock;
 mod unsafe_cell;
 
 pub(crate) mod cell {
@@ -64,11 +65,14 @@ pub(crate) mod sync {
 
     #[cfg(not(all(feature = "parking_lot", not(miri))))]
     #[allow(unused_imports)]
-    pub(crate) use std::sync::{Condvar, MutexGuard, RwLock, RwLockReadGuard, WaitTimeoutResult};
+    pub(crate) use std::sync::{Condvar, MutexGuard, RwLockReadGuard, WaitTimeoutResult};
 
     #[cfg(not(all(feature = "parking_lot", not(miri))))]
     pub(crate) use crate::loom::std::mutex::Mutex;
 
+    #[cfg(not(all(feature = "parking_lot", not(miri))))]
+    pub(crate) use crate::loom::std::rwlock::RwLock;
+
     pub(crate) mod atomic {
         pub(crate) use crate::loom::std::atomic_u16::AtomicU16;
         pub(crate) use crate::loom::std::atomic_u32::AtomicU32;
@@ -91,22 +95,16 @@ pub(crate) mod sys {
         match std::env::var(ENV_WORKER_THREADS) {
             Ok(s) => {
                 let n = s.parse().unwrap_or_else(|e| {
-                    panic!(
-                        "\"{}\" must be usize, error: {}, value: {}",
-                        ENV_WORKER_THREADS, e, s
-                    )
+                    panic!("\"{ENV_WORKER_THREADS}\" must be usize, error: {e}, value: {s}")
                 });
-                assert!(n > 0, "\"{}\" cannot be set to 0", ENV_WORKER_THREADS);
+                assert!(n > 0, "\"{ENV_WORKER_THREADS}\" cannot be set to 0");
                 n
             }
             Err(std::env::VarError::NotPresent) => {
                 std::thread::available_parallelism().map_or(1, NonZeroUsize::get)
             }
             Err(std::env::VarError::NotUnicode(e)) => {
-                panic!(
-                    "\"{}\" must be valid unicode, error: {:?}",
-                    ENV_WORKER_THREADS, e
-                )
+                panic!("\"{ENV_WORKER_THREADS}\" must be valid unicode, error: {e:?}")
             }
         }
     }

tokio/src/loom/std/parking_lot.rs

@@ -96,12 +96,20 @@ impl<T> RwLock<T> {
         RwLock(PhantomData, parking_lot::RwLock::new(t))
     }
 
-    pub(crate) fn read(&self) -> LockResult<RwLockReadGuard<'_, T>> {
-        Ok(RwLockReadGuard(PhantomData, self.1.read()))
+    pub(crate) fn read(&self) -> RwLockReadGuard<'_, T> {
+        RwLockReadGuard(PhantomData, self.1.read())
     }
 
-    pub(crate) fn write(&self) -> LockResult<RwLockWriteGuard<'_, T>> {
-        Ok(RwLockWriteGuard(PhantomData, self.1.write()))
+    pub(crate) fn try_read(&self) -> Option<RwLockReadGuard<'_, T>> {
+        Some(RwLockReadGuard(PhantomData, self.1.read()))
+    }
+
+    pub(crate) fn write(&self) -> RwLockWriteGuard<'_, T> {
+        RwLockWriteGuard(PhantomData, self.1.write())
+    }
+
+    pub(crate) fn try_write(&self) -> Option<RwLockWriteGuard<'_, T>> {
+        Some(RwLockWriteGuard(PhantomData, self.1.write()))
     }
 }
 

tokio/src/loom/std/rwlock.rs

@@ -0,0 +1,48 @@
+use std::sync::{self, RwLockReadGuard, RwLockWriteGuard, TryLockError};
+
+/// Adapter for `std::sync::RwLock` that removes the poisoning aspects
+/// from its api.
+#[derive(Debug)]
+pub(crate) struct RwLock<T: ?Sized>(sync::RwLock<T>);
+
+#[allow(dead_code)]
+impl<T> RwLock<T> {
+    #[inline]
+    pub(crate) fn new(t: T) -> Self {
+        Self(sync::RwLock::new(t))
+    }
+
+    #[inline]
+    pub(crate) fn read(&self) -> RwLockReadGuard<'_, T> {
+        match self.0.read() {
+            Ok(guard) => guard,
+            Err(p_err) => p_err.into_inner(),
+        }
+    }
+
+    #[inline]
+    pub(crate) fn try_read(&self) -> Option<RwLockReadGuard<'_, T>> {
+        match self.0.try_read() {
+            Ok(guard) => Some(guard),
+            Err(TryLockError::Poisoned(p_err)) => Some(p_err.into_inner()),
+            Err(TryLockError::WouldBlock) => None,
+        }
+    }
+
+    #[inline]
+    pub(crate) fn write(&self) -> RwLockWriteGuard<'_, T> {
+        match self.0.write() {
+            Ok(guard) => guard,
+            Err(p_err) => p_err.into_inner(),
+        }
+    }
+
+    #[inline]
+    pub(crate) fn try_write(&self) -> Option<RwLockWriteGuard<'_, T>> {
+        match self.0.try_write() {
+            Ok(guard) => Some(guard),
+            Err(TryLockError::Poisoned(p_err)) => Some(p_err.into_inner()),
+            Err(TryLockError::WouldBlock) => None,
+        }
+    }
+}

tokio/src/macros/cfg.rs

@@ -97,7 +97,6 @@ macro_rules! cfg_fs {
     ($($item:item)*) => {
         $(
             #[cfg(feature = "fs")]
-            #[cfg(not(target_os = "wasi"))]
             #[cfg_attr(docsrs, doc(cfg(feature = "fs")))]
             $item
         )*

tokio/src/macros/join.rs

@@ -1,60 +1,72 @@
-/// Waits on multiple concurrent branches, returning when **all** branches
-/// complete.
-///
-/// The `join!` macro must be used inside of async functions, closures, and
-/// blocks.
-///
-/// The `join!` macro takes a list of async expressions and evaluates them
-/// concurrently on the same task. Each async expression evaluates to a future
-/// and the futures from each expression are multiplexed on the current task.
-///
-/// When working with async expressions returning `Result`, `join!` will wait
-/// for **all** branches complete regardless if any complete with `Err`. Use
-/// [`try_join!`] to return early when `Err` is encountered.
-///
-/// [`try_join!`]: crate::try_join
-///
-/// # Notes
-///
-/// The supplied futures are stored inline and do not require allocating a
-/// `Vec`.
-///
-/// ### Runtime characteristics
-///
-/// By running all async expressions on the current task, the expressions are
-/// able to run **concurrently** but not in **parallel**. This means all
-/// expressions are run on the same thread and if one branch blocks the thread,
-/// all other expressions will be unable to continue. If parallelism is
-/// required, spawn each async expression using [`tokio::spawn`] and pass the
-/// join handle to `join!`.
-///
-/// [`tokio::spawn`]: crate::spawn
-///
-/// # Examples
-///
-/// Basic join with two branches
-///
-/// ```
-/// async fn do_stuff_async() {
-///     // async work
-/// }
-///
-/// async fn more_async_work() {
-///     // more here
-/// }
-///
-/// #[tokio::main]
-/// async fn main() {
-///     let (first, second) = tokio::join!(
-///         do_stuff_async(),
-///         more_async_work());
-///
-///     // do something with the values
-/// }
-/// ```
-#[macro_export]
-#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
-macro_rules! join {
+macro_rules! doc {
+    ($join:item) => {
+        /// Waits on multiple concurrent branches, returning when **all** branches
+        /// complete.
+        ///
+        /// The `join!` macro must be used inside of async functions, closures, and
+        /// blocks.
+        ///
+        /// The `join!` macro takes a list of async expressions and evaluates them
+        /// concurrently on the same task. Each async expression evaluates to a future
+        /// and the futures from each expression are multiplexed on the current task.
+        ///
+        /// When working with async expressions returning `Result`, `join!` will wait
+        /// for **all** branches complete regardless if any complete with `Err`. Use
+        /// [`try_join!`] to return early when `Err` is encountered.
+        ///
+        /// [`try_join!`]: crate::try_join
+        ///
+        /// # Notes
+        ///
+        /// The supplied futures are stored inline and do not require allocating a
+        /// `Vec`.
+        ///
+        /// ### Runtime characteristics
+        ///
+        /// By running all async expressions on the current task, the expressions are
+        /// able to run **concurrently** but not in **parallel**. This means all
+        /// expressions are run on the same thread and if one branch blocks the thread,
+        /// all other expressions will be unable to continue. If parallelism is
+        /// required, spawn each async expression using [`tokio::spawn`] and pass the
+        /// join handle to `join!`.
+        ///
+        /// [`tokio::spawn`]: crate::spawn
+        ///
+        /// # Examples
+        ///
+        /// Basic join with two branches
+        ///
+        /// ```
+        /// async fn do_stuff_async() {
+        ///     // async work
+        /// }
+        ///
+        /// async fn more_async_work() {
+        ///     // more here
+        /// }
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let (first, second) = tokio::join!(
+        ///         do_stuff_async(),
+        ///         more_async_work());
+        ///
+        ///     // do something with the values
+        /// }
+        /// ```
+        #[macro_export]
+        #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
+        $join
+    };
+}
+
+#[cfg(doc)]
+doc! {macro_rules! join {
+    ($($future:expr),*) => { unimplemented!() }
+}}
+
+#[cfg(not(doc))]
+doc! {macro_rules! join {
     (@ {
         // One `_` for each branch in the `join!` macro. This is not used once
         // normalization is complete.
@@ -163,4 +175,4 @@ macro_rules! join {
     };
 
     () => { async {}.await }
-}
+}}

tokio/src/macros/support.rs

@@ -1,7 +1,8 @@
 cfg_macros! {
-    pub use crate::future::poll_fn;
     pub use crate::future::maybe_done::maybe_done;
 
+    pub use std::future::poll_fn;
+
     #[doc(hidden)]
     pub fn thread_rng_n(n: u32) -> u32 {
         crate::runtime::context::thread_rng_n(n)

tokio/src/macros/try_join.rs

@@ -1,106 +1,118 @@
-/// Waits on multiple concurrent branches, returning when **all** branches
-/// complete with `Ok(_)` or on the first `Err(_)`.
-///
-/// The `try_join!` macro must be used inside of async functions, closures, and
-/// blocks.
-///
-/// Similar to [`join!`], the `try_join!` macro takes a list of async
-/// expressions and evaluates them concurrently on the same task. Each async
-/// expression evaluates to a future and the futures from each expression are
-/// multiplexed on the current task. The `try_join!` macro returns when **all**
-/// branches return with `Ok` or when the **first** branch returns with `Err`.
-///
-/// [`join!`]: macro@join
-///
-/// # Notes
-///
-/// The supplied futures are stored inline and do not require allocating a
-/// `Vec`.
-///
-/// ### Runtime characteristics
-///
-/// By running all async expressions on the current task, the expressions are
-/// able to run **concurrently** but not in **parallel**. This means all
-/// expressions are run on the same thread and if one branch blocks the thread,
-/// all other expressions will be unable to continue. If parallelism is
-/// required, spawn each async expression using [`tokio::spawn`] and pass the
-/// join handle to `try_join!`.
-///
-/// [`tokio::spawn`]: crate::spawn
-///
-/// # Examples
-///
-/// Basic `try_join` with two branches.
-///
-/// ```
-/// async fn do_stuff_async() -> Result<(), &'static str> {
-///     // async work
-/// # Ok(())
-/// }
-///
-/// async fn more_async_work() -> Result<(), &'static str> {
-///     // more here
-/// # Ok(())
-/// }
-///
-/// #[tokio::main]
-/// async fn main() {
-///     let res = tokio::try_join!(
-///         do_stuff_async(),
-///         more_async_work());
-///
-///     match res {
-///          Ok((first, second)) => {
-///              // do something with the values
-///          }
-///          Err(err) => {
-///             println!("processing failed; error = {}", err);
-///          }
-///     }
-/// }
-/// ```
-///
-/// Using `try_join!` with spawned tasks.
-///
-/// ```
-/// use tokio::task::JoinHandle;
-///
-/// async fn do_stuff_async() -> Result<(), &'static str> {
-///     // async work
-/// # Err("failed")
-/// }
-///
-/// async fn more_async_work() -> Result<(), &'static str> {
-///     // more here
-/// # Ok(())
-/// }
-///
-/// async fn flatten<T>(handle: JoinHandle<Result<T, &'static str>>) -> Result<T, &'static str> {
-///     match handle.await {
-///         Ok(Ok(result)) => Ok(result),
-///         Ok(Err(err)) => Err(err),
-///         Err(err) => Err("handling failed"),
-///     }
-/// }
-///
-/// #[tokio::main]
-/// async fn main() {
-///     let handle1 = tokio::spawn(do_stuff_async());
-///     let handle2 = tokio::spawn(more_async_work());
-///     match tokio::try_join!(flatten(handle1), flatten(handle2)) {
-///         Ok(val) => {
-///             // do something with the values
-///         }
-///         Err(err) => {
-///             println!("Failed with {}.", err);
-///             # assert_eq!(err, "failed");
-///         }
-///     }
-/// }
-/// ```
-#[macro_export]
-#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
-macro_rules! try_join {
+macro_rules! doc {
+    ($try_join:item) => {
+        /// Waits on multiple concurrent branches, returning when **all** branches
+        /// complete with `Ok(_)` or on the first `Err(_)`.
+        ///
+        /// The `try_join!` macro must be used inside of async functions, closures, and
+        /// blocks.
+        ///
+        /// Similar to [`join!`], the `try_join!` macro takes a list of async
+        /// expressions and evaluates them concurrently on the same task. Each async
+        /// expression evaluates to a future and the futures from each expression are
+        /// multiplexed on the current task. The `try_join!` macro returns when **all**
+        /// branches return with `Ok` or when the **first** branch returns with `Err`.
+        ///
+        /// [`join!`]: macro@join
+        ///
+        /// # Notes
+        ///
+        /// The supplied futures are stored inline and do not require allocating a
+        /// `Vec`.
+        ///
+        /// ### Runtime characteristics
+        ///
+        /// By running all async expressions on the current task, the expressions are
+        /// able to run **concurrently** but not in **parallel**. This means all
+        /// expressions are run on the same thread and if one branch blocks the thread,
+        /// all other expressions will be unable to continue. If parallelism is
+        /// required, spawn each async expression using [`tokio::spawn`] and pass the
+        /// join handle to `try_join!`.
+        ///
+        /// [`tokio::spawn`]: crate::spawn
+        ///
+        /// # Examples
+        ///
+        /// Basic `try_join` with two branches.
+        ///
+        /// ```
+        /// async fn do_stuff_async() -> Result<(), &'static str> {
+        ///     // async work
+        /// # Ok(())
+        /// }
+        ///
+        /// async fn more_async_work() -> Result<(), &'static str> {
+        ///     // more here
+        /// # Ok(())
+        /// }
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let res = tokio::try_join!(
+        ///         do_stuff_async(),
+        ///         more_async_work());
+        ///
+        ///     match res {
+        ///          Ok((first, second)) => {
+        ///              // do something with the values
+        ///          }
+        ///          Err(err) => {
+        ///             println!("processing failed; error = {}", err);
+        ///          }
+        ///     }
+        /// }
+        /// ```
+        ///
+        /// Using `try_join!` with spawned tasks.
+        ///
+        /// ```
+        /// use tokio::task::JoinHandle;
+        ///
+        /// async fn do_stuff_async() -> Result<(), &'static str> {
+        ///     // async work
+        /// # Err("failed")
+        /// }
+        ///
+        /// async fn more_async_work() -> Result<(), &'static str> {
+        ///     // more here
+        /// # Ok(())
+        /// }
+        ///
+        /// async fn flatten<T>(handle: JoinHandle<Result<T, &'static str>>) -> Result<T, &'static str> {
+        ///     match handle.await {
+        ///         Ok(Ok(result)) => Ok(result),
+        ///         Ok(Err(err)) => Err(err),
+        ///         Err(err) => Err("handling failed"),
+        ///     }
+        /// }
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let handle1 = tokio::spawn(do_stuff_async());
+        ///     let handle2 = tokio::spawn(more_async_work());
+        ///     match tokio::try_join!(flatten(handle1), flatten(handle2)) {
+        ///         Ok(val) => {
+        ///             // do something with the values
+        ///         }
+        ///         Err(err) => {
+        ///             println!("Failed with {}.", err);
+        ///             # assert_eq!(err, "failed");
+        ///         }
+        ///     }
+        /// }
+        /// ```
+        #[macro_export]
+        #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
+        $try_join
+    };
+}
+
+#[cfg(doc)]
+doc! {macro_rules! try_join {
+    ($($future:expr),*) => { unimplemented!() }
+}}
+
+#[cfg(not(doc))]
+doc! {macro_rules! try_join {
     (@ {
         // One `_` for each branch in the `try_join!` macro. This is not used once
         // normalization is complete.
@@ -215,4 +227,4 @@ macro_rules! try_join {
     };
 
     () => { async { Ok(()) }.await }
-}
+}}

tokio/src/net/mod.rs

@@ -46,6 +46,7 @@ cfg_net! {
         pub use tcp::socket::TcpSocket;
 
         mod udp;
+        #[doc(inline)]
         pub use udp::UdpSocket;
     }
 }

tokio/src/net/tcp/socket.rs

@@ -63,7 +63,7 @@ cfg_net! {
     ///     // previous one.
     ///     //
     ///     // On Windows, this allows rebinding sockets which are actively in use,
-    ///     // which allows “socket hijacking”, so we explicitly don't set it here.
+    ///     // which allows "socket hijacking", so we explicitly don't set it here.
     ///     // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
     ///     socket.set_reuseaddr(true)?;
     ///     socket.bind(addr)?;
@@ -425,7 +425,7 @@ impl TcpSocket {
     /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
     /// let socket = TcpSocket::new_v4()?;
     ///
-    /// println!("{:?}", socket.nodelay()?);
+    /// socket.set_nodelay(true)?;
     /// # Ok(())
     /// # }
     /// ```
@@ -445,9 +445,9 @@ impl TcpSocket {
     /// use tokio::net::TcpSocket;
     ///
     /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
-    /// let stream = TcpSocket::new_v4()?;
+    /// let socket = TcpSocket::new_v4()?;
     ///
-    /// stream.set_nodelay(true)?;
+    /// println!("{:?}", socket.nodelay()?);
     /// # Ok(())
     /// # }
     /// ```
@@ -787,6 +787,9 @@ impl fmt::Debug for TcpSocket {
     }
 }
 
+// These trait implementations can't be build on Windows, so we completely
+// ignore them, even when building documentation.
+#[cfg(unix)]
 cfg_unix! {
     impl AsRawFd for TcpSocket {
         fn as_raw_fd(&self) -> RawFd {

tokio/src/net/tcp/split.rs

@@ -8,10 +8,10 @@
 //! split has no associated overhead and enforces all invariants at the type
 //! level.
 
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite, Interest, ReadBuf, Ready};
 use crate::net::TcpStream;
 
+use std::future::poll_fn;
 use std::io;
 use std::net::{Shutdown, SocketAddr};
 use std::pin::Pin;
@@ -69,7 +69,7 @@ impl ReadHalf<'_> {
     /// use tokio::io::{self, ReadBuf};
     /// use tokio::net::TcpStream;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {

tokio/src/net/tcp/split_owned.rs

@@ -8,11 +8,11 @@
 //! split has no associated overhead and enforces all invariants at the type
 //! level.
 
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite, Interest, ReadBuf, Ready};
 use crate::net::TcpStream;
 
 use std::error::Error;
+use std::future::poll_fn;
 use std::net::{Shutdown, SocketAddr};
 use std::pin::Pin;
 use std::sync::Arc;
@@ -124,7 +124,7 @@ impl OwnedReadHalf {
     /// use tokio::io::{self, ReadBuf};
     /// use tokio::net::TcpStream;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {

tokio/src/net/tcp/stream.rs

@@ -1,6 +1,6 @@
 cfg_not_wasi! {
-    use crate::future::poll_fn;
     use crate::net::{to_socket_addrs, ToSocketAddrs};
+    use std::future::poll_fn;
     use std::time::Duration;
 }
 
@@ -340,7 +340,7 @@ impl TcpStream {
     /// use tokio::io::{self, ReadBuf};
     /// use tokio::net::TcpStream;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {
@@ -1278,7 +1278,7 @@ impl TcpStream {
 
     // == Poll IO functions that takes `&self` ==
     //
-    // To read or write without mutable access to the `UnixStream`, combine the
+    // To read or write without mutable access to the `TcpStream`, combine the
     // `poll_read_ready` or `poll_write_ready` methods with the `try_read` or
     // `try_write` methods.
 

tokio/src/net/unix/listener.rs

@@ -81,7 +81,7 @@ impl UnixListener {
         let addr = {
             let os_str_bytes = path.as_ref().as_os_str().as_bytes();
             if os_str_bytes.starts_with(b"\0") {
-                StdSocketAddr::from_abstract_name(os_str_bytes)?
+                StdSocketAddr::from_abstract_name(&os_str_bytes[1..])?
             } else {
                 StdSocketAddr::from_pathname(path)?
             }

tokio/src/net/unix/mod.rs

@@ -26,14 +26,14 @@ pub use ucred::UCred;
 
 pub mod pipe;
 
-/// A type representing process and process group IDs.
+/// A type representing user ID.
 #[allow(non_camel_case_types)]
 pub type uid_t = u32;
 
-/// A type representing user ID.
+/// A type representing group ID.
 #[allow(non_camel_case_types)]
 pub type gid_t = u32;
 
-/// A type representing group ID.
+/// A type representing process and process group IDs.
 #[allow(non_camel_case_types)]
 pub type pid_t = i32;

tokio/src/net/unix/socketaddr.rs

@@ -2,6 +2,10 @@ use std::fmt;
 use std::path::Path;
 
 /// An address associated with a Tokio Unix socket.
+///
+/// This type is a thin wrapper around [`std::os::unix::net::SocketAddr`]. You
+/// can convert to and from the standard library `SocketAddr` type using the
+/// [`From`] trait.
 pub struct SocketAddr(pub(super) std::os::unix::net::SocketAddr);
 
 impl SocketAddr {
@@ -29,3 +33,15 @@ impl fmt::Debug for SocketAddr {
         self.0.fmt(fmt)
     }
 }
+
+impl From<std::os::unix::net::SocketAddr> for SocketAddr {
+    fn from(value: std::os::unix::net::SocketAddr) -> Self {
+        SocketAddr(value)
+    }
+}
+
+impl From<SocketAddr> for std::os::unix::net::SocketAddr {
+    fn from(value: SocketAddr) -> Self {
+        value.0
+    }
+}

tokio/src/net/unix/stream.rs

@@ -1,4 +1,3 @@
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite, Interest, PollEvented, ReadBuf, Ready};
 use crate::net::unix::split::{split, ReadHalf, WriteHalf};
 use crate::net::unix::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
@@ -6,6 +5,7 @@ use crate::net::unix::ucred::{self, UCred};
 use crate::net::unix::SocketAddr;
 
 use std::fmt;
+use std::future::poll_fn;
 use std::io::{self, Read, Write};
 use std::net::Shutdown;
 #[cfg(target_os = "android")]
@@ -77,7 +77,7 @@ impl UnixStream {
         let addr = {
             let os_str_bytes = path.as_ref().as_os_str().as_bytes();
             if os_str_bytes.starts_with(b"\0") {
-                StdSocketAddr::from_abstract_name(os_str_bytes)?
+                StdSocketAddr::from_abstract_name(&os_str_bytes[1..])?
             } else {
                 StdSocketAddr::from_pathname(path)?
             }
@@ -142,7 +142,7 @@ impl UnixStream {
     ///             // if the readiness event is a false positive.
     ///             match stream.try_read(&mut data) {
     ///                 Ok(n) => {
-    ///                     println!("read {} bytes", n);        
+    ///                     println!("read {} bytes", n);
     ///                 }
     ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
     ///                     continue;

tokio/src/net/windows/named_pipe.rs

@@ -17,7 +17,7 @@ cfg_io_util! {
 }
 
 // Hide imports which are not used when generating documentation.
-#[cfg(not(docsrs))]
+#[cfg(windows)]
 mod doc {
     pub(super) use crate::os::windows::ffi::OsStrExt;
     pub(super) mod windows_sys {
@@ -30,7 +30,7 @@ mod doc {
 }
 
 // NB: none of these shows up in public API, so don't document them.
-#[cfg(docsrs)]
+#[cfg(not(windows))]
 mod doc {
     pub(super) mod mio_windows {
         pub type NamedPipe = crate::doc::NotDefinedHere;

tokio/src/runtime/blocking/pool.rs

@@ -8,6 +8,7 @@ use crate::runtime::builder::ThreadNameFn;
 use crate::runtime::task::{self, JoinHandle};
 use crate::runtime::{Builder, Callback, Handle, BOX_FUTURE_THRESHOLD};
 use crate::util::metric_atomics::MetricAtomicUsize;
+use crate::util::trace::{blocking_task, SpawnMeta};
 
 use std::collections::{HashMap, VecDeque};
 use std::fmt;
@@ -299,19 +300,29 @@ impl Spawner {
         F: FnOnce() -> R + Send + 'static,
         R: Send + 'static,
     {
-        let (join_handle, spawn_result) =
-            if cfg!(debug_assertions) && std::mem::size_of::<F>() > BOX_FUTURE_THRESHOLD {
-                self.spawn_blocking_inner(Box::new(func), Mandatory::NonMandatory, None, rt)
-            } else {
-                self.spawn_blocking_inner(func, Mandatory::NonMandatory, None, rt)
-            };
+        let fn_size = std::mem::size_of::<F>();
+        let (join_handle, spawn_result) = if fn_size > BOX_FUTURE_THRESHOLD {
+            self.spawn_blocking_inner(
+                Box::new(func),
+                Mandatory::NonMandatory,
+                SpawnMeta::new_unnamed(fn_size),
+                rt,
+            )
+        } else {
+            self.spawn_blocking_inner(
+                func,
+                Mandatory::NonMandatory,
+                SpawnMeta::new_unnamed(fn_size),
+                rt,
+            )
+        };
 
         match spawn_result {
             Ok(()) => join_handle,
             // Compat: do not panic here, return the join_handle even though it will never resolve
             Err(SpawnError::ShuttingDown) => join_handle,
             Err(SpawnError::NoThreads(e)) => {
-                panic!("OS can't spawn worker thread: {}", e)
+                panic!("OS can't spawn worker thread: {e}")
             }
         }
     }
@@ -327,18 +338,19 @@ impl Spawner {
             F: FnOnce() -> R + Send + 'static,
             R: Send + 'static,
         {
-            let (join_handle, spawn_result) = if cfg!(debug_assertions) && std::mem::size_of::<F>() > BOX_FUTURE_THRESHOLD {
+            let fn_size = std::mem::size_of::<F>();
+            let (join_handle, spawn_result) = if fn_size > BOX_FUTURE_THRESHOLD {
                 self.spawn_blocking_inner(
                     Box::new(func),
                     Mandatory::Mandatory,
-                    None,
+                    SpawnMeta::new_unnamed(fn_size),
                     rt,
                 )
             } else {
                 self.spawn_blocking_inner(
                     func,
                     Mandatory::Mandatory,
-                    None,
+                    SpawnMeta::new_unnamed(fn_size),
                     rt,
                 )
             };
@@ -356,35 +368,16 @@ impl Spawner {
         &self,
         func: F,
         is_mandatory: Mandatory,
-        name: Option<&str>,
+        spawn_meta: SpawnMeta<'_>,
         rt: &Handle,
     ) -> (JoinHandle<R>, Result<(), SpawnError>)
     where
         F: FnOnce() -> R + Send + 'static,
         R: Send + 'static,
     {
-        let fut = BlockingTask::new(func);
         let id = task::Id::next();
-        #[cfg(all(tokio_unstable, feature = "tracing"))]
-        let fut = {
-            use tracing::Instrument;
-            let location = std::panic::Location::caller();
-            let span = tracing::trace_span!(
-                target: "tokio::task::blocking",
-                "runtime.spawn",
-                kind = %"blocking",
-                task.name = %name.unwrap_or_default(),
-                task.id = id.as_u64(),
-                "fn" = %std::any::type_name::<F>(),
-                loc.file = location.file(),
-                loc.line = location.line(),
-                loc.col = location.column(),
-            );
-            fut.instrument(span)
-        };
-
-        #[cfg(not(all(tokio_unstable, feature = "tracing")))]
-        let _ = name;
+        let fut =
+            blocking_task::<F, BlockingTask<F>>(BlockingTask::new(func), spawn_meta, id.as_u64());
 
         let (task, handle) = task::unowned(fut, BlockingSchedule::new(rt), id);
 

tokio/src/runtime/builder.rs

@@ -1,13 +1,16 @@
 #![cfg_attr(loom, allow(unused_imports))]
 
 use crate::runtime::handle::Handle;
-#[cfg(tokio_unstable)]
-use crate::runtime::TaskMeta;
 use crate::runtime::{blocking, driver, Callback, HistogramBuilder, Runtime, TaskCallback};
+#[cfg(tokio_unstable)]
+use crate::runtime::{metrics::HistogramConfiguration, LocalOptions, LocalRuntime, TaskMeta};
 use crate::util::rand::{RngSeed, RngSeedGenerator};
 
+use crate::runtime::blocking::BlockingPool;
+use crate::runtime::scheduler::CurrentThread;
 use std::fmt;
 use std::io;
+use std::thread::ThreadId;
 use std::time::Duration;
 
 /// Builds Tokio Runtime with custom configuration values.
@@ -703,6 +706,12 @@ impl Builder {
     ///
     /// This *does not* support [`LocalSet`](crate::task::LocalSet) at this time.
     ///
+    /// **Note**: This is an [unstable API][unstable]. The public API of this type
+    /// may break in 1.x releases. See [the documentation on unstable
+    /// features][unstable] for details.
+    ///
+    /// [unstable]: crate#unstable-features
+    ///
     /// # Examples
     ///
     /// ```
@@ -725,6 +734,7 @@ impl Builder {
     /// # }
     /// ```
     #[cfg(all(not(loom), tokio_unstable))]
+    #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
     pub fn on_task_spawn<F>(&mut self, f: F) -> &mut Self
     where
         F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
@@ -745,6 +755,12 @@ impl Builder {
     ///
     /// This *does not* support [`LocalSet`](crate::task::LocalSet) at this time.
     ///
+    /// **Note**: This is an [unstable API][unstable]. The public API of this type
+    /// may break in 1.x releases. See [the documentation on unstable
+    /// features][unstable] for details.
+    ///
+    /// [unstable]: crate#unstable-features
+    ///
     /// # Examples
     ///
     /// ```
@@ -767,6 +783,7 @@ impl Builder {
     /// # }
     /// ```
     #[cfg(all(not(loom), tokio_unstable))]
+    #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
     pub fn on_task_terminate<F>(&mut self, f: F) -> &mut Self
     where
         F: Fn(&TaskMeta<'_>) + Send + Sync + 'static,
@@ -800,6 +817,37 @@ impl Builder {
         }
     }
 
+    /// Creates the configured `LocalRuntime`.
+    ///
+    /// The returned `LocalRuntime` instance is ready to spawn tasks.
+    ///
+    /// # Panics
+    /// This will panic if `current_thread` is not the selected runtime flavor.
+    /// All other runtime flavors are unsupported by [`LocalRuntime`].
+    ///
+    /// [`LocalRuntime`]: [crate::runtime::LocalRuntime]
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::runtime::Builder;
+    ///
+    /// let rt  = Builder::new_current_thread().build_local(&mut Default::default()).unwrap();
+    ///
+    /// rt.block_on(async {
+    ///     println!("Hello from the Tokio runtime");
+    /// });
+    /// ```
+    #[allow(unused_variables, unreachable_patterns)]
+    #[cfg(tokio_unstable)]
+    #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
+    pub fn build_local(&mut self, options: &LocalOptions) -> io::Result<LocalRuntime> {
+        match &self.kind {
+            Kind::CurrentThread => self.build_current_thread_local_runtime(),
+            _ => panic!("Only current_thread is supported when building a local runtime"),
+        }
+    }
+
     fn get_cfg(&self, workers: usize) -> driver::Cfg {
         driver::Cfg {
             enable_pause_time: match self.kind {
@@ -1065,33 +1113,49 @@ impl Builder {
         ///
         /// The histogram uses fixed bucket sizes. In other words, the histogram
         /// buckets are not dynamic based on input values. Use the
-        /// `metrics_poll_count_histogram_` builder methods to configure the
+        /// `metrics_poll_time_histogram` builder methods to configure the
         /// histogram details.
         ///
+        /// By default, a linear histogram with 10 buckets each 100 microseconds wide will be used.
+        /// This has an extremely low memory footprint, but may not provide enough granularity. For
+        /// better granularity with low memory usage, use [`metrics_poll_time_histogram_configuration()`]
+        /// to select [`LogHistogram`] instead.
+        ///
         /// # Examples
         ///
         /// ```
         /// use tokio::runtime;
         ///
         /// let rt = runtime::Builder::new_multi_thread()
-        ///     .enable_metrics_poll_count_histogram()
+        ///     .enable_metrics_poll_time_histogram()
         ///     .build()
         ///     .unwrap();
         /// # // Test default values here
         /// # fn us(n: u64) -> std::time::Duration { std::time::Duration::from_micros(n) }
         /// # let m = rt.handle().metrics();
-        /// # assert_eq!(m.poll_count_histogram_num_buckets(), 10);
-        /// # assert_eq!(m.poll_count_histogram_bucket_range(0), us(0)..us(100));
-        /// # assert_eq!(m.poll_count_histogram_bucket_range(1), us(100)..us(200));
+        /// # assert_eq!(m.poll_time_histogram_num_buckets(), 10);
+        /// # assert_eq!(m.poll_time_histogram_bucket_range(0), us(0)..us(100));
+        /// # assert_eq!(m.poll_time_histogram_bucket_range(1), us(100)..us(200));
         /// ```
         ///
         /// [`Handle::metrics()`]: crate::runtime::Handle::metrics
         /// [`Instant::now()`]: std::time::Instant::now
-        pub fn enable_metrics_poll_count_histogram(&mut self) -> &mut Self {
+        /// [`LogHistogram`]: crate::runtime::LogHistogram
+        /// [`metrics_poll_time_histogram_configuration()`]: Builder::metrics_poll_time_histogram_configuration
+        pub fn enable_metrics_poll_time_histogram(&mut self) -> &mut Self {
             self.metrics_poll_count_histogram_enable = true;
             self
         }
 
+        /// Deprecated. Use [`enable_metrics_poll_time_histogram()`] instead.
+        ///
+        /// [`enable_metrics_poll_time_histogram()`]: Builder::enable_metrics_poll_time_histogram
+        #[deprecated(note = "`poll_count_histogram` related methods have been renamed `poll_time_histogram` to better reflect their functionality.")]
+        #[doc(hidden)]
+        pub fn enable_metrics_poll_count_histogram(&mut self) -> &mut Self {
+            self.enable_metrics_poll_time_histogram()
+        }
+
         /// Sets the histogram scale for tracking the distribution of task poll
         /// times.
         ///
@@ -1108,14 +1172,83 @@ impl Builder {
         /// ```
         /// use tokio::runtime::{self, HistogramScale};
         ///
+        /// # #[allow(deprecated)]
         /// let rt = runtime::Builder::new_multi_thread()
-        ///     .enable_metrics_poll_count_histogram()
+        ///     .enable_metrics_poll_time_histogram()
         ///     .metrics_poll_count_histogram_scale(HistogramScale::Log)
         ///     .build()
         ///     .unwrap();
         /// ```
+        #[deprecated(note = "use `metrics_poll_time_histogram_configuration`")]
         pub fn metrics_poll_count_histogram_scale(&mut self, histogram_scale: crate::runtime::HistogramScale) -> &mut Self {
-            self.metrics_poll_count_histogram.scale = histogram_scale;
+            self.metrics_poll_count_histogram.legacy_mut(|b|b.scale = histogram_scale);
+            self
+        }
+
+        /// Configure the histogram for tracking poll times
+        ///
+        /// By default, a linear histogram with 10 buckets each 100 microseconds wide will be used.
+        /// This has an extremely low memory footprint, but may not provide enough granularity. For
+        /// better granularity with low memory usage, use [`LogHistogram`] instead.
+        ///
+        /// # Examples
+        /// Configure a [`LogHistogram`] with [default configuration]:
+        /// ```
+        /// use tokio::runtime;
+        /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_time_histogram()
+        ///     .metrics_poll_time_histogram_configuration(
+        ///         HistogramConfiguration::log(LogHistogram::default())
+        ///     )
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        ///
+        /// Configure a linear histogram with 100 buckets, each 10μs wide
+        /// ```
+        /// use tokio::runtime;
+        /// use std::time::Duration;
+        /// use tokio::runtime::HistogramConfiguration;
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_time_histogram()
+        ///     .metrics_poll_time_histogram_configuration(
+        ///         HistogramConfiguration::linear(Duration::from_micros(10), 100)
+        ///     )
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        ///
+        /// Configure a [`LogHistogram`] with the following settings:
+        /// - Measure times from 100ns to 120s
+        /// - Max error of 0.1
+        /// - No more than 1024 buckets
+        /// ```
+        /// use std::time::Duration;
+        /// use tokio::runtime;
+        /// use tokio::runtime::{HistogramConfiguration, LogHistogram};
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_time_histogram()
+        ///     .metrics_poll_time_histogram_configuration(
+        ///         HistogramConfiguration::log(LogHistogram::builder()
+        ///             .max_value(Duration::from_secs(120))
+        ///             .min_value(Duration::from_nanos(100))
+        ///             .max_error(0.1)
+        ///             .max_buckets(1024)
+        ///             .expect("configuration uses 488 buckets")
+        ///         )
+        ///     )
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        ///
+        /// [`LogHistogram`]: crate::runtime::LogHistogram
+        /// [default configuration]: crate::runtime::LogHistogramBuilder
+        pub fn metrics_poll_time_histogram_configuration(&mut self, configuration: HistogramConfiguration) -> &mut Self {
+            self.metrics_poll_count_histogram.histogram_type = configuration.inner;
             self
         }
 
@@ -1139,19 +1272,22 @@ impl Builder {
         /// use tokio::runtime;
         /// use std::time::Duration;
         ///
+        /// # #[allow(deprecated)]
         /// let rt = runtime::Builder::new_multi_thread()
-        ///     .enable_metrics_poll_count_histogram()
+        ///     .enable_metrics_poll_time_histogram()
         ///     .metrics_poll_count_histogram_resolution(Duration::from_micros(100))
         ///     .build()
         ///     .unwrap();
         /// ```
+        #[deprecated(note = "use `metrics_poll_time_histogram_configuration`")]
         pub fn metrics_poll_count_histogram_resolution(&mut self, resolution: Duration) -> &mut Self {
             assert!(resolution > Duration::from_secs(0));
             // Sanity check the argument and also make the cast below safe.
             assert!(resolution <= Duration::from_secs(1));
 
             let resolution = resolution.as_nanos() as u64;
-            self.metrics_poll_count_histogram.resolution = resolution;
+
+            self.metrics_poll_count_histogram.legacy_mut(|b|b.resolution = resolution);
             self
         }
 
@@ -1170,14 +1306,16 @@ impl Builder {
         /// ```
         /// use tokio::runtime;
         ///
+        /// # #[allow(deprecated)]
         /// let rt = runtime::Builder::new_multi_thread()
-        ///     .enable_metrics_poll_count_histogram()
+        ///     .enable_metrics_poll_time_histogram()
         ///     .metrics_poll_count_histogram_buckets(15)
         ///     .build()
         ///     .unwrap();
         /// ```
+        #[deprecated(note = "use `metrics_poll_time_histogram_configuration`")]
         pub fn metrics_poll_count_histogram_buckets(&mut self, buckets: usize) -> &mut Self {
-            self.metrics_poll_count_histogram.num_buckets = buckets;
+            self.metrics_poll_count_histogram.legacy_mut(|b|b.num_buckets = buckets);
             self
         }
     }
@@ -1191,8 +1329,40 @@ impl Builder {
     }
 
     fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
-        use crate::runtime::scheduler::{self, CurrentThread};
-        use crate::runtime::{runtime::Scheduler, Config};
+        use crate::runtime::runtime::Scheduler;
+
+        let (scheduler, handle, blocking_pool) =
+            self.build_current_thread_runtime_components(None)?;
+
+        Ok(Runtime::from_parts(
+            Scheduler::CurrentThread(scheduler),
+            handle,
+            blocking_pool,
+        ))
+    }
+
+    #[cfg(tokio_unstable)]
+    fn build_current_thread_local_runtime(&mut self) -> io::Result<LocalRuntime> {
+        use crate::runtime::local_runtime::LocalRuntimeScheduler;
+
+        let tid = std::thread::current().id();
+
+        let (scheduler, handle, blocking_pool) =
+            self.build_current_thread_runtime_components(Some(tid))?;
+
+        Ok(LocalRuntime::from_parts(
+            LocalRuntimeScheduler::CurrentThread(scheduler),
+            handle,
+            blocking_pool,
+        ))
+    }
+
+    fn build_current_thread_runtime_components(
+        &mut self,
+        local_tid: Option<ThreadId>,
+    ) -> io::Result<(CurrentThread, Handle, BlockingPool)> {
+        use crate::runtime::scheduler;
+        use crate::runtime::Config;
 
         let (driver, driver_handle) = driver::Driver::new(self.get_cfg(1))?;
 
@@ -1227,17 +1397,14 @@ impl Builder {
                 seed_generator: seed_generator_1,
                 metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(),
             },
+            local_tid,
         );
 
         let handle = Handle {
             inner: scheduler::Handle::CurrentThread(handle),
         };
 
-        Ok(Runtime::from_parts(
-            Scheduler::CurrentThread(scheduler),
-            handle,
-            blocking_pool,
-        ))
+        Ok((scheduler, handle, blocking_pool))
     }
 
     fn metrics_poll_count_histogram_builder(&self) -> Option<HistogramBuilder> {

tokio/src/runtime/context.rs

@@ -183,7 +183,14 @@ cfg_rt! {
     #[track_caller]
     pub(super) fn with_scheduler<R>(f: impl FnOnce(Option<&scheduler::Context>) -> R) -> R {
         let mut f = Some(f);
-        CONTEXT.try_with(|c| c.scheduler.with(f.take().unwrap()))
+        CONTEXT.try_with(|c| {
+            let f = f.take().unwrap();
+            if matches!(c.runtime.get(), EnterRuntime::Entered { .. }) {
+                c.scheduler.with(f)
+            } else {
+                f(None)
+            }
+        })
             .unwrap_or_else(|_| (f.take().unwrap())(None))
     }
 

tokio/src/runtime/coop.rs

@@ -135,8 +135,11 @@ cfg_rt! {
 }
 
 cfg_coop! {
+    use pin_project_lite::pin_project;
     use std::cell::Cell;
-    use std::task::{Context, Poll};
+    use std::future::Future;
+    use std::pin::Pin;
+    use std::task::{ready, Context, Poll};
 
     #[must_use]
     pub(crate) struct RestoreOnPending(Cell<Budget>);
@@ -240,6 +243,44 @@ cfg_coop! {
             self.0.is_none()
         }
     }
+
+    pin_project! {
+        /// Future wrapper to ensure cooperative scheduling.
+        ///
+        /// When being polled `poll_proceed` is called before the inner future is polled to check
+        /// if the inner future has exceeded its budget. If the inner future resolves, this will
+        /// automatically call `RestoreOnPending::made_progress` before resolving this future with
+        /// the result of the inner one. If polling the inner future is pending, polling this future
+        /// type will also return a `Poll::Pending`.
+        #[must_use = "futures do nothing unless polled"]
+        pub(crate) struct Coop<F: Future> {
+            #[pin]
+            pub(crate) fut: F,
+        }
+    }
+
+    impl<F: Future> Future for Coop<F> {
+        type Output = F::Output;
+
+        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            let coop = ready!(poll_proceed(cx));
+            let me = self.project();
+            if let Poll::Ready(ret) = me.fut.poll(cx) {
+                coop.made_progress();
+                Poll::Ready(ret)
+            } else {
+                Poll::Pending
+            }
+        }
+    }
+
+    /// Run a future with a budget constraint for cooperative scheduling.
+    /// If the future exceeds its budget while being polled, control is yielded back to the
+    /// runtime.
+    #[inline]
+    pub(crate) fn cooperative<F: Future>(fut: F) -> Coop<F> {
+        Coop { fut }
+    }
 }
 
 #[cfg(all(test, not(loom)))]
@@ -255,7 +296,7 @@ mod test {
 
     #[test]
     fn budgeting() {
-        use futures::future::poll_fn;
+        use std::future::poll_fn;
         use tokio_test::*;
 
         assert!(get().0.is_none());

tokio/src/runtime/handle.rs

@@ -18,10 +18,11 @@ pub struct Handle {
 use crate::runtime::task::JoinHandle;
 use crate::runtime::BOX_FUTURE_THRESHOLD;
 use crate::util::error::{CONTEXT_MISSING_ERROR, THREAD_LOCAL_DESTROYED_ERROR};
+use crate::util::trace::SpawnMeta;
 
 use std::future::Future;
 use std::marker::PhantomData;
-use std::{error, fmt};
+use std::{error, fmt, mem};
 
 /// Runtime context guard.
 ///
@@ -189,10 +190,11 @@ impl Handle {
         F: Future + Send + 'static,
         F::Output: Send + 'static,
     {
-        if cfg!(debug_assertions) && std::mem::size_of::<F>() > BOX_FUTURE_THRESHOLD {
-            self.spawn_named(Box::pin(future), None)
+        let fut_size = mem::size_of::<F>();
+        if fut_size > BOX_FUTURE_THRESHOLD {
+            self.spawn_named(Box::pin(future), SpawnMeta::new_unnamed(fut_size))
         } else {
-            self.spawn_named(future, None)
+            self.spawn_named(future, SpawnMeta::new_unnamed(fut_size))
         }
     }
 
@@ -248,8 +250,8 @@ impl Handle {
     /// # Panics
     ///
     /// This function panics if the provided future panics, if called within an
-    /// asynchronous execution context, or if a timer future is executed on a
-    /// runtime that has been shut down.
+    /// asynchronous execution context, or if a timer future is executed on a runtime that has been
+    /// shut down.
     ///
     /// # Examples
     ///
@@ -296,15 +298,16 @@ impl Handle {
     /// [`tokio::time`]: crate::time
     #[track_caller]
     pub fn block_on<F: Future>(&self, future: F) -> F::Output {
-        if cfg!(debug_assertions) && std::mem::size_of::<F>() > BOX_FUTURE_THRESHOLD {
-            self.block_on_inner(Box::pin(future))
+        let fut_size = mem::size_of::<F>();
+        if fut_size > BOX_FUTURE_THRESHOLD {
+            self.block_on_inner(Box::pin(future), SpawnMeta::new_unnamed(fut_size))
         } else {
-            self.block_on_inner(future)
+            self.block_on_inner(future, SpawnMeta::new_unnamed(fut_size))
         }
     }
 
     #[track_caller]
-    fn block_on_inner<F: Future>(&self, future: F) -> F::Output {
+    fn block_on_inner<F: Future>(&self, future: F, _meta: SpawnMeta<'_>) -> F::Output {
         #[cfg(all(
             tokio_unstable,
             tokio_taskdump,
@@ -316,7 +319,7 @@ impl Handle {
 
         #[cfg(all(tokio_unstable, feature = "tracing"))]
         let future =
-            crate::util::trace::task(future, "block_on", None, super::task::Id::next().as_u64());
+            crate::util::trace::task(future, "block_on", _meta, super::task::Id::next().as_u64());
 
         // Enter the runtime context. This sets the current driver handles and
         // prevents blocking an existing runtime.
@@ -326,7 +329,7 @@ impl Handle {
     }
 
     #[track_caller]
-    pub(crate) fn spawn_named<F>(&self, future: F, _name: Option<&str>) -> JoinHandle<F::Output>
+    pub(crate) fn spawn_named<F>(&self, future: F, _meta: SpawnMeta<'_>) -> JoinHandle<F::Output>
     where
         F: Future + Send + 'static,
         F::Output: Send + 'static,
@@ -341,10 +344,35 @@ impl Handle {
         ))]
         let future = super::task::trace::Trace::root(future);
         #[cfg(all(tokio_unstable, feature = "tracing"))]
-        let future = crate::util::trace::task(future, "task", _name, id.as_u64());
+        let future = crate::util::trace::task(future, "task", _meta, id.as_u64());
         self.inner.spawn(future, id)
     }
 
+    #[track_caller]
+    #[allow(dead_code)]
+    pub(crate) unsafe fn spawn_local_named<F>(
+        &self,
+        future: F,
+        _meta: SpawnMeta<'_>,
+    ) -> JoinHandle<F::Output>
+    where
+        F: Future + 'static,
+        F::Output: 'static,
+    {
+        let id = crate::runtime::task::Id::next();
+        #[cfg(all(
+            tokio_unstable,
+            tokio_taskdump,
+            feature = "rt",
+            target_os = "linux",
+            any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
+        ))]
+        let future = super::task::trace::Trace::root(future);
+        #[cfg(all(tokio_unstable, feature = "tracing"))]
+        let future = crate::util::trace::task(future, "task", _meta, id.as_u64());
+        self.inner.spawn_local(future, id)
+    }
+
     /// Returns the flavor of the current `Runtime`.
     ///
     /// # Examples

tokio/src/runtime/io/driver.rs

@@ -154,7 +154,7 @@ impl Driver {
                 // In case of wasm32_wasi this error happens, when trying to poll without subscriptions
                 // just return from the park, as there would be nothing, which wakes us up.
             }
-            Err(e) => panic!("unexpected error when polling the I/O driver: {:?}", e),
+            Err(e) => panic!("unexpected error when polling the I/O driver: {e:?}"),
         }
 
         // Process all the events that came in, dispatching appropriately
@@ -168,8 +168,7 @@ impl Driver {
                 self.signal_ready = true;
             } else {
                 let ready = Ready::from_mio(event);
-                // Use std::ptr::from_exposed_addr when stable
-                let ptr: *const ScheduledIo = token.0 as *const _;
+                let ptr = super::EXPOSE_IO.from_exposed_addr(token.0);
 
                 // Safety: we ensure that the pointers used as tokens are not freed
                 // until they are both deregistered from mio **and** we know the I/O