chore: prepare Tokio v1.42.1

The broadcast channel does not require values to implement `Sync` yet it calls
the `.clone()` method without synchronizing. This is unsound logic. This patch
adds per-value synchronization on receive to handle this case. It is unlikely
any usage of the broadcast channel is currently at risk of the unsoundeness
issue as it requires accessing a `!Sync` type during `.clone()`, which would be
very unusual when using the broadcast channel.

.cirrus.yml

@@ -1,7 +1,7 @@
 only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
 auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
 freebsd_instance:
-  image_family: freebsd-14-1
+  image_family: freebsd-14-2
 env:
   RUST_STABLE: 1.81
   RUST_NIGHTLY: nightly-2024-05-05

.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

@@ -18,7 +18,7 @@ env:
   rust_stable: stable
   rust_nightly: nightly-2024-05-05
   # Pin a specific miri version
-  rust_miri_nightly: nightly-2024-09-19
+  rust_miri_nightly: nightly-2024-10-21
   rust_clippy: '1.77'
   # When updating this, also update:
   # - README.md
@@ -180,7 +180,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
@@ -283,7 +283,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
@@ -460,10 +460,18 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Check semver
+      - name: Check `tokio` semver
         uses: obi1kenobi/cargo-semver-checks-action@v2
         with:
           rust-toolchain: ${{ env.rust_stable }}
+          package: tokio
+          release-type: minor
+      - name: Check semver for rest of the workspace
+        if: ${{ !startsWith(github.event.pull_request.base.ref, 'tokio-1.') }}
+        uses: obi1kenobi/cargo-semver-checks-action@v2
+        with:
+          rust-toolchain: ${{ env.rust_stable }}
+          exclude: tokio
           release-type: minor
 
   cross-check:
@@ -694,7 +702,14 @@ jobs:
           toolchain: ${{ env.rust_min }}
       - uses: Swatinem/rust-cache@v2
       - name: "check --workspace --all-features"
-        run: cargo check --workspace --all-features
+        run: |
+          if [[ "${{ github.event.pull_request.base.ref }}" =~ ^tokio-1\..* ]]; then
+            # Only check `tokio` crate as the PR is backporting to an earlier tokio release.
+            cargo check -p tokio --all-features
+          else
+            # Check all crates in the workspace
+            cargo check --workspace --all-features
+          fi
         env:
           RUSTFLAGS: "" # remove -Dwarnings
 
@@ -768,7 +783,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 }}
@@ -780,8 +803,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
@@ -985,7 +1008,7 @@ jobs:
       - name: Install cargo-hack, wasmtime, and cargo-wasi
         uses: taiki-e/install-action@v2
         with:
-          tool: cargo-hack,wasmtime,cargo-wasi
+          tool: cargo-hack,wasmtime
 
       - uses: Swatinem/rust-cache@v2
       - name: WASI test tokio full
@@ -1011,9 +1034,12 @@ jobs:
 
       - name: test tests-integration --features wasi-rt
         # TODO: this should become: `cargo hack wasi test --each-feature`
-        run: cargo wasi test --test rt_yield --features wasi-rt
+        run: cargo test --target ${{ matrix.target }} --test rt_yield --features wasi-rt
         if: matrix.target == 'wasm32-wasip1'
         working-directory: tests-integration
+        env:
+          CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
+          RUSTFLAGS: -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: test tests-integration --features wasi-threads-rt
         run: cargo test --target ${{ matrix.target }} --features wasi-threads-rt
@@ -1035,7 +1061,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 }}
@@ -1046,7 +1072,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
@@ -1106,11 +1132,11 @@ 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.

.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

@@ -201,7 +201,7 @@ run loom tests that test unstable features.
 You can run miri tests with
 ```
 MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields" \
-    cargo +nightly miri test --features full --lib
+    cargo +nightly miri test --features full --lib --tests
 ```
 
 ### Performing spellcheck on tokio codebase
@@ -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;
 

Cargo.toml

@@ -17,3 +17,16 @@ members = [
 
 [workspace.metadata.spellcheck]
 config = "spellcheck.toml"
+
+[workspace.lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = [
+  'cfg(fuzzing)',
+  'cfg(loom)',
+  'cfg(mio_unsupported_force_poll_poll)',
+  'cfg(tokio_allow_from_blocking_fd)',
+  'cfg(tokio_internal_mt_counters)',
+  'cfg(tokio_no_parking_lot)',
+  'cfg(tokio_no_tuning_tests)',
+  'cfg(tokio_taskdump)',
+  'cfg(tokio_unstable)',
+] }

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.41.0", features = ["full"] }
+tokio = { version = "1.42.1", 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**.
@@ -94,3 +95,6 @@ path = "named-pipe-multi-client.rs"
 [[example]]
 name = "dump"
 path = "dump.rs"
+
+[lints]
+workspace = true

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!"),
     }
 

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

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

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/Cargo.toml

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

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

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

@@ -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/tests/codecs.rs

@@ -75,32 +75,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 +94,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 +253,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 +268,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 +301,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_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/time_delay_queue.rs

@@ -111,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;
 
@@ -328,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/CHANGELOG.md

@@ -1,3 +1,56 @@
+# 1.42.1 (April 7nd, 2025)
+
+This release fixes a soundness issue in the broadcast channel. The channel
+accepts values that are `Send` but `!Sync`. Previously, the channel called
+`clone()` on these values without synchronizing. This release fixes the channel
+by synchronizing calls to `.clone()` (Thanks Austin Bonander for finding and
+reporting the issue).
+
+### Fixed
+
+- sync: synchronize `clone()` call in broadcast channel ([#7232])
+
+[#7232]: https://github.com/tokio-rs/tokio/pull/7232
+
+# 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
@@ -210,6 +263,20 @@ Yanked. Please use 1.39.1 instead.
 [#6709]: https://github.com/tokio-rs/tokio/pull/6709
 [#6710]: https://github.com/tokio-rs/tokio/pull/6710
 
+# 1.38.2 (April 2nd, 2025)
+
+This release fixes a soundness issue in the broadcast channel. The channel
+accepts values that are `Send` but `!Sync`. Previously, the channel called
+`clone()` on these values without synchronizing. This release fixes the channel
+by synchronizing calls to `.clone()` (Thanks Austin Bonander for finding and
+reporting the issue).
+
+### Fixed
+
+- sync: synchronize `clone()` call in broadcast channel ([#7232])
+
+[#7232]: https://github.com/tokio-rs/tokio/pull/7232
+
 # 1.38.1 (July 16th, 2024)
 
 This release fixes the bug identified as ([#6682]), which caused timers not

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.41.0"
+version = "1.42.1"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]

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.41.0", features = ["full"] }
+tokio = { version = "1.42.1", 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/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/io/async_fd.rs

@@ -872,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/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.
 //!
@@ -633,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/std/mod.rs

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

@@ -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/stream.rs

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

@@ -322,7 +322,7 @@ impl Spawner {
             // 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}")
             }
         }
     }

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

tokio/src/runtime/io/registration_set.rs

@@ -106,7 +106,7 @@ impl RegistrationSet {
 
         for io in pending {
             // safety: the registration is part of our list
-            unsafe { self.remove(synced, io.as_ref()) }
+            unsafe { self.remove(synced, &io) }
         }
 
         self.num_pending_release.store(0, Release);
@@ -114,9 +114,12 @@ impl RegistrationSet {
 
     // This function is marked as unsafe, because the caller must make sure that
     // `io` is part of the registration set.
-    pub(super) unsafe fn remove(&self, synced: &mut Synced, io: &ScheduledIo) {
-        super::EXPOSE_IO.unexpose_provenance(io);
-        let _ = synced.registrations.remove(io.into());
+    pub(super) unsafe fn remove(&self, synced: &mut Synced, io: &Arc<ScheduledIo>) {
+        // SAFETY: Pointers into an Arc are never null.
+        let io = unsafe { NonNull::new_unchecked(Arc::as_ptr(io).cast_mut()) };
+
+        super::EXPOSE_IO.unexpose_provenance(io.as_ptr());
+        let _ = synced.registrations.remove(io);
     }
 }
 

tokio/src/runtime/io/scheduled_io.rs

@@ -206,43 +206,23 @@ impl ScheduledIo {
     ///    specific tick.
     /// - `f`: a closure returning a new readiness value given the previous
     ///   readiness.
-    pub(super) fn set_readiness(&self, tick: Tick, f: impl Fn(Ready) -> Ready) {
-        let mut current = self.readiness.load(Acquire);
+    pub(super) fn set_readiness(&self, tick_op: Tick, f: impl Fn(Ready) -> Ready) {
+        let _ = self.readiness.fetch_update(AcqRel, Acquire, |curr| {
+            // If the io driver is shut down, then you are only allowed to clear readiness.
+            debug_assert!(SHUTDOWN.unpack(curr) == 0 || matches!(tick_op, Tick::Clear(_)));
 
-        // If the io driver is shut down, then you are only allowed to clear readiness.
-        debug_assert!(SHUTDOWN.unpack(current) == 0 || matches!(tick, Tick::Clear(_)));
+            const MAX_TICK: usize = TICK.max_value() + 1;
+            let tick = TICK.unpack(curr);
 
-        loop {
-            // Mask out the tick bits so that the modifying function doesn't see
-            // them.
-            let current_readiness = Ready::from_usize(current);
-            let new = f(current_readiness);
-
-            let new_tick = match tick {
-                Tick::Set => {
-                    let current = TICK.unpack(current);
-                    current.wrapping_add(1) % (TICK.max_value() + 1)
-                }
-                Tick::Clear(t) => {
-                    if TICK.unpack(current) as u8 != t {
-                        // Trying to clear readiness with an old event!
-                        return;
-                    }
-
-                    t as usize
-                }
+            let new_tick = match tick_op {
+                // Trying to clear readiness with an old event!
+                Tick::Clear(t) if tick as u8 != t => return None,
+                Tick::Clear(t) => t as usize,
+                Tick::Set => tick.wrapping_add(1) % MAX_TICK,
             };
-            let next = TICK.pack(new_tick, new.as_usize());
-
-            match self
-                .readiness
-                .compare_exchange(current, next, AcqRel, Acquire)
-            {
-                Ok(_) => return,
-                // we lost the race, retry!
-                Err(actual) => current = actual,
-            }
-        }
+            let ready = Ready::from_usize(READINESS.unpack(curr));
+            Some(TICK.pack(new_tick, f(ready).as_usize()))
+        });
     }
 
     /// Notifies all pending waiters that have registered interest in `ready`.
@@ -335,22 +315,16 @@ impl ScheduledIo {
         if ready.is_empty() && !is_shutdown {
             // Update the task info
             let mut waiters = self.waiters.lock();
-            let slot = match direction {
+            let waker = match direction {
                 Direction::Read => &mut waiters.reader,
                 Direction::Write => &mut waiters.writer,
             };
 
             // Avoid cloning the waker if one is already stored that matches the
             // current task.
-            match slot {
-                Some(existing) => {
-                    if !existing.will_wake(cx.waker()) {
-                        existing.clone_from(cx.waker());
-                    }
-                }
-                None => {
-                    *slot = Some(cx.waker().clone());
-                }
+            match waker {
+                Some(waker) => waker.clone_from(cx.waker()),
+                None => *waker = Some(cx.waker().clone()),
             }
 
             // Try again, in case the readiness was changed while we were
@@ -465,12 +439,11 @@ impl Future for Readiness<'_> {
                 State::Init => {
                     // Optimistically check existing readiness
                     let curr = scheduled_io.readiness.load(SeqCst);
-                    let ready = Ready::from_usize(READINESS.unpack(curr));
                     let is_shutdown = SHUTDOWN.unpack(curr) != 0;
 
                     // Safety: `waiter.interest` never changes
                     let interest = unsafe { (*waiter.get()).interest };
-                    let ready = ready.intersection(interest);
+                    let ready = Ready::from_usize(READINESS.unpack(curr)).intersection(interest);
 
                     if !ready.is_empty() || is_shutdown {
                         // Currently ready!
@@ -538,10 +511,7 @@ impl Future for Readiness<'_> {
                         *state = State::Done;
                     } else {
                         // Update the waker, if necessary.
-                        if !w.waker.as_ref().unwrap().will_wake(cx.waker()) {
-                            w.waker = Some(cx.waker().clone());
-                        }
-
+                        w.waker.as_mut().unwrap().clone_from(cx.waker());
                         return Poll::Pending;
                     }
 
@@ -566,8 +536,7 @@ impl Future for Readiness<'_> {
 
                     // The readiness state could have been cleared in the meantime,
                     // but we allow the returned ready set to be empty.
-                    let curr_ready = Ready::from_usize(READINESS.unpack(curr));
-                    let ready = curr_ready.intersection(w.interest);
+                    let ready = Ready::from_usize(READINESS.unpack(curr)).intersection(w.interest);
 
                     return Poll::Ready(ReadyEvent {
                         tick,

tokio/src/runtime/metrics/histogram.rs

@@ -264,14 +264,14 @@ impl HistogramBuilder {
             }
             None => self.histogram_type,
         };
-        let num_buckets = self.histogram_type.num_buckets();
+        let num_buckets = histogram_type.num_buckets();
 
         Histogram {
             buckets: (0..num_buckets)
                 .map(|_| MetricAtomicU64::new(0))
                 .collect::<Vec<_>>()
                 .into_boxed_slice(),
-            histogram_type: histogram_type,
+            histogram_type,
         }
     }
 }
@@ -303,6 +303,13 @@ mod test {
         .build()
     }
 
+    #[test]
+    fn test_legacy_builder() {
+        let mut builder = HistogramBuilder::new();
+        builder.legacy_mut(|b| b.num_buckets = 20);
+        assert_eq!(builder.build().num_buckets(), 20);
+    }
+
     #[test]
     fn log_scale_resolution_1() {
         let h = HistogramBuilder {
@@ -355,6 +362,9 @@ mod test {
 
         b.measure(4096, 1);
         assert_bucket_eq!(b, 9, 1);
+
+        b.measure(u64::MAX, 1);
+        assert_bucket_eq!(b, 9, 2);
     }
 
     #[test]

tokio/src/runtime/park.rs

@@ -109,7 +109,7 @@ impl Inner {
 
                 return;
             }
-            Err(actual) => panic!("inconsistent park state; actual = {}", actual),
+            Err(actual) => panic!("inconsistent park state; actual = {actual}"),
         }
 
         loop {
@@ -155,7 +155,7 @@ impl Inner {
 
                 return;
             }
-            Err(actual) => panic!("inconsistent park_timeout state; actual = {}", actual),
+            Err(actual) => panic!("inconsistent park_timeout state; actual = {actual}"),
         }
 
         // Wait with a timeout, and if we spuriously wake up or otherwise wake up
@@ -167,7 +167,7 @@ impl Inner {
         match self.state.swap(EMPTY, SeqCst) {
             NOTIFIED => {} // got a notification, hurray!
             PARKED => {}   // no notification, alas
-            n => panic!("inconsistent park_timeout state: {}", n),
+            n => panic!("inconsistent park_timeout state: {n}"),
         }
     }
 

tokio/src/runtime/scheduler/multi_thread/park.rs

@@ -151,7 +151,7 @@ impl Inner {
 
                 return;
             }
-            Err(actual) => panic!("inconsistent park state; actual = {}", actual),
+            Err(actual) => panic!("inconsistent park state; actual = {actual}"),
         }
 
         loop {
@@ -188,7 +188,7 @@ impl Inner {
 
                 return;
             }
-            Err(actual) => panic!("inconsistent park state; actual = {}", actual),
+            Err(actual) => panic!("inconsistent park state; actual = {actual}"),
         }
 
         driver.park(handle);
@@ -196,7 +196,7 @@ impl Inner {
         match self.state.swap(EMPTY, SeqCst) {
             NOTIFIED => {}      // got a notification, hurray!
             PARKED_DRIVER => {} // no notification, alas
-            n => panic!("inconsistent park_timeout state: {}", n),
+            n => panic!("inconsistent park_timeout state: {n}"),
         }
     }
 
@@ -211,7 +211,7 @@ impl Inner {
             NOTIFIED => {} // already unparked
             PARKED_CONDVAR => self.unpark_condvar(),
             PARKED_DRIVER => driver.unpark(),
-            actual => panic!("inconsistent state in unpark; actual = {}", actual),
+            actual => panic!("inconsistent state in unpark; actual = {actual}"),
         }
     }
 

tokio/src/runtime/scheduler/multi_thread/queue.rs

@@ -264,9 +264,7 @@ impl<T> Local<T> {
         assert_eq!(
             tail.wrapping_sub(head) as usize,
             LOCAL_QUEUE_CAPACITY,
-            "queue is not full; tail = {}; head = {}",
-            tail,
-            head
+            "queue is not full; tail = {tail}; head = {head}"
         );
 
         let prev = pack(head, head);
@@ -490,8 +488,7 @@ impl<T> Steal<T> {
 
         assert!(
             n <= LOCAL_QUEUE_CAPACITY as UnsignedShort / 2,
-            "actual = {}",
-            n
+            "actual = {n}"
         );
 
         let (first, _) = unpack(next_packed);

tokio/src/runtime/signal/mod.rs

@@ -118,7 +118,7 @@ impl Driver {
                 Ok(0) => panic!("EOF on self-pipe"),
                 Ok(_) => continue, // Keep reading
                 Err(e) if e.kind() == std_io::ErrorKind::WouldBlock => break,
-                Err(e) => panic!("Bad read on self-pipe: {}", e),
+                Err(e) => panic!("Bad read on self-pipe: {e}"),
             }
         }
 

tokio/src/runtime/time/wheel/mod.rs

@@ -298,13 +298,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 {
@@ -313,9 +307,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 {
@@ -323,9 +315,7 @@ mod test {
                     assert_eq!(
                         level,
                         level_for(0, a as u64),
-                        "level_for({}) -- binary = {:b}",
-                        a,
-                        a
+                        "level_for({a}) -- binary = {a:b}"
                     );
                 }
 
@@ -334,9 +324,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/src/signal/registry.rs

@@ -76,7 +76,7 @@ impl<S: Storage> Registry<S> {
     fn register_listener(&self, event_id: EventId) -> watch::Receiver<()> {
         self.storage
             .event_info(event_id)
-            .unwrap_or_else(|| panic!("invalid event_id: {}", event_id))
+            .unwrap_or_else(|| panic!("invalid event_id: {event_id}"))
             .tx
             .subscribe()
     }

tokio/src/signal/unix.rs

@@ -258,7 +258,7 @@ fn signal_enable(signal: SignalKind, handle: &Handle) -> io::Result<()> {
     if signal < 0 || signal_hook_registry::FORBIDDEN.contains(&signal) {
         return Err(Error::new(
             ErrorKind::Other,
-            format!("Refusing to register signal {}", signal),
+            format!("Refusing to register signal {signal}"),
         ));
     }
 

tokio/src/signal/windows.rs

@@ -12,13 +12,15 @@ use crate::signal::RxFuture;
 use std::io;
 use std::task::{Context, Poll};
 
-#[cfg(not(docsrs))]
+#[cfg(windows)]
 #[path = "windows/sys.rs"]
 mod imp;
-#[cfg(not(docsrs))]
+
+#[cfg(windows)]
 pub(crate) use self::imp::{OsExtraData, OsStorage};
 
-#[cfg(docsrs)]
+// For building documentation on Unix machines when the `docsrs` flag is set.
+#[cfg(not(windows))]
 #[path = "windows/stub.rs"]
 mod imp;
 

tokio/src/sync/broadcast.rs

@@ -118,7 +118,7 @@
 
 use crate::loom::cell::UnsafeCell;
 use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
-use crate::loom::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard};
+use crate::loom::sync::{Arc, Mutex, MutexGuard};
 use crate::runtime::coop::cooperative;
 use crate::util::linked_list::{self, GuardedLinkedList, LinkedList};
 use crate::util::WakeList;
@@ -255,7 +255,7 @@ pub mod error {
         fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
             match self {
                 RecvError::Closed => write!(f, "channel closed"),
-                RecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
+                RecvError::Lagged(amt) => write!(f, "channel lagged by {amt}"),
             }
         }
     }
@@ -291,7 +291,7 @@ pub mod error {
             match self {
                 TryRecvError::Empty => write!(f, "channel empty"),
                 TryRecvError::Closed => write!(f, "channel closed"),
-                TryRecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
+                TryRecvError::Lagged(amt) => write!(f, "channel lagged by {amt}"),
             }
         }
     }
@@ -304,7 +304,7 @@ use self::error::{RecvError, SendError, TryRecvError};
 /// Data shared between senders and receivers.
 struct Shared<T> {
     /// slots in the channel.
-    buffer: Box<[RwLock<Slot<T>>]>,
+    buffer: Box<[Mutex<Slot<T>>]>,
 
     /// Mask a position -> index.
     mask: usize,
@@ -348,7 +348,7 @@ struct Slot<T> {
     ///
     /// The value is set by `send` when the write lock is held. When a reader
     /// drops, `rem` is decremented. When it hits zero, the value is dropped.
-    val: UnsafeCell<Option<T>>,
+    val: Option<T>,
 }
 
 /// An entry in the wait queue.
@@ -386,7 +386,7 @@ generate_addr_of_methods! {
 }
 
 struct RecvGuard<'a, T> {
-    slot: RwLockReadGuard<'a, Slot<T>>,
+    slot: MutexGuard<'a, Slot<T>>,
 }
 
 /// Receive a value future.
@@ -395,11 +395,15 @@ struct Recv<'a, T> {
     receiver: &'a mut Receiver<T>,
 
     /// Entry in the waiter `LinkedList`.
-    waiter: UnsafeCell<Waiter>,
+    waiter: WaiterCell,
 }
 
-unsafe impl<'a, T: Send> Send for Recv<'a, T> {}
-unsafe impl<'a, T: Send> Sync for Recv<'a, T> {}
+// The wrapper around `UnsafeCell` isolates the unsafe impl `Send` and `Sync`
+// from `Recv`.
+struct WaiterCell(UnsafeCell<Waiter>);
+
+unsafe impl Send for WaiterCell {}
+unsafe impl Sync for WaiterCell {}
 
 /// Max number of receivers. Reserve space to lock.
 const MAX_RECEIVERS: usize = usize::MAX >> 2;
@@ -467,12 +471,6 @@ pub fn channel<T: Clone>(capacity: usize) -> (Sender<T>, Receiver<T>) {
     (tx, rx)
 }
 
-unsafe impl<T: Send> Send for Sender<T> {}
-unsafe impl<T: Send> Sync for Sender<T> {}
-
-unsafe impl<T: Send> Send for Receiver<T> {}
-unsafe impl<T: Send> Sync for Receiver<T> {}
-
 impl<T> Sender<T> {
     /// Creates the sending-half of the [`broadcast`] channel.
     ///
@@ -511,10 +509,10 @@ impl<T> Sender<T> {
         let mut buffer = Vec::with_capacity(capacity);
 
         for i in 0..capacity {
-            buffer.push(RwLock::new(Slot {
+            buffer.push(Mutex::new(Slot {
                 rem: AtomicUsize::new(0),
                 pos: (i as u64).wrapping_sub(capacity as u64),
-                val: UnsafeCell::new(None),
+                val: None,
             }));
         }
 
@@ -600,7 +598,7 @@ impl<T> Sender<T> {
         tail.pos = tail.pos.wrapping_add(1);
 
         // Get the slot
-        let mut slot = self.shared.buffer[idx].write();
+        let mut slot = self.shared.buffer[idx].lock();
 
         // Track the position
         slot.pos = pos;
@@ -609,7 +607,7 @@ impl<T> Sender<T> {
         slot.rem.with_mut(|v| *v = rem);
 
         // Write the value
-        slot.val = UnsafeCell::new(Some(value));
+        slot.val = Some(value);
 
         // Release the slot lock before notifying the receivers.
         drop(slot);
@@ -696,7 +694,7 @@ impl<T> Sender<T> {
         while low < high {
             let mid = low + (high - low) / 2;
             let idx = base_idx.wrapping_add(mid) & self.shared.mask;
-            if self.shared.buffer[idx].read().rem.load(SeqCst) == 0 {
+            if self.shared.buffer[idx].lock().rem.load(SeqCst) == 0 {
                 low = mid + 1;
             } else {
                 high = mid;
@@ -738,7 +736,7 @@ impl<T> Sender<T> {
         let tail = self.shared.tail.lock();
 
         let idx = (tail.pos.wrapping_sub(1) & self.shared.mask as u64) as usize;
-        self.shared.buffer[idx].read().rem.load(SeqCst) == 0
+        self.shared.buffer[idx].lock().rem.load(SeqCst) == 0
     }
 
     /// Returns the number of active receivers.
@@ -1058,7 +1056,7 @@ impl<T> Receiver<T> {
         let idx = (self.next & self.shared.mask as u64) as usize;
 
         // The slot holding the next value to read
-        let mut slot = self.shared.buffer[idx].read();
+        let mut slot = self.shared.buffer[idx].lock();
 
         if slot.pos != self.next {
             // Release the `slot` lock before attempting to acquire the `tail`
@@ -1075,7 +1073,7 @@ impl<T> Receiver<T> {
             let mut tail = self.shared.tail.lock();
 
             // Acquire slot lock again
-            slot = self.shared.buffer[idx].read();
+            slot = self.shared.buffer[idx].lock();
 
             // Make sure the position did not change. This could happen in the
             // unlikely event that the buffer is wrapped between dropping the
@@ -1367,12 +1365,12 @@ impl<'a, T> Recv<'a, T> {
     fn new(receiver: &'a mut Receiver<T>) -> Recv<'a, T> {
         Recv {
             receiver,
-            waiter: UnsafeCell::new(Waiter {
+            waiter: WaiterCell(UnsafeCell::new(Waiter {
                 queued: AtomicBool::new(false),
                 waker: None,
                 pointers: linked_list::Pointers::new(),
                 _p: PhantomPinned,
-            }),
+            })),
         }
     }
 
@@ -1384,7 +1382,7 @@ impl<'a, T> Recv<'a, T> {
             is_unpin::<&mut Receiver<T>>();
 
             let me = self.get_unchecked_mut();
-            (me.receiver, &me.waiter)
+            (me.receiver, &me.waiter.0)
         }
     }
 }
@@ -1418,6 +1416,7 @@ impl<'a, T> Drop for Recv<'a, T> {
         // `Shared::notify_rx` before we drop the object.
         let queued = self
             .waiter
+            .0
             .with(|ptr| unsafe { (*ptr).queued.load(Acquire) });
 
         // If the waiter is queued, we need to unlink it from the waiters list.
@@ -1432,6 +1431,7 @@ impl<'a, T> Drop for Recv<'a, T> {
             // `Relaxed` order suffices because we hold the tail lock.
             let queued = self
                 .waiter
+                .0
                 .with_mut(|ptr| unsafe { (*ptr).queued.load(Relaxed) });
 
             if queued {
@@ -1440,7 +1440,7 @@ impl<'a, T> Drop for Recv<'a, T> {
                 // safety: tail lock is held and the wait node is verified to be in
                 // the list.
                 unsafe {
-                    self.waiter.with_mut(|ptr| {
+                    self.waiter.0.with_mut(|ptr| {
                         tail.waiters.remove((&mut *ptr).into());
                     });
                 }
@@ -1486,7 +1486,7 @@ impl<'a, T> RecvGuard<'a, T> {
     where
         T: Clone,
     {
-        self.slot.val.with(|ptr| unsafe { (*ptr).clone() })
+        self.slot.val.clone()
     }
 }
 
@@ -1494,8 +1494,7 @@ impl<'a, T> Drop for RecvGuard<'a, T> {
     fn drop(&mut self) {
         // Decrement the remaining counter
         if 1 == self.slot.rem.fetch_sub(1, SeqCst) {
-            // Safety: Last receiver, drop the value
-            self.slot.val.with_mut(|ptr| unsafe { *ptr = None });
+            self.slot.val = None;
         }
     }
 }

tokio/src/sync/rwlock.rs

@@ -274,8 +274,7 @@ impl<T: ?Sized> RwLock<T> {
     {
         assert!(
             max_reads <= MAX_READS,
-            "a RwLock may not be created with more than {} readers",
-            MAX_READS
+            "a RwLock may not be created with more than {MAX_READS} readers"
         );
 
         #[cfg(all(tokio_unstable, feature = "tracing"))]

tokio/src/time/error.rs

@@ -96,7 +96,7 @@ impl fmt::Display for Error {
             Kind::AtCapacity => "timer is at capacity and cannot create a new entry",
             Kind::Invalid => "timer duration exceeds maximum duration",
         };
-        write!(fmt, "{}", descr)
+        write!(fmt, "{descr}")
     }
 }
 

tokio/src/time/instant.rs

@@ -116,7 +116,7 @@ impl Instant {
     }
 
     /// Returns the amount of time elapsed since this instant was created,
-    /// or zero duration if that this instant is in the future.
+    /// or zero duration if this instant is in the future.
     ///
     /// # Examples
     ///

tokio/src/time/sleep.rs

@@ -447,7 +447,7 @@ impl Future for Sleep {
         let _ao_poll_span = self.inner.ctx.async_op_poll_span.clone().entered();
         match ready!(self.as_mut().poll_elapsed(cx)) {
             Ok(()) => Poll::Ready(()),
-            Err(e) => panic!("timer error: {}", e),
+            Err(e) => panic!("timer error: {e}"),
         }
     }
 }

tokio/tests/buffered.rs

@@ -9,7 +9,7 @@ use std::net::TcpStream;
 use std::thread;
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `socket` on miri.
 async fn echo_server() {
     const N: usize = 1024;
 

tokio/tests/coop_budget.rs

@@ -22,7 +22,7 @@ use tokio::net::UdpSocket;
 /// Since we are both sending and receiving, that should happen once per 64 packets, because budgets are of size 128
 /// and there are two budget events per packet, a send and a recv.
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `socket` on miri.
 async fn coop_budget_udp_send_recv() {
     const BUDGET: usize = 128;
     const N_ITERATIONS: usize = 1024;

tokio/tests/fs_copy.rs

@@ -5,7 +5,7 @@ use tempfile::tempdir;
 use tokio::fs;
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `fchmod` in miri.
 async fn copy() {
     let dir = tempdir().unwrap();
 
@@ -22,7 +22,7 @@ async fn copy() {
 }
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `fchmod` in miri.
 async fn copy_permissions() {
     let dir = tempdir().unwrap();
     let from_path = dir.path().join("foo.txt");

tokio/tests/fs_file.rs

@@ -211,7 +211,7 @@ async fn file_debug_fmt() {
     let file = File::open(tempfile.path()).await.unwrap();
 
     assert_eq!(
-        &format!("{:?}", file)[0..33],
+        &format!("{file:?}")[0..33],
         "tokio::fs::File { std: File { fd:"
     );
 }

tokio/tests/fs_link.rs

@@ -7,7 +7,7 @@ use std::io::Write;
 use tempfile::tempdir;
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `linkat` in miri.
 async fn test_hard_link() {
     let dir = tempdir().unwrap();
     let src = dir.path().join("src.txt");

tokio/tests/fs_open_options.rs

@@ -59,8 +59,7 @@ async fn open_options_mode() {
     // TESTING HACK: use Debug output to check the stored data
     assert!(
         mode.contains("mode: 420 ") || mode.contains("mode: 0o000644 "),
-        "mode is: {}",
-        mode
+        "mode is: {mode}"
     );
 }
 

tokio/tests/fs_try_exists.rs

@@ -5,7 +5,7 @@ use tempfile::tempdir;
 use tokio::fs;
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `chmod` in miri.
 async fn try_exists() {
     let dir = tempdir().unwrap();
 

tokio/tests/io_async_fd.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(unix, feature = "full", not(miri)))]
+#![cfg(all(unix, feature = "full"))]
 
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::sync::{
@@ -141,13 +141,14 @@ fn drain(mut fd: &FileDescriptor, mut amt: usize) {
         match fd.read(&mut buf[..]) {
             Err(e) if e.kind() == ErrorKind::WouldBlock => {}
             Ok(0) => panic!("unexpected EOF"),
-            Err(e) => panic!("unexpected error: {:?}", e),
+            Err(e) => panic!("unexpected error: {e:?}"),
             Ok(x) => amt -= x,
         }
     }
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn initially_writable() {
     let (a, b) = socketpair();
 
@@ -166,6 +167,7 @@ async fn initially_writable() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn reset_readable() {
     let (a, mut b) = socketpair();
 
@@ -210,6 +212,7 @@ async fn reset_readable() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn reset_writable() {
     let (a, b) = socketpair();
 
@@ -247,6 +250,7 @@ impl<T: AsRawFd> AsRawFd for ArcFd<T> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn drop_closes() {
     let (a, mut b) = socketpair();
 
@@ -287,6 +291,7 @@ async fn drop_closes() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn reregister() {
     let (a, _b) = socketpair();
 
@@ -296,7 +301,8 @@ async fn reregister() {
 }
 
 #[tokio::test]
-async fn try_io() {
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
+async fn guard_try_io() {
     let (a, mut b) = socketpair();
 
     b.write_all(b"0").unwrap();
@@ -331,6 +337,109 @@ async fn try_io() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
+async fn try_io_readable() {
+    let (a, mut b) = socketpair();
+    let mut afd_a = AsyncFd::new(a).unwrap();
+
+    // Give the runtime some time to update bookkeeping.
+    tokio::task::yield_now().await;
+
+    {
+        let mut called = false;
+        let _ = afd_a.try_io_mut(Interest::READABLE, |_| {
+            called = true;
+            Ok(())
+        });
+        assert!(
+            !called,
+            "closure should not have been called, since socket should not be readable"
+        );
+    }
+
+    // Make `a` readable by writing to `b`.
+    // Give the runtime some time to update bookkeeping.
+    b.write_all(&[0]).unwrap();
+    tokio::task::yield_now().await;
+
+    {
+        let mut called = false;
+        let _ = afd_a.try_io(Interest::READABLE, |_| {
+            called = true;
+            Ok(())
+        });
+        assert!(
+            called,
+            "closure should have been called, since socket should have data available to read"
+        );
+    }
+
+    {
+        let mut called = false;
+        let _ = afd_a.try_io(Interest::READABLE, |_| {
+            called = true;
+            io::Result::<()>::Err(ErrorKind::WouldBlock.into())
+        });
+        assert!(
+            called,
+            "closure should have been called, since socket should have data available to read"
+        );
+    }
+
+    {
+        let mut called = false;
+        let _ = afd_a.try_io(Interest::READABLE, |_| {
+            called = true;
+            Ok(())
+        });
+        assert!(!called, "closure should not have been called, since socket readable state should have been cleared");
+    }
+}
+
+#[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
+async fn try_io_writable() {
+    let (a, _b) = socketpair();
+    let afd_a = AsyncFd::new(a).unwrap();
+
+    // Give the runtime some time to update bookkeeping.
+    tokio::task::yield_now().await;
+
+    {
+        let mut called = false;
+        let _ = afd_a.try_io(Interest::WRITABLE, |_| {
+            called = true;
+            Ok(())
+        });
+        assert!(
+            called,
+            "closure should have been called, since socket should still be marked as writable"
+        );
+    }
+    {
+        let mut called = false;
+        let _ = afd_a.try_io(Interest::WRITABLE, |_| {
+            called = true;
+            io::Result::<()>::Err(ErrorKind::WouldBlock.into())
+        });
+        assert!(
+            called,
+            "closure should have been called, since socket should still be marked as writable"
+        );
+    }
+
+    {
+        let mut called = false;
+        let _ = afd_a.try_io(Interest::WRITABLE, |_| {
+            called = true;
+            Ok(())
+        });
+        assert!(!called, "closure should not have been called, since socket writable state should have been cleared");
+    }
+}
+
+#[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn multiple_waiters() {
     let (a, mut b) = socketpair();
     let afd_a = Arc::new(AsyncFd::new(a).unwrap());
@@ -379,6 +488,7 @@ async fn multiple_waiters() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn poll_fns() {
     let (a, b) = socketpair();
     let afd_a = Arc::new(AsyncFd::new(a).unwrap());
@@ -472,6 +582,7 @@ fn rt() -> tokio::runtime::Runtime {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_wakes_currently_pending() {
     let rt = rt();
 
@@ -493,6 +604,7 @@ fn driver_shutdown_wakes_currently_pending() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_wakes_future_pending() {
     let rt = rt();
 
@@ -508,6 +620,7 @@ fn driver_shutdown_wakes_future_pending() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_wakes_pending_race() {
     // TODO: make this a loom test
     for _ in 0..100 {
@@ -538,6 +651,7 @@ async fn poll_writable<T: AsRawFd>(fd: &AsyncFd<T>) -> std::io::Result<AsyncFdRe
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_wakes_currently_pending_polls() {
     let rt = rt();
 
@@ -560,6 +674,7 @@ fn driver_shutdown_wakes_currently_pending_polls() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_wakes_poll() {
     let rt = rt();
 
@@ -576,6 +691,7 @@ fn driver_shutdown_wakes_poll() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_then_clear_readiness() {
     let rt = rt();
 
@@ -593,6 +709,7 @@ fn driver_shutdown_then_clear_readiness() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 fn driver_shutdown_wakes_poll_race() {
     // TODO: make this a loom test
     for _ in 0..100 {
@@ -615,6 +732,7 @@ fn driver_shutdown_wakes_poll_race() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No socket in miri.
 #[cfg(any(target_os = "linux", target_os = "android"))]
 async fn priority_event_on_oob_data() {
     use std::net::SocketAddr;
@@ -655,7 +773,7 @@ fn send_oob_data<S: AsRawFd>(stream: &S, data: &[u8]) -> io::Result<usize> {
 }
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn clear_ready_matching_clears_ready() {
     use tokio::io::{Interest, Ready};
 
@@ -679,7 +797,7 @@ async fn clear_ready_matching_clears_ready() {
 }
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn clear_ready_matching_clears_ready_mut() {
     use tokio::io::{Interest, Ready};
 
@@ -703,8 +821,8 @@ async fn clear_ready_matching_clears_ready_mut() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No socket in miri.
 #[cfg(target_os = "linux")]
-#[cfg_attr(miri, ignore)]
 async fn await_error_readiness_timestamping() {
     use std::net::{Ipv4Addr, SocketAddr};
 
@@ -760,8 +878,8 @@ fn configure_timestamping_socket(udp_socket: &std::net::UdpSocket) -> std::io::R
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 #[cfg(target_os = "linux")]
-#[cfg_attr(miri, ignore)]
 async fn await_error_readiness_invalid_address() {
     use std::net::{Ipv4Addr, SocketAddr};
     use tokio::io::{Interest, Ready};

tokio/tests/io_copy_bidirectional.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi does not support bind()
+#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support bind()
 
 use std::time::Duration;
 use tokio::io::{self, copy_bidirectional, AsyncReadExt, AsyncWriteExt};
@@ -59,6 +59,7 @@ where
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 async fn test_basic_transfer() {
     symmetric(|_handle, mut a, mut b| async move {
         a.write_all(b"test").await.unwrap();
@@ -70,6 +71,7 @@ async fn test_basic_transfer() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 async fn test_transfer_after_close() {
     symmetric(|handle, mut a, mut b| async move {
         AsyncWriteExt::shutdown(&mut a).await.unwrap();
@@ -89,6 +91,7 @@ async fn test_transfer_after_close() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 async fn blocking_one_side_does_not_block_other() {
     symmetric(|handle, mut a, mut b| async move {
         block_write(&mut a).await;

tokio/tests/io_driver.rs

@@ -1,6 +1,6 @@
 #![warn(rust_2018_idioms)]
 // Wasi does not support panic recovery or threading
-#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
+#![cfg(all(feature = "full", not(target_os = "wasi")))]
 
 use tokio::net::TcpListener;
 use tokio::runtime;
@@ -32,6 +32,7 @@ impl<T> Task<T> {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn test_drop_on_notify() {
     // When the reactor receives a kernel notification, it notifies the
     // task that holds the associated socket. If this notification results in
@@ -90,6 +91,7 @@ fn test_drop_on_notify() {
 #[should_panic(
     expected = "A Tokio 1.x context was found, but IO is disabled. Call `enable_io` on the runtime builder to enable IO."
 )]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn panics_when_io_disabled() {
     let rt = runtime::Builder::new_current_thread().build().unwrap();
 

tokio/tests/io_driver_drop.rs

@@ -1,11 +1,12 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi does not support bind
+#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support bind
 
 use tokio::net::TcpListener;
 use tokio::runtime;
 use tokio_test::{assert_err, assert_pending, assert_ready, task};
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn tcp_doesnt_block() {
     let rt = rt();
 
@@ -25,6 +26,7 @@ fn tcp_doesnt_block() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn drop_wakes() {
     let rt = rt();
 

tokio/tests/io_read_to_string.rs

@@ -23,9 +23,9 @@ async fn to_string_does_not_truncate_on_utf8_error() {
     let mut s = "abc".to_string();
 
     match AsyncReadExt::read_to_string(&mut data.as_slice(), &mut s).await {
-        Ok(len) => panic!("Should fail: {} bytes.", len),
+        Ok(len) => panic!("Should fail: {len} bytes."),
         Err(err) if err.to_string() == "stream did not contain valid UTF-8" => {}
-        Err(err) => panic!("Fail: {}.", err),
+        Err(err) => panic!("Fail: {err}."),
     }
 
     assert_eq!(s, "abc");
@@ -40,9 +40,9 @@ async fn to_string_does_not_truncate_on_io_error() {
     let mut s = "abc".to_string();
 
     match AsyncReadExt::read_to_string(&mut mock, &mut s).await {
-        Ok(len) => panic!("Should fail: {} bytes.", len),
+        Ok(len) => panic!("Should fail: {len} bytes."),
         Err(err) if err.to_string() == "whoops" => {}
-        Err(err) => panic!("Fail: {}.", err),
+        Err(err) => panic!("Fail: {err}."),
     }
 
     assert_eq!(s, "abc");

tokio/tests/net_bind_resource.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi doesn't support panic recovery or bind
+#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi doesn't support panic recovery or bind
 
 use tokio::net::TcpListener;
 
@@ -7,6 +7,7 @@ use std::net;
 
 #[test]
 #[should_panic]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn no_runtime_panics_binding_net_tcp_listener() {
     let listener = net::TcpListener::bind("127.0.0.1:0").expect("failed to bind listener");
     let _ = TcpListener::try_from(listener);

tokio/tests/net_lookup_host.rs

@@ -23,7 +23,7 @@ async fn lookup_str_socket_addr() {
 }
 
 #[tokio::test]
-#[cfg_attr(miri, ignore)]
+#[cfg_attr(miri, ignore)] // No `getaddrinfo` in miri.
 async fn resolve_dns() -> io::Result<()> {
     let mut hosts = net::lookup_host("localhost:3000").await?;
     let host = hosts.next().unwrap();

tokio/tests/net_panic.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
+#![cfg(all(feature = "full", not(target_os = "wasi")))]
 #![cfg(panic = "unwind")]
 
 use std::error::Error;
@@ -12,6 +12,7 @@ mod support {
 use support::panic::test_panic;
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn udp_socket_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
     use std::net::SocketAddr;
     use tokio::net::UdpSocket;
@@ -34,6 +35,7 @@ fn udp_socket_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn tcp_listener_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
     let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
     std_listener.set_nonblocking(true).unwrap();
@@ -52,6 +54,7 @@ fn tcp_listener_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn tcp_stream_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
     let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
 
@@ -73,6 +76,7 @@ fn tcp_stream_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
 
 #[test]
 #[cfg(unix)]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn unix_listener_bind_panic_caller() -> Result<(), Box<dyn Error>> {
     use tokio::net::UnixListener;
 
@@ -94,6 +98,7 @@ fn unix_listener_bind_panic_caller() -> Result<(), Box<dyn Error>> {
 
 #[test]
 #[cfg(unix)]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn unix_listener_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
     use tokio::net::UnixListener;
 
@@ -116,6 +121,7 @@ fn unix_listener_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
 
 #[test]
 #[cfg(unix)]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn unix_stream_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
     use tokio::net::UnixStream;
 
@@ -139,6 +145,7 @@ fn unix_stream_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
 
 #[test]
 #[cfg(unix)]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn unix_datagram_from_std_panic_caller() -> Result<(), Box<dyn Error>> {
     use std::os::unix::net::UnixDatagram as StdUDS;
     use tokio::net::UnixDatagram;

tokio/tests/net_unix_pipe.rs

@@ -1,6 +1,5 @@
 #![cfg(feature = "full")]
 #![cfg(unix)]
-#![cfg(not(miri))]
 
 use tokio::io::{AsyncReadExt, AsyncWriteExt, Interest};
 use tokio::net::unix::pipe;
@@ -38,6 +37,7 @@ impl AsRef<Path> for TempFifo {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn fifo_simple_send() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -69,6 +69,7 @@ async fn fifo_simple_send() -> io::Result<()> {
 
 #[tokio::test]
 #[cfg(target_os = "linux")]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn fifo_simple_send_sender_first() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -105,6 +106,7 @@ async fn write_and_close(path: impl AsRef<Path>, msg: &[u8]) -> io::Result<()> {
 /// Checks EOF behavior with single reader and writers sequentially opening
 /// and closing a FIFO.
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn fifo_multiple_writes() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -133,6 +135,7 @@ async fn fifo_multiple_writes() -> io::Result<()> {
 /// with writers sequentially opening and closing a FIFO.
 #[tokio::test]
 #[cfg(target_os = "linux")]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 async fn fifo_resilient_reader() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -163,6 +166,7 @@ async fn fifo_resilient_reader() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `O_NONBLOCK` for open64 in miri.
 async fn open_detects_not_a_fifo() -> io::Result<()> {
     let dir = tempfile::Builder::new()
         .prefix("tokio-fifo-tests")
@@ -185,6 +189,7 @@ async fn open_detects_not_a_fifo() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn from_file() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -221,6 +226,7 @@ async fn from_file() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `fstat` in miri.
 async fn from_file_detects_not_a_fifo() -> io::Result<()> {
     let dir = tempfile::Builder::new()
         .prefix("tokio-fifo-tests")
@@ -245,6 +251,7 @@ async fn from_file_detects_not_a_fifo() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn from_file_detects_wrong_access_mode() -> io::Result<()> {
     let fifo = TempFifo::new("wrong_access_mode")?;
 
@@ -276,6 +283,7 @@ fn is_nonblocking<T: AsRawFd>(fd: &T) -> io::Result<bool> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn from_file_sets_nonblock() -> io::Result<()> {
     let fifo = TempFifo::new("sets_nonblock")?;
 
@@ -303,6 +311,7 @@ fn writable_by_poll(writer: &pipe::Sender) -> bool {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn try_read_write() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -343,6 +352,7 @@ async fn try_read_write() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn try_read_write_vectored() -> io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -390,6 +400,7 @@ async fn try_read_write_vectored() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No `mkfifo` in miri.
 async fn try_read_buf() -> std::io::Result<()> {
     const DATA: &[u8] = b"this is some data to write to the fifo";
 
@@ -458,6 +469,7 @@ async fn anon_pipe_simple_send() -> io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn anon_pipe_spawn_echo() -> std::io::Result<()> {
     use tokio::process::Command;
 
@@ -488,6 +500,7 @@ async fn anon_pipe_spawn_echo() -> std::io::Result<()> {
 
 #[tokio::test]
 #[cfg(target_os = "linux")]
+#[cfg_attr(miri, ignore)] // No `fstat` in miri.
 async fn anon_pipe_from_owned_fd() -> std::io::Result<()> {
     use nix::fcntl::OFlag;
 
@@ -507,6 +520,7 @@ async fn anon_pipe_from_owned_fd() -> std::io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn anon_pipe_into_nonblocking_fd() -> std::io::Result<()> {
     let (tx, rx) = pipe::pipe()?;
 
@@ -520,6 +534,7 @@ async fn anon_pipe_into_nonblocking_fd() -> std::io::Result<()> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // No F_GETFL for fcntl in miri.
 async fn anon_pipe_into_blocking_fd() -> std::io::Result<()> {
     let (tx, rx) = pipe::pipe()?;
 

tokio/tests/no_rt.rs

@@ -1,4 +1,4 @@
-#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi does not support panic recovery
+#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support panic recovery
 
 use tokio::net::TcpStream;
 use tokio::sync::oneshot;
@@ -20,6 +20,7 @@ fn timeout_panics_when_no_tokio_context() {
 #[should_panic(
     expected = "there is no reactor running, must be called from the context of a Tokio 1.x runtime"
 )]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn panics_when_no_reactor() {
     let srv = TcpListener::bind("127.0.0.1:0").unwrap();
     let addr = srv.local_addr().unwrap();
@@ -36,6 +37,7 @@ async fn timeout_value() {
 #[should_panic(
     expected = "there is no reactor running, must be called from the context of a Tokio 1.x runtime"
 )]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn io_panics_when_no_tokio_context() {
     let _ = tokio::net::TcpListener::from_std(std::net::TcpListener::bind("127.0.0.1:0").unwrap());
 }

tokio/tests/process_issue_42.rs

@@ -20,7 +20,7 @@ async fn issue_42() {
         task::spawn(async {
             let processes = (0..10usize).map(|i| {
                 let mut child = Command::new("echo")
-                    .arg(format!("I am spawned process #{}", i))
+                    .arg(format!("I am spawned process #{i}"))
                     .stdin(Stdio::null())
                     .stdout(Stdio::null())
                     .stderr(Stdio::null())

tokio/tests/rt_common.rs

@@ -542,6 +542,7 @@ rt_test! {
     }
 
     #[cfg(not(target_os="wasi"))] // Wasi does not support bind
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn block_on_socket() {
         let rt = rt();
@@ -616,6 +617,7 @@ rt_test! {
     }
 
     #[cfg(not(target_os="wasi"))] // Wasi does not support bind
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn socket_from_blocking() {
         let rt = rt();
@@ -686,6 +688,7 @@ rt_test! {
     // concern. There also isn't a great/obvious solution to take. For now, the
     // test is disabled.
     #[cfg(not(windows))]
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[cfg(not(target_os="wasi"))] // Wasi does not support bind or threads
     fn io_driver_called_when_under_load() {
         let rt = rt();
@@ -740,6 +743,7 @@ rt_test! {
     /// spuriously.
     #[test]
     #[cfg(not(target_os="wasi"))]
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     fn yield_defers_until_park() {
         for _ in 0..10 {
             if yield_defers_until_park_inner() {
@@ -839,6 +843,7 @@ rt_test! {
     }
 
     #[cfg(not(target_os="wasi"))] // Wasi does not support threads
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn client_server_block_on() {
         let rt = rt();
@@ -1004,6 +1009,7 @@ rt_test! {
     }
 
     #[cfg(not(target_os="wasi"))] // Wasi doesn't support UDP or bind()
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn io_notify_while_shutting_down() {
         use tokio::net::UdpSocket;
@@ -1135,6 +1141,7 @@ rt_test! {
     }
 
     #[cfg(not(target_os = "wasi"))] // Wasi does not support bind
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn local_set_block_on_socket() {
         let rt = rt();
@@ -1157,6 +1164,7 @@ rt_test! {
     }
 
     #[cfg(not(target_os = "wasi"))] // Wasi does not support bind
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn local_set_client_server_block_on() {
         let rt = rt();

tokio/tests/rt_handle_block_on.rs

@@ -212,6 +212,7 @@ rt_test! {
     // ==== net ======
 
     #[test]
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     fn tcp_listener_bind() {
         let rt = rt();
         let _enter = rt.enter();
@@ -262,6 +263,7 @@ rt_test! {
     }
 
     #[test]
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     fn udp_socket_bind() {
         let rt = rt();
         let _enter = rt.enter();
@@ -422,6 +424,7 @@ rt_test! {
 #[cfg(not(target_os = "wasi"))]
 multi_threaded_rt_test! {
     #[cfg(unix)]
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
     #[test]
     fn unix_listener_bind() {
         let rt = rt();

tokio/tests/rt_metrics.rs

@@ -66,10 +66,10 @@ fn global_queue_depth_current_thread() {
 
 #[test]
 fn global_queue_depth_multi_thread() {
-    let rt = threaded();
-    let metrics = rt.metrics();
-
     for _ in 0..10 {
+        let rt = threaded();
+        let metrics = rt.metrics();
+
         if let Ok(_blocking_tasks) = try_block_threaded(&rt) {
             for i in 0..10 {
                 assert_eq!(i, metrics.global_queue_depth());
@@ -93,7 +93,7 @@ fn try_block_threaded(rt: &Runtime) -> Result<Vec<mpsc::Sender<()>>, mpsc::RecvT
 
             // Spawn a task per runtime worker to block it.
             rt.spawn(async move {
-                tx.send(()).unwrap();
+                tx.send(()).ok();
                 barrier.recv().ok();
             });
 

tokio/tests/rt_threaded.rs

@@ -189,6 +189,7 @@ fn lifo_slot_budget() {
 }
 
 #[test]
+#[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn spawn_shutdown() {
     let rt = rt();
     let (tx, rx) = mpsc::channel();

tokio/tests/rt_unstable_metrics.rs

@@ -8,12 +8,12 @@
 ))]
 
 use std::future::Future;
-use std::sync::{Arc, Mutex};
+use std::sync::{mpsc, Arc, Mutex};
 use std::task::Poll;
 use std::thread;
 use tokio::macros::support::poll_fn;
 
-use tokio::runtime::{HistogramConfiguration, LogHistogram, Runtime};
+use tokio::runtime::{HistogramConfiguration, HistogramScale, LogHistogram, Runtime};
 use tokio::task::consume_budget;
 use tokio::time::{self, Duration};
 
@@ -295,42 +295,34 @@ fn worker_noop_count() {
 }
 
 #[test]
-#[ignore] // this test is flaky, see https://github.com/tokio-rs/tokio/issues/6470
 fn worker_steal_count() {
     // This metric only applies to the multi-threaded runtime.
-    //
-    // We use a blocking channel to backup one worker thread.
-    use std::sync::mpsc::channel;
+    for _ in 0..10 {
+        let rt = threaded_no_lifo();
+        let metrics = rt.metrics();
 
-    let rt = threaded_no_lifo();
-    let metrics = rt.metrics();
+        let successfully_spawned_stealable_task = rt.block_on(async {
+            // The call to `try_spawn_stealable_task` may time out, which means
+            // that the sending task couldn't be scheduled due to a deadlock in
+            // the runtime.
+            // This is expected behaviour, we just retry until we succeed or
+            // exhaust all tries, the latter causing this test to fail.
+            try_spawn_stealable_task().await.is_ok()
+        });
 
-    rt.block_on(async {
-        let (tx, rx) = channel();
+        drop(rt);
 
-        // Move to the runtime.
-        tokio::spawn(async move {
-            // Spawn the task that sends to the channel
-            //
-            // Since the lifo slot is disabled, this task is stealable.
-            tokio::spawn(async move {
-                tx.send(()).unwrap();
-            });
+        if successfully_spawned_stealable_task {
+            let n: u64 = (0..metrics.num_workers())
+                .map(|i| metrics.worker_steal_count(i))
+                .sum();
 
-            // Blocking receive on the channel.
-            rx.recv().unwrap();
-        })
-        .await
-        .unwrap();
-    });
+            assert_eq!(1, n);
+            return;
+        }
+    }
 
-    drop(rt);
-
-    let n: u64 = (0..metrics.num_workers())
-        .map(|i| metrics.worker_steal_count(i))
-        .sum();
-
-    assert_eq!(1, n);
+    panic!("exhausted every try to schedule the stealable task");
 }
 
 #[test]
@@ -432,6 +424,21 @@ fn log_histogram() {
     assert_eq!(N, n);
 }
 
+#[test]
+#[allow(deprecated)]
+fn legacy_log_histogram() {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .enable_metrics_poll_time_histogram()
+        .metrics_poll_count_histogram_scale(HistogramScale::Log)
+        .metrics_poll_count_histogram_resolution(Duration::from_micros(50))
+        .metrics_poll_count_histogram_buckets(20)
+        .build()
+        .unwrap();
+    let num_buckets = rt.metrics().poll_time_histogram_num_buckets();
+    assert_eq!(num_buckets, 20);
+}
+
 #[test]
 fn log_histogram_default_configuration() {
     let rt = tokio::runtime::Builder::new_current_thread()
@@ -835,6 +842,30 @@ fn io_driver_ready_count() {
     assert_eq!(metrics.io_driver_ready_count(), 1);
 }
 
+async fn try_spawn_stealable_task() -> Result<(), mpsc::RecvTimeoutError> {
+    // We use a blocking channel to synchronize the tasks.
+    let (tx, rx) = mpsc::channel();
+
+    // Make sure we are in the context of the runtime.
+    tokio::spawn(async move {
+        // Spawn the task that sends to the channel.
+        //
+        // Note that the runtime needs to have the lifo slot disabled to make
+        // this task stealable.
+        tokio::spawn(async move {
+            tx.send(()).unwrap();
+        });
+
+        // Blocking receive on the channel, timing out if the sending task
+        // wasn't scheduled in time.
+        rx.recv_timeout(Duration::from_secs(1))
+    })
+    .await
+    .unwrap()?;
+
+    Ok(())
+}
+
 fn current_thread() -> Runtime {
     tokio::runtime::Builder::new_current_thread()
         .enable_all()

tokio/tests/signal_drop_recv.rs

@@ -1,7 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
-#![cfg(not(miri))]
+#![cfg(not(miri))] // No `sigaction` in Miri.
 
 mod support {
     pub mod signal;

tokio/tests/signal_drop_rt.rs

@@ -1,7 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
-#![cfg(not(miri))]
+#![cfg(not(miri))] // No `sigaction` in miri.
 
 mod support {
     pub mod signal;

tokio/tests/signal_drop_signal.rs

@@ -1,7 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
-#![cfg(not(miri))]
+#![cfg(not(miri))] // No `sigaction` in miri.
 
 mod support {
     pub mod signal;