Replace Linux Mutex and Condvar with futex based ones.

library/std/src/sync/condvar/tests.rs

@@ -191,7 +191,7 @@ fn wait_timeout_wake() {
 
 #[test]
 #[should_panic]
-#[cfg_attr(not(unix), ignore)]
+#[cfg(all(unix, not(target_os = "linux"), not(target_os = "android")))]
 fn two_mutexes() {
     let m = Arc::new(Mutex::new(()));
     let m2 = m.clone();

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

@@ -12,7 +12,7 @@ use crate::sync::atomic::AtomicI32;
 use crate::time::Duration;
 
 #[cfg(any(target_os = "linux", target_os = "android"))]
-pub fn futex_wait(futex: &AtomicI32, expected: i32, timeout: Option<Duration>) {
+pub fn futex_wait(futex: &AtomicI32, expected: i32, timeout: Option<Duration>) -> bool {
     let timespec = timeout.and_then(|d| {
         Some(libc::timespec {
             // Sleep forever if the timeout is longer than fits in a timespec.
@@ -21,15 +21,16 @@ pub fn futex_wait(futex: &AtomicI32, expected: i32, timeout: Option<Duration>) {
             tv_nsec: d.subsec_nanos() as _,
         })
     });
-    unsafe {
+    let r = unsafe {
         libc::syscall(
             libc::SYS_futex,
             futex as *const AtomicI32,
             libc::FUTEX_WAIT | libc::FUTEX_PRIVATE_FLAG,
             expected,
             timespec.as_ref().map_or(null(), |d| d as *const libc::timespec),
-        );
-    }
+        )
+    };
+    !(r < 0 && super::os::errno() == libc::ETIMEDOUT)
 }
 
 #[cfg(target_os = "emscripten")]
@@ -65,6 +66,18 @@ pub fn futex_wake(futex: &AtomicI32) {
     }
 }
 
+#[cfg(any(target_os = "linux", target_os = "android"))]
+pub fn futex_wake_all(futex: &AtomicI32) {
+    unsafe {
+        libc::syscall(
+            libc::SYS_futex,
+            futex as *const AtomicI32,
+            libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG,
+            i32::MAX,
+        );
+    }
+}
+
 #[cfg(target_os = "emscripten")]
 pub fn futex_wake(futex: &AtomicI32) {
     extern "C" {

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

@@ -0,0 +1,159 @@
+use crate::sync::atomic::{
+    AtomicI32,
+    Ordering::{Acquire, Relaxed, Release},
+};
+use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
+use crate::time::Duration;
+
+pub type MovableMutex = Mutex;
+pub type MovableCondvar = Condvar;
+
+pub struct Mutex {
+    /// 0: unlocked
+    /// 1: locked, no other threads waiting
+    /// 2: locked, and other threads waiting (contended)
+    futex: AtomicI32,
+}
+
+impl Mutex {
+    pub const fn new() -> Self {
+        Self { futex: AtomicI32::new(0) }
+    }
+
+    #[inline]
+    pub unsafe fn init(&mut self) {}
+
+    #[inline]
+    pub unsafe fn destroy(&self) {}
+
+    #[inline]
+    pub unsafe fn try_lock(&self) -> bool {
+        self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_ok()
+    }
+
+    #[inline]
+    pub unsafe fn lock(&self) {
+        if self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_err() {
+            self.lock_contended();
+        }
+    }
+
+    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 succesfully 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) -> i32 {
+        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);
+
+            if state == 0 || 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.
+            futex_wake(&self.futex);
+        }
+    }
+}
+
+pub struct Condvar {
+    // The value of this atomic is simply incremented on every notification.
+    // This is used by `.wait()` to not miss any notifications after
+    // unlocking the mutex and before waiting for notifications.
+    futex: AtomicI32,
+}
+
+impl Condvar {
+    #[inline]
+    pub const fn new() -> Self {
+        Self { futex: AtomicI32::new(0) }
+    }
+
+    #[inline]
+    pub unsafe fn init(&mut self) {}
+
+    #[inline]
+    pub unsafe fn destroy(&self) {}
+
+    // All the memory orderings here are `Relaxed`,
+    // because synchronization is done by unlocking and locking the mutex.
+
+    #[inline]
+    pub unsafe fn notify_one(&self) {
+        self.futex.fetch_add(1, Relaxed);
+        futex_wake(&self.futex);
+    }
+
+    #[inline]
+    pub unsafe fn notify_all(&self) {
+        self.futex.fetch_add(1, Relaxed);
+        futex_wake_all(&self.futex);
+    }
+
+    #[inline]
+    pub unsafe fn wait(&self, mutex: &Mutex) {
+        self.wait_optional_timeout(mutex, None);
+    }
+
+    #[inline]
+    pub unsafe fn wait_timeout(&self, mutex: &Mutex, timeout: Duration) -> bool {
+        self.wait_optional_timeout(mutex, Some(timeout))
+    }
+
+    unsafe fn wait_optional_timeout(&self, mutex: &Mutex, timeout: Option<Duration>) -> bool {
+        // Check the notification counter before we unlock the mutex.
+        let futex_value = self.futex.load(Relaxed);
+
+        // Unlock the mutex before going to sleep.
+        mutex.unlock();
+
+        // Wait, but only if there hasn't been any
+        // notification since we unlocked the mutex.
+        let r = futex_wait(&self.futex, futex_value, timeout);
+
+        // Lock the mutex again.
+        mutex.lock();
+
+        r
+    }
+}

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

@@ -1,8 +1,24 @@
-mod pthread_condvar;
-mod pthread_mutex;
-mod pthread_remutex;
-mod pthread_rwlock;
-pub use pthread_condvar::{Condvar, MovableCondvar};
-pub use pthread_mutex::{MovableMutex, Mutex};
-pub use pthread_remutex::ReentrantMutex;
-pub use pthread_rwlock::{MovableRWLock, RWLock};
+cfg_if::cfg_if! {
+    if #[cfg(any(
+        target_os = "linux",
+        target_os = "android",
+    ))] {
+        mod futex;
+        #[allow(dead_code)]
+        mod pthread_mutex; // Only used for PthreadMutexAttr, needed by pthread_remutex.
+        mod pthread_remutex; // FIXME: Implement this using a futex
+        mod pthread_rwlock; // FIXME: Implement this using a futex
+        pub use futex::{Mutex, MovableMutex, Condvar, MovableCondvar};
+        pub use pthread_remutex::ReentrantMutex;
+        pub use pthread_rwlock::{RWLock, MovableRWLock};
+    } else {
+        mod pthread_mutex;
+        mod pthread_remutex;
+        mod pthread_rwlock;
+        mod pthread_condvar;
+        pub use pthread_mutex::{Mutex, MovableMutex};
+        pub use pthread_remutex::ReentrantMutex;
+        pub use pthread_rwlock::{RWLock, MovableRWLock};
+        pub use pthread_condvar::{Condvar, MovableCondvar};
+    }
+}