Windows: Implement mutex using futex

Well, the Windows equivalent: `WaitOnAddress`, `WakeByAddressSingle` and `WakeByAddressAll`.

Author: ChrisDenton

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

@@ -1,4 +1,10 @@
 use crate::sync::atomic::{AtomicU32, Ordering::Relaxed};
+#[cfg(windows)]
+use crate::sys::api::{
+    wait_on_address as futex_wait, wake_by_address_all as futex_wake_all,
+    wake_by_address_single as futex_wake,
+};
+#[cfg(not(windows))]
 use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
 use crate::sys::locks::Mutex;
 use crate::time::Duration;

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 @@ cfg_if::cfg_if! {
     } 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/condvar/windows.rs renamed to library/std/src/sys/locks/condvar/windows7.rs

@@ -1,96 +1,9 @@
-use crate::sync::atomic::{
-    AtomicU32,
-    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,
-}
-
-impl Mutex {
-    #[inline]
-    pub const fn new() -> Self {
-        Self { futex: AtomicU32::new(0) }
-    }
-
-    #[inline]
-    pub fn try_lock(&self) -> bool {
-        self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_ok()
-    }
-
-    #[inline]
-    pub fn lock(&self) {
-        if self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_err() {
-            self.lock_contended();
-        }
-    }
-
-    #[cold]
-    fn lock_contended(&self) {
-        // Spin first to speed things up if the lock is released quickly.
-        let mut state = self.spin();
-
-        // 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) {
-                Ok(_) => return, // Locked!
-                Err(s) => state = s,
-            }
-        }
-
-        loop {
-            // Put the lock in contended state.
-            // We avoid an unnecessary write if it as already set to 2,
-            // 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.
-                return;
-            }
-
-            // Wait for the futex to change state, assuming it is still 2.
-            futex_wait(&self.futex, 2, None);
-
-            // Spin again after waking up.
-            state = self.spin();
-        }
-    }
-
-    fn spin(&self) -> u32 {
-        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 {
-                return state;
-            }
-
-            crate::hint::spin_loop();
-            spin -= 1;
-        }
-    }
-
-    #[inline]
-    pub unsafe fn unlock(&self) {
-        if self.futex.swap(0, Release) == 2 {
-            // We only wake up one thread. When that thread locks the mutex, it
-            // will mark the mutex as contended (2) (see lock_contended above),
-            // which makes sure that any other waiting threads will also be
-            // woken up eventually.
-            self.wake();
-        }
-    }
-
-    #[cold]
-    fn wake(&self) {
-        futex_wake(&self.futex);
+cfg_if::cfg_if! {
+    if #[cfg(windows)] {
+        mod windows;
+        pub use windows::*;
+    } else {
+        mod unix;
+        pub use unix::*;
     }
 }

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

@@ -0,0 +1,96 @@
+use crate::sync::atomic::{
+    AtomicU32,
+    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,
+}
+
+impl Mutex {
+    #[inline]
+    pub const fn new() -> Self {
+        Self { futex: AtomicU32::new(0) }
+    }
+
+    #[inline]
+    pub fn try_lock(&self) -> bool {
+        self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_ok()
+    }
+
+    #[inline]
+    pub fn lock(&self) {
+        if self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_err() {
+            self.lock_contended();
+        }
+    }
+
+    #[cold]
+    fn lock_contended(&self) {
+        // Spin first to speed things up if the lock is released quickly.
+        let mut state = self.spin();
+
+        // 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) {
+                Ok(_) => return, // Locked!
+                Err(s) => state = s,
+            }
+        }
+
+        loop {
+            // Put the lock in contended state.
+            // We avoid an unnecessary write if it as already set to 2,
+            // 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.
+                return;
+            }
+
+            // Wait for the futex to change state, assuming it is still 2.
+            futex_wait(&self.futex, 2, None);
+
+            // Spin again after waking up.
+            state = self.spin();
+        }
+    }
+
+    fn spin(&self) -> u32 {
+        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 {
+                return state;
+            }
+
+            crate::hint::spin_loop();
+            spin -= 1;
+        }
+    }
+
+    #[inline]
+    pub unsafe fn unlock(&self) {
+        if self.futex.swap(0, Release) == 2 {
+            // We only wake up one thread. When that thread locks the mutex, it
+            // will mark the mutex as contended (2) (see lock_contended above),
+            // which makes sure that any other waiting threads will also be
+            // woken up eventually.
+            self.wake();
+        }
+    }
+
+    #[cold]
+    fn wake(&self) {
+        futex_wake(&self.futex);
+    }
+}

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

@@ -0,0 +1,65 @@
+use crate::sync::atomic::{
+    AtomicI8,
+    Ordering::{Acquire, Relaxed, Release},
+};
+use crate::sys::api;
+
+pub struct Mutex {
+    state: AtomicI8,
+}
+
+const UNLOCKED: i8 = 0;
+const LOCKED: i8 = 1;
+const CONTENDED: i8 = 2;
+
+impl Mutex {
+    #[inline]
+    pub const fn new() -> Mutex {
+        Mutex { state: AtomicI8::new(UNLOCKED) }
+    }
+
+    #[inline]
+    pub fn lock(&self) {
+        if let Err(state) = self.state.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed) {
+            self.lock_contended(state)
+        }
+    }
+
+    #[cold]
+    fn lock_contended(&self, mut state: i8) {
+        // Note: WaitOnAddress is already quite spin-happy so we don't do any further spinning on top.
+        loop {
+            // Put the lock in contended state.
+            // We avoid an unnecessary write if it as already set to CONTENDED,
+            // to be friendlier for the caches.
+            if state != CONTENDED && self.state.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 CONTENDED.
+            api::wait_on_address(&self.state, CONTENDED, None);
+            state = self.state.load(Relaxed);
+        }
+    }
+
+    #[inline]
+    pub fn try_lock(&self) -> bool {
+        self.state.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed).is_ok()
+    }
+
+    #[inline]
+    pub unsafe fn unlock(&self) {
+        if self.state.swap(UNLOCKED, Release) == CONTENDED {
+            // We only wake up one thread. When that thread locks the mutex, it
+            // 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();
+        }
+    }
+
+    #[cold]
+    fn wake(&self) {
+        api::wake_by_address_single(&self.state);
+    }
+}

library/std/src/sys/locks/mutex/futex/windows.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 @@ cfg_if::cfg_if! {
     ))] {
         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/mutex/mod.rs

@@ -2,6 +2,12 @@ use crate::sync::atomic::{
     AtomicU32,
     Ordering::{Acquire, Relaxed, Release},
 };
+#[cfg(windows)]
+use crate::sys::api::{
+    wait_on_address as futex_wait, wake_by_address_all as futex_wake_all,
+    wake_by_address_single as futex_wake,
+};
+#[cfg(not(windows))]
 use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
 
 pub struct RwLock {
@@ -291,7 +297,10 @@ impl RwLock {
     }
 
     /// Spin for a while, but stop directly at the given condition.
+    ///
+    /// We avoid spinning on Windows because the futex implementation spins enough.
     #[inline]
+    #[cfg(not(windows))]
     fn spin_until(&self, f: impl Fn(u32) -> bool) -> u32 {
         let mut spin = 100; // Chosen by fair dice roll.
         loop {
@@ -304,6 +313,12 @@ impl RwLock {
         }
     }
 
+    #[inline]
+    #[cfg(windows)]
+    fn spin_until(&self, _f: impl Fn(u32) -> bool) -> u32 {
+        self.state.load(Relaxed)
+    }
+
     #[inline]
     fn spin_write(&self) -> u32 {
         // Stop spinning when it's unlocked or when there's waiting writers, to keep things somewhat fair.

library/std/src/sys/locks/mutex/windows.rs renamed to library/std/src/sys/locks/mutex/windows7.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 @@ cfg_if::cfg_if! {
     } 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;