Implement condvar using futex requeue in Linux

Author: ruihe774

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

@@ -115,6 +115,17 @@ pub fn futex_wake_all(futex: &AtomicU32) {
     }
 }
 
+/// Wakes one waiter on `futex` and requeue the remaining waiters to `futex2`.
+#[cfg(any(target_os = "linux", target_os = "android"))]
+pub fn futex_requeue(futex: &AtomicU32, futex2: &AtomicU32) {
+    let ptr = futex as *const AtomicU32;
+    let ptr2 = futex2 as *const AtomicU32;
+    let op = libc::FUTEX_REQUEUE | libc::FUTEX_PRIVATE_FLAG;
+    unsafe {
+        libc::syscall(libc::SYS_futex, ptr, op, 1, i32::MAX, ptr2);
+    }
+}
+
 // FreeBSD doesn't tell us how many threads are woken up, so this always returns false.
 #[cfg(target_os = "freebsd")]
 pub fn futex_wake(futex: &AtomicU32) -> bool {

library/std/src/sys/sync/condvar/futex_requeue.rs

@@ -0,0 +1,65 @@
+use crate::sync::atomic::Ordering::Relaxed;
+use crate::sys::futex::{Futex, futex_requeue, futex_wait, futex_wake};
+use crate::sys::sync::Mutex;
+use crate::time::Duration;
+
+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: Futex,
+    // The futex to requeue to.
+    // Its value is simply incremented when a waiter has been woken and acquired mutex.
+    futex2: Futex,
+}
+
+impl Condvar {
+    #[inline]
+    pub const fn new() -> Self {
+        Self { futex: Futex::new(0), futex2: Futex::new(0) }
+    }
+
+    // All the memory orderings here are `Relaxed`,
+    // because synchronization is done by unlocking and locking the mutex.
+
+    pub fn notify_one(&self) {
+        self.futex.fetch_add(1, Relaxed);
+        futex_wake(&self.futex);
+    }
+
+    pub fn notify_all(&self) {
+        self.futex.fetch_add(1, Relaxed);
+        futex_requeue(&self.futex, &self.futex2);
+    }
+
+    pub unsafe fn wait(&self, mutex: &Mutex) {
+        self.wait_optional_timeout(mutex, None);
+    }
+
+    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 {
+        // Examine 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();
+
+        // Wake another waiter.
+        if r {
+            self.futex2.fetch_add(1, Relaxed);
+            futex_wake(&self.futex2);
+        }
+
+        r
+    }
+}

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

@@ -1,8 +1,9 @@
 cfg_if::cfg_if! {
-    if #[cfg(any(
+    if #[cfg(any(target_os = "linux", target_os = "android"))] {
+        mod futex_requeue;
+        pub use futex_requeue::Condvar;
+    } else if #[cfg(any(
         all(target_os = "windows", not(target_vendor="win7")),
-        target_os = "linux",
-        target_os = "android",
         target_os = "freebsd",
         target_os = "openbsd",
         target_os = "dragonfly",