Auto merge of #121956 - ChrisDenton:srwlock, r=joboet

Windows: Implement condvar, mutex and rwlock using futex

Well, the Windows equivalent: [`WaitOnAddress`,](https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-waitonaddress) [`WakeByAddressSingle`](https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-wakebyaddresssingle) and [`WakeByAddressAll`](https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-wakebyaddressall).

Note that Windows flavoured futexes can be different sizes (1, 2, 4 or 8 bytes). I took advantage of that in the `Mutex` implementation.

I also edited the Mutex implementation a bit more than necessary. I was having trouble keeping in my head what 0, 1 and 2 meant so I replaced them with consts.

I *think* we're maybe spinning a bit much. `WaitOnAddress` seems to be looping quite a bit too. But for now I've keep the implementations the same. I do wonder if it'd be worth reducing or removing our spinning on Windows.

This also adds a new shim to miri, because of course it does.

Fixes #121949

library/std/src/sys/locks/condvar/mod.rs

@@ -1,5 +1,6 @@
 cfg_if::cfg_if! {
     if #[cfg(any(
+        all(target_os = "windows", not(target_vendor="win7")),
         target_os = "linux",
         target_os = "android",
         target_os = "freebsd",
@@ -14,9 +15,9 @@
     } else if #[cfg(target_family = "unix")] {
         mod pthread;
         pub use pthread::Condvar;
-    } else if #[cfg(target_os = "windows")] {
-        mod windows;
-        pub use windows::Condvar;
+    } else if #[cfg(all(target_os = "windows", target_vendor = "win7"))] {
+        mod windows7;
+        pub use windows7::Condvar;
     } else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
         mod sgx;
         pub use sgx::Condvar;

library/std/src/sys/locks/mutex/futex.rs

@@ -1,30 +1,42 @@
 use crate::sync::atomic::{
-    AtomicU32,
+    self,
     Ordering::{Acquire, Relaxed, Release},
 };
 use crate::sys::futex::{futex_wait, futex_wake};
 
-pub struct Mutex {
-    /// 0: unlocked
-    /// 1: locked, no other threads waiting
-    /// 2: locked, and other threads waiting (contended)
-    futex: AtomicU32,
+cfg_if::cfg_if! {
+if #[cfg(windows)] {
+    // On Windows we can have a smol futex
+    type Atomic = atomic::AtomicU8;
+    type State = u8;
+} else {
+    type Atomic = atomic::AtomicU32;
+    type State = u32;
 }
+}
+
+pub struct Mutex {
+    futex: Atomic,
+}
+
+const UNLOCKED: State = 0;
+const LOCKED: State = 1; // locked, no other threads waiting
+const CONTENDED: State = 2; // locked, and other threads waiting (contended)
 
 impl Mutex {
     #[inline]
     pub const fn new() -> Self {
-        Self { futex: AtomicU32::new(0) }
+        Self { futex: Atomic::new(UNLOCKED) }
     }
 
     #[inline]
     pub fn try_lock(&self) -> bool {
-        self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_ok()
+        self.futex.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed).is_ok()
     }
 
     #[inline]
     pub fn lock(&self) {
-        if self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_err() {
+        if self.futex.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed).is_err() {
             self.lock_contended();
         }
     }
@@ -36,8 +48,8 @@ fn lock_contended(&self) {
 
         // If it's unlocked now, attempt to take the lock
         // without marking it as contended.
-        if state == 0 {
-            match self.futex.compare_exchange(0, 1, Acquire, Relaxed) {
+        if state == UNLOCKED {
+            match self.futex.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed) {
                 Ok(_) => return, // Locked!
                 Err(s) => state = s,
             }
@@ -45,31 +57,31 @@ fn lock_contended(&self) {
 
         loop {
             // Put the lock in contended state.
-            // We avoid an unnecessary write if it as already set to 2,
+            // We avoid an unnecessary write if it as already set to CONTENDED,
             // to be friendlier for the caches.
-            if state != 2 && self.futex.swap(2, Acquire) == 0 {
-                // We changed it from 0 to 2, so we just successfully locked it.
+            if state != CONTENDED && self.futex.swap(CONTENDED, Acquire) == UNLOCKED {
+                // We changed it from UNLOCKED to CONTENDED, so we just successfully locked it.
                 return;
             }
 
-            // Wait for the futex to change state, assuming it is still 2.
-            futex_wait(&self.futex, 2, None);
+            // Wait for the futex to change state, assuming it is still CONTENDED.
+            futex_wait(&self.futex, CONTENDED, None);
 
             // Spin again after waking up.
             state = self.spin();
         }
     }
 
-    fn spin(&self) -> u32 {
+    fn spin(&self) -> State {
         let mut spin = 100;
         loop {
             // We only use `load` (and not `swap` or `compare_exchange`)
             // while spinning, to be easier on the caches.
             let state = self.futex.load(Relaxed);
 
-            // We stop spinning when the mutex is unlocked (0),
-            // but also when it's contended (2).
-            if state != 1 || spin == 0 {
+            // We stop spinning when the mutex is UNLOCKED,
+            // but also when it's CONTENDED.
+            if state != LOCKED || spin == 0 {
                 return state;
             }
 
@@ -80,9 +92,9 @@ fn spin(&self) -> u32 {
 
     #[inline]
     pub unsafe fn unlock(&self) {
-        if self.futex.swap(0, Release) == 2 {
+        if self.futex.swap(UNLOCKED, Release) == CONTENDED {
             // We only wake up one thread. When that thread locks the mutex, it
-            // will mark the mutex as contended (2) (see lock_contended above),
+            // will mark the mutex as CONTENDED (see lock_contended above),
             // which makes sure that any other waiting threads will also be
             // woken up eventually.
             self.wake();

library/std/src/sys/locks/mutex/mod.rs

@@ -1,5 +1,6 @@
 cfg_if::cfg_if! {
     if #[cfg(any(
+        all(target_os = "windows", not(target_vendor = "win7")),
         target_os = "linux",
         target_os = "android",
         target_os = "freebsd",
@@ -19,9 +20,9 @@
     ))] {
         mod pthread;
         pub use pthread::{Mutex, raw};
-    } else if #[cfg(target_os = "windows")] {
-        mod windows;
-        pub use windows::{Mutex, raw};
+    } else if #[cfg(all(target_os = "windows", target_vendor = "win7"))] {
+        mod windows7;
+        pub use windows7::{Mutex, raw};
     } else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
         mod sgx;
         pub use sgx::Mutex;

library/std/src/sys/locks/rwlock/mod.rs

@@ -1,5 +1,6 @@
 cfg_if::cfg_if! {
     if #[cfg(any(
+        all(target_os = "windows", not(target_vendor = "win7")),
         target_os = "linux",
         target_os = "android",
         target_os = "freebsd",
@@ -14,9 +15,9 @@
     } else if #[cfg(target_family = "unix")] {
         mod queue;
         pub use queue::RwLock;
-    } else if #[cfg(target_os = "windows")] {
-        mod windows;
-        pub use windows::RwLock;
+    } else if #[cfg(all(target_os = "windows", target_vendor = "win7"))] {
+        mod windows7;
+        pub use windows7::RwLock;
     } else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
         mod sgx;
         pub use sgx::RwLock;

library/std/src/sys/pal/windows/c.rs

@@ -36,6 +36,7 @@
 pub type LPWCH = *mut WCHAR;
 pub type LPWSTR = *mut WCHAR;
 
+#[cfg(target_vendor = "win7")]
 pub type PSRWLOCK = *mut SRWLOCK;
 
 pub type socklen_t = c_int;
@@ -50,7 +51,9 @@
 pub const EXIT_SUCCESS: u32 = 0;
 pub const EXIT_FAILURE: u32 = 1;
 
+#[cfg(target_vendor = "win7")]
 pub const CONDITION_VARIABLE_INIT: CONDITION_VARIABLE = CONDITION_VARIABLE { Ptr: ptr::null_mut() };
+#[cfg(target_vendor = "win7")]
 pub const SRWLOCK_INIT: SRWLOCK = SRWLOCK { Ptr: ptr::null_mut() };
 pub const INIT_ONCE_STATIC_INIT: INIT_ONCE = INIT_ONCE { Ptr: ptr::null_mut() };
 
@@ -373,6 +376,7 @@ pub fn WaitOnAddress(
         dwmilliseconds: u32,
     ) -> BOOL;
     pub fn WakeByAddressSingle(address: *const c_void);
+    pub fn WakeByAddressAll(address: *const c_void);
 }
 
 #[cfg(target_vendor = "win7")]

library/std/src/sys/pal/windows/futex.rs

@@ -0,0 +1,85 @@
+use super::api;
+use crate::sys::c;
+use crate::sys::dur2timeout;
+use core::ffi::c_void;
+use core::mem;
+use core::ptr;
+use core::sync::atomic::{
+    AtomicBool, AtomicI16, AtomicI32, AtomicI64, AtomicI8, AtomicIsize, AtomicPtr, AtomicU16,
+    AtomicU32, AtomicU64, AtomicU8, AtomicUsize,
+};
+use core::time::Duration;
+
+pub unsafe trait Waitable {
+    type Atomic;
+}
+macro_rules! unsafe_waitable_int {
+    ($(($int:ty, $atomic:ty)),*$(,)?) => {
+        $(
+            unsafe impl Waitable for $int {
+                type Atomic = $atomic;
+            }
+        )*
+    };
+}
+unsafe_waitable_int! {
+    (bool, AtomicBool),
+    (i8, AtomicI8),
+    (i16, AtomicI16),
+    (i32, AtomicI32),
+    (i64, AtomicI64),
+    (isize, AtomicIsize),
+    (u8, AtomicU8),
+    (u16, AtomicU16),
+    (u32, AtomicU32),
+    (u64, AtomicU64),
+    (usize, AtomicUsize),
+}
+unsafe impl<T> Waitable for *const T {
+    type Atomic = AtomicPtr<T>;
+}
+unsafe impl<T> Waitable for *mut T {
+    type Atomic = AtomicPtr<T>;
+}
+
+pub fn wait_on_address<W: Waitable>(
+    address: &W::Atomic,
+    compare: W,
+    timeout: Option<Duration>,
+) -> bool {
+    unsafe {
+        let addr = ptr::from_ref(address).cast::<c_void>();
+        let size = mem::size_of::<W>();
+        let compare_addr = ptr::addr_of!(compare).cast::<c_void>();
+        let timeout = timeout.map(dur2timeout).unwrap_or(c::INFINITE);
+        c::WaitOnAddress(addr, compare_addr, size, timeout) == c::TRUE
+    }
+}
+
+pub fn wake_by_address_single<T>(address: &T) {
+    unsafe {
+        let addr = ptr::from_ref(address).cast::<c_void>();
+        c::WakeByAddressSingle(addr);
+    }
+}
+
+pub fn wake_by_address_all<T>(address: &T) {
+    unsafe {
+        let addr = ptr::from_ref(address).cast::<c_void>();
+        c::WakeByAddressAll(addr);
+    }
+}
+
+pub fn futex_wait<W: Waitable>(futex: &W::Atomic, expected: W, timeout: Option<Duration>) -> bool {
+    // return false only on timeout
+    wait_on_address(futex, expected, timeout) || api::get_last_error().code != c::ERROR_TIMEOUT
+}
+
+pub fn futex_wake<T>(futex: &T) -> bool {
+    wake_by_address_single(futex);
+    false
+}
+
+pub fn futex_wake_all<T>(futex: &T) {
+    wake_by_address_all(futex)
+}

library/std/src/sys/pal/windows/mod.rs

@@ -17,6 +17,8 @@
 pub mod c;
 pub mod env;
 pub mod fs;
+#[cfg(not(target_vendor = "win7"))]
+pub mod futex;
 pub mod handle;
 pub mod io;
 pub mod net;

src/tools/miri/src/shims/windows/foreign_items.rs

@@ -366,6 +366,12 @@ fn emulate_foreign_item_inner(
 
                 this.WakeByAddressSingle(ptr_op)?;
             }
+            "WakeByAddressAll" => {
+                let [ptr_op] =
+                    this.check_shim(abi, Abi::System { unwind: false }, link_name, args)?;
+
+                this.WakeByAddressAll(ptr_op)?;
+            }
 
             // Dynamic symbol loading
             "GetProcAddress" => {

src/tools/miri/src/shims/windows/sync.rs

@@ -384,6 +384,21 @@ fn WakeByAddressSingle(&mut self, ptr_op: &OpTy<'tcx, Provenance>) -> InterpResu
 
         Ok(())
     }
+    fn WakeByAddressAll(&mut self, ptr_op: &OpTy<'tcx, Provenance>) -> InterpResult<'tcx> {
+        let this = self.eval_context_mut();
+
+        let ptr = this.read_pointer(ptr_op)?;
+
+        // See the Linux futex implementation for why this fence exists.
+        this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+        while let Some(thread) = this.futex_wake(ptr.addr().bytes(), u32::MAX) {
+            this.unblock_thread(thread);
+            this.unregister_timeout_callback_if_exists(thread);
+        }
+
+        Ok(())
+    }
 
     fn SleepConditionVariableSRW(
         &mut self,

tests/debuginfo/mutex.rs

@@ -10,7 +10,7 @@
 //
 // cdb-command:dx m,d
 // cdb-check:m,d              [Type: std::sync::mutex::Mutex<i32>]
-// cdb-check:    [...] inner            [Type: std::sys::locks::mutex::windows::Mutex]
+// cdb-check:    [...] inner            [Type: std::sys::locks::mutex::futex::Mutex]
 // cdb-check:    [...] poison           [Type: std::sync::poison::Flag]
 // cdb-check:    [...] data             : 0 [Type: core::cell::UnsafeCell<i32>]
 

tests/debuginfo/rwlock-read.rs

@@ -16,7 +16,7 @@
 // cdb-command:dx r
 // cdb-check:r                [Type: std::sync::rwlock::RwLockReadGuard<i32>]
 // cdb-check:    [...] data             : NonNull([...]: 0) [Type: core::ptr::non_null::NonNull<i32>]
-// cdb-check:    [...] inner_lock       : [...] [Type: std::sys::locks::rwlock::windows::RwLock *]
+// cdb-check:    [...] inner_lock       : [...] [Type: std::sys::locks::rwlock::futex::RwLock *]
 
 #[allow(unused_variables)]