Merge remote-tracking branch 'upstream/master' into tokio-usdt

Author: conradludgate

.github/labeler.yml

@@ -19,3 +19,7 @@ R-loom-multi-thread:
 - tokio/src/runtime/scheduler/multi_thread/**
 - tokio/src/runtime/task/*
 - tokio/src/runtime/task/**
+
+R-loom-util:
+- tokio-util/src/*
+- tokio-util/src/**/*

.github/workflows/ci.yml

@@ -18,7 +18,7 @@ env:
   rust_stable: stable
   rust_nightly: nightly-2025-10-12
   # Pin a specific miri version
-  rust_miri_nightly: nightly-2025-06-02
+  rust_miri_nightly: nightly-2025-11-09
   rust_clippy: '1.88'
   # When updating this, also update:
   # - README.md

.github/workflows/loom.yml

@@ -95,3 +95,19 @@ jobs:
         working-directory: tokio
         env:
           SCOPE: ${{ matrix.scope }}
+
+  loom-util:
+    name: loom tokio-util
+    # base_ref is null when it's not a pull request
+    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-util') || (github.base_ref == null))
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v5
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+            toolchain: ${{ env.rust_stable }}
+      - uses: Swatinem/rust-cache@v2
+      - name: run tests
+        run: cargo test --lib --release --features full -- --nocapture
+        working-directory: tokio-util

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -33,6 +33,12 @@ impl<S: Stream> ChunksTimeout<S> {
             cap: max_size,
         }
     }
+
+    /// Consumes the [`ChunksTimeout`] and then returns all buffered items.
+    pub fn into_remainder(mut self: Pin<&mut Self>) -> Vec<S::Item> {
+        let me = self.as_mut().project();
+        std::mem::take(me.items)
+    }
 }
 
 impl<S: Stream> Stream for ChunksTimeout<S> {

tokio-stream/src/wrappers.rs

@@ -22,9 +22,9 @@ cfg_sync! {
 }
 
 cfg_signal! {
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     mod signal_unix;
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     pub use signal_unix::SignalStream;
 
     #[cfg(any(windows, docsrs))]
@@ -39,12 +39,14 @@ cfg_time! {
 }
 
 cfg_net! {
+    #[cfg(not(loom))]
     mod tcp_listener;
+    #[cfg(not(loom))]
     pub use tcp_listener::TcpListenerStream;
 
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     mod unix_listener;
-    #[cfg(unix)]
+    #[cfg(all(unix, not(loom)))]
     pub use unix_listener::UnixListenerStream;
 }
 
@@ -57,6 +59,8 @@ cfg_io_util! {
 }
 
 cfg_fs! {
+    #[cfg(not(loom))]
     mod read_dir;
+    #[cfg(not(loom))]
     pub use read_dir::ReadDirStream;
 }

tokio-stream/tests/stream_chunks_timeout.rs

@@ -0,0 +1,30 @@
+#![warn(rust_2018_idioms)]
+
+use futures::FutureExt;
+use std::error::Error;
+use tokio::time;
+use tokio::time::Duration;
+use tokio_stream::{self as stream, StreamExt};
+use tokio_test::assert_pending;
+use tokio_test::task;
+
+#[tokio::test(start_paused = true)]
+async fn stream_chunks_remainder() -> Result<(), Box<dyn Error>> {
+    let stream1 =
+        stream::iter([5]).then(move |n| time::sleep(Duration::from_secs(1)).map(move |_| n));
+
+    let inner = stream::iter([1, 2, 3, 4]).chain(stream1);
+    tokio::pin!(inner);
+
+    let chunked = (&mut inner).chunks_timeout(10, Duration::from_millis(20));
+
+    let mut chunked = task::spawn(chunked);
+    assert_pending!(chunked.poll_next());
+
+    let remainder = chunked.enter(|_, stream| stream.into_remainder());
+
+    assert_eq!(remainder, vec![1, 2, 3, 4]);
+    time::advance(Duration::from_secs(2)).await;
+    assert_eq!(inner.next().await, Some(5));
+    Ok(())
+}

tokio-util/CHANGELOG.md

@@ -1,3 +1,33 @@
+# 0.7.17 (November 2nd, 2025)
+
+The MSRV is increased to 1.71.
+
+### Added
+
+- codec: add `{FramedRead,FramedWrite}::into_parts()` ([#7566])
+- time: add `#[track_caller]` to `FutureExt::timeout` ([#7588])
+- task: add `tokio_util::task::JoinQueue` ([#7590])
+
+### Changed
+
+- codec: remove unnecessary trait bounds on all Framed constructors ([#7716])
+
+### Documented
+
+- time: clarify the cancellation safety of the `DelayQueue` ([#7564])
+- docs: fix some docs links ([#7654])
+- task: simplify the example of `TaskTracker` ([#7657])
+- task: clarify the behavior of several `spawn_local` methods ([#7669])
+
+[#7564]: https://github.com/tokio-rs/tokio/pull/7564
+[#7566]: https://github.com/tokio-rs/tokio/pull/7566
+[#7588]: https://github.com/tokio-rs/tokio/pull/7588
+[#7590]: https://github.com/tokio-rs/tokio/pull/7590
+[#7654]: https://github.com/tokio-rs/tokio/pull/7654
+[#7657]: https://github.com/tokio-rs/tokio/pull/7657
+[#7669]: https://github.com/tokio-rs/tokio/pull/7669
+[#7716]: https://github.com/tokio-rs/tokio/pull/7716
+
 # 0.7.16 (August 3rd, 2025)
 
 ### Added

tokio-util/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-util"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.7.x" git tag.
-version = "0.7.16"
+version = "0.7.17"
 edition = "2021"
 rust-version = "1.71"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -57,6 +57,9 @@ futures-test = "0.3.5"
 parking_lot = "0.12.0"
 tempfile = "3.1.0"
 
+[target.'cfg(loom)'.dev-dependencies]
+loom = { version = "0.7", features = ["futures", "checkpoint"] }
+
 [package.metadata.docs.rs]
 all-features = true
 # enable unstable features in the documentation

tokio-util/src/either.rs

@@ -212,7 +212,7 @@ where
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, not(loom)))]
 mod tests {
     use super::*;
     use tokio::io::{repeat, AsyncReadExt, Repeat};

tokio-util/src/loom.rs

@@ -1 +1,9 @@
-pub(crate) use std::sync;
+//! This module abstracts over `loom` and `std::sync` types depending on whether we
+//! are running loom tests or not.
+
+pub(crate) mod sync {
+    #[cfg(all(test, loom))]
+    pub(crate) use loom::sync::{Arc, Mutex, MutexGuard};
+    #[cfg(not(all(test, loom)))]
+    pub(crate) use std::sync::{Arc, Mutex, MutexGuard};
+}