Merge pull request #1 from richkcho/ricch/fix_timeout

Fix bug in windows wait, add tests, update CI to use cargo nextest

Author: DouglasDwyer

.config/nextest.toml

@@ -0,0 +1,2 @@
+[profile.default]
+slow-timeout = { period = "10s", terminate-after = 2, grace-period = "10s" }

.github/workflows/rust.yml

@@ -3,36 +3,27 @@ on:
   push: { branches: "main" }
   pull_request: { branches: "*" }
 jobs:
-  check-fmt:
-    name: Check formatting
-    runs-on: ubuntu-latest
+  build-and-test:
+    strategy:
+      matrix:
+        os: [ubuntu-latest, ubuntu-22.04-arm, windows-latest, windows-11-arm, macos-latest]
+    runs-on: ${{ matrix.os }}
     steps:
-    - uses: actions/checkout@v1
-    - run: cargo fmt -- --check
-  linux:
-    name: Build and test on Linux
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: Build
-      run: cargo build --verbose
-    - name: Run tests
-      run: cargo test --verbose
-  windows:
-    name: Build and test on Windows
-    runs-on: windows-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: Build
-      run: cargo build --verbose
-    - name: Run tests
-      run: cargo test --verbose
-  macos:
-    name: Build and test on macOS
-    runs-on: macos-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: Build
-      run: cargo build --verbose
-    - name: Run tests
-      run: cargo test --verbose
+      - uses: actions/checkout@v4
+      - name: Install toolchain components
+        run: rustup component add rustfmt clippy
+      - name: Install cargo-binstall
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-binstall
+      - name: Install cargo-nextest
+        run: cargo binstall cargo-nextest --secure --no-confirm
+      - name: Check formatting
+        run: cargo fmt --all -- --check
+      - name: Run clippy
+        run: cargo clippy --all-targets -- -D warnings
+      - name: Build
+        run: cargo build --verbose
+      - name: Run tests
+        run: cargo nextest run --verbose
+

Cargo.toml

@@ -12,7 +12,7 @@ categories = ["concurrency", "os", "no-std"]
 libc = { version = "0.2", default-features = false }
 
 [target.'cfg(windows)'.dependencies]
-windows-sys = { version = "0.59.0", default-features = false, features = ["Win32_System_Threading", "Win32_Foundation"] }
+windows-sys = { version = "0.61.0", default-features = false, features = ["Win32_System_Threading", "Win32_Foundation"] }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 js-sys = { version = "0.3.67", default-features = false }

src/windows.rs

@@ -18,7 +18,13 @@ impl AtomicWaitImpl for AtomicU32 {
                 &value as *const _ as *const _,
                 size_of::<Self>(),
                 timeout
-                    .map(|x| x.as_millis().max(u64::MAX as u128) as u32)
+                    .map(|x| {
+                        // Clamp to a finite u32 millisecond timeout. INFINITE (0xFFFFFFFF)
+                        // means no timeout, so avoid ever passing that when a timeout is set.
+                        let ms = x.as_millis();
+                        let capped = ms.min(u32::MAX as u128 - 1);
+                        capped as u32
+                    })
                     .unwrap_or(INFINITE),
             );
         }
@@ -43,7 +49,7 @@ impl AtomicWaitImpl for AtomicU64 {
                 &value as *const _ as *const _,
                 size_of::<Self>(),
                 timeout
-                    .map(|x| x.as_millis().max(u64::MAX as u128) as u32)
+                    .map(|x| x.as_millis().min(u32::MAX as u128 - 1) as u32)
                     .unwrap_or(INFINITE),
             );
         }

tests/test.rs

@@ -37,3 +37,71 @@ fn wait_wake() {
         assert!((90..400).contains(&t.elapsed().as_millis()));
     });
 }
+
+#[test]
+fn wait_timeout() {
+    let a = AtomicU32::new(0);
+    a.wait_timeout(0, Duration::from_millis(1));
+}
+
+#[test]
+fn stress_many_waiters_notify_all() {
+    use std::sync::Arc;
+    let a = Arc::new(AtomicU32::new(0));
+    let woke = Arc::new(AtomicU32::new(0));
+
+    let threads = 64;
+    std::thread::scope(|s| {
+        for _ in 0..threads {
+            let a = a.clone();
+            let woke = woke.clone();
+            s.spawn(move || {
+                while a.load(Relaxed) == 0 {
+                    a.wait(0);
+                }
+                woke.fetch_add(1, Relaxed);
+            });
+        }
+
+        // Give threads time to start waiting
+        sleep(Duration::from_millis(50));
+        a.store(1, Relaxed);
+        a.notify_all();
+    });
+
+    assert_eq!(woke.load(Relaxed), threads);
+}
+
+#[test]
+fn stress_ping_pong_many_iters() {
+    use std::sync::Arc;
+    let state = Arc::new(AtomicU32::new(0));
+    let iters = 5_000u32;
+
+    std::thread::scope(|s| {
+        let state_c = state.clone();
+        s.spawn(move || {
+            // Consumer: wait for 1, reset to 0, and notify producer.
+            for _ in 0..iters {
+                while state_c.load(Relaxed) != 1 {
+                    // Wait while the state is 0; use a short timeout to be resilient to spurious wakes.
+                    state_c.wait_timeout(0, Duration::from_millis(10));
+                }
+                state_c.store(0, Relaxed);
+                state_c.notify_one();
+            }
+        });
+
+        // Producer: set to 1, notify consumer, then wait until it resets to 0.
+        for _ in 0..iters {
+            state.store(1, Relaxed);
+            state.notify_one();
+            while state.load(Relaxed) != 0 {
+                state.wait_timeout(1, Duration::from_millis(10));
+            }
+        }
+    });
+
+    // Final state should be 0 after a complete ping-pong.
+    assert_eq!(state.load(Relaxed), 0);
+}