tokio-util: enable loom testing for cancellation token

.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-sync:
+  - tokio-util/src/sync/*
+  - tokio-util/src/sync/**/*

.github/workflows/loom.yml

@@ -95,3 +95,19 @@ jobs:
         working-directory: tokio
         env:
           SCOPE: ${{ matrix.scope }}
+
+  loom-util-sync:
+    name: loom tokio-util::sync
+    # 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-sync') || (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 sync::tests
+        working-directory: tokio-util

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

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

tokio-util/src/net/mod.rs

@@ -1,3 +1,5 @@
+#![cfg(not(loom))]
+
 //! TCP/UDP/Unix helpers for tokio.
 
 use crate::either::Either;

tokio-util/src/sync/mod.rs

@@ -15,3 +15,6 @@ pub use poll_semaphore::PollSemaphore;
 
 mod reusable_box;
 pub use reusable_box::ReusableBoxFuture;
+
+#[cfg(test)]
+mod tests;

tokio-util/src/sync/tests/mod.rs

@@ -1 +1,2 @@
-
+#[cfg(loom)]
+mod loom_cancellation_token;

tokio-util/src/udp/mod.rs

@@ -1,3 +1,5 @@
+#![cfg(not(loom))]
+
 //! UDP framing
 
 mod frame;

tokio-util/tests/udp.rs

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