Adds detection of mixed usage of FUTEX_WAIT and FUTEX_WAKE

Author: ruihe774

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

@@ -26,6 +26,7 @@ pub fn futex_wait(futex: &AtomicU32, expected: u32, timeout: Option<Duration>) -
     use super::time::Timespec;
     use crate::ptr::null;
     use crate::sync::atomic::Ordering::Relaxed;
+    use crate::sys::cvt;
 
     // Calculate the timeout as an absolute timespec.
     //
@@ -40,7 +41,7 @@ pub fn futex_wait(futex: &AtomicU32, expected: u32, timeout: Option<Duration>) -
             return true;
         }
 
-        let r = unsafe {
+        let r = cvt(unsafe {
             cfg_if::cfg_if! {
                 if #[cfg(target_os = "freebsd")] {
                     // FreeBSD doesn't have futex(), but it has
@@ -77,12 +78,16 @@ pub fn futex_wait(futex: &AtomicU32, expected: u32, timeout: Option<Duration>) -
                     compile_error!("unknown target_os");
                 }
             }
-        };
-
-        match (r < 0).then(super::os::errno) {
-            Some(libc::ETIMEDOUT) => return false,
-            Some(libc::EINTR) => continue,
-            _ => return true,
+        });
+
+        match r {
+            Ok(_) => return true,
+            Err(e) => match e.raw_os_error() {
+                Some(libc::ETIMEDOUT) => return false,
+                Some(libc::EINTR) => continue,
+                Some(libc::EAGAIN) => return true,
+                _ => panic!("failed to wait on futex: {e:?}"),
+            },
         }
     }
 }
@@ -95,19 +100,22 @@ pub fn futex_wait(futex: &AtomicU32, expected: u32, timeout: Option<Duration>) -
 /// On some platforms, this always returns false.
 #[cfg(any(target_os = "linux", target_os = "android"))]
 pub fn futex_wake(futex: &AtomicU32) -> bool {
+    use crate::sys::cvt;
     let ptr = futex as *const AtomicU32;
     let op = libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG;
-    unsafe { libc::syscall(libc::SYS_futex, ptr, op, 1) > 0 }
+    cvt(unsafe { libc::syscall(libc::SYS_futex, ptr, op, 1) })
+        .expect("failed to wake futex waiters")
+        > 0
 }
 
 /// Wakes up all threads that are waiting on `futex_wait` on this futex.
 #[cfg(any(target_os = "linux", target_os = "android"))]
 pub fn futex_wake_all(futex: &AtomicU32) {
+    use crate::sys::cvt;
     let ptr = futex as *const AtomicU32;
     let op = libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG;
-    unsafe {
-        libc::syscall(libc::SYS_futex, ptr, op, i32::MAX);
-    }
+    cvt(unsafe { libc::syscall(libc::SYS_futex, ptr, op, i32::MAX) })
+        .expect("failed to wake futex waiters");
 }
 
 // FreeBSD doesn't tell us how many threads are woken up, so this always returns false.

src/tools/miri/src/shims/unix/linux/sync.rs

@@ -187,6 +187,13 @@ pub fn futex<'tcx>(
             // It's not uncommon for `addr` to be passed as another type than `*mut i32`, such as `*const AtomicI32`.
             let futex_val = this.read_scalar_atomic(&addr, AtomicReadOrd::Relaxed)?.to_i32()?;
             if val == futex_val {
+                // Check that the top waiter (if exists) is waiting using FUTEX_WAIT_*.
+                if this.futex_waiter_count(addr_usize) != 0 && this.futex_top_waiter_extra(addr_usize).is_some() {
+                    this.set_last_error(LibcError("EINVAL"))?;
+                    this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                    return interp_ok(());
+                }
+
                 // The value still matches, so we block the thread and make it wait for FUTEX_WAKE.
                 this.futex_wait(
                     addr_usize,
@@ -245,6 +252,16 @@ pub fn futex<'tcx>(
             // before doing the syscall.
             this.atomic_fence(AtomicFenceOrd::SeqCst)?;
             let mut n = 0;
+
+            // Check that the waiter is waiting using FUTEX_WAIT_*.
+            if this.futex_waiter_count(addr_usize) != 0 {
+                if this.futex_top_waiter_extra(addr_usize).is_some() {
+                    this.set_last_error(LibcError("EINVAL"))?;
+                    this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                    return interp_ok(());
+                }
+            }
+
             #[allow(clippy::arithmetic_side_effects)]
             for _ in 0..val {
                 if this.futex_wake(addr_usize, bitset)? {
@@ -280,6 +297,15 @@ pub fn futex<'tcx>(
 
             if futex_val == 0 {
                 // 0 means unlocked - then lock it.
+
+                // Check that there is no waiters.
+                if this.futex_waiter_count(addr_usize) != 0 {
+                    this.set_last_error(LibcError("EINVAL"))?;
+                    this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                    return interp_ok(());
+                }
+
+                // Declare ownership.
                 this.write_scalar_atomic(Scalar::from_u32(tid), &addr, AtomicWriteOrd::Relaxed)?;
 
                 // This ensures all loads afterwards get updated value of *addr.
@@ -293,15 +319,26 @@ pub fn futex<'tcx>(
                 this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
             } else {
                 // Other values mean locked.
-                // Mark the futex as contended.
-                this.write_scalar_atomic(
-                    Scalar::from_u32(futex_val | futex_waiters),
-                    &addr,
-                    AtomicWriteOrd::Relaxed,
-                )?;
 
-                // This ensures all loads afterwards get updated value of *addr.
-                this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+                if this.futex_waiter_count(addr_usize) == 0 {
+                    // Mark the futex as contended.
+                    this.write_scalar_atomic(
+                        Scalar::from_u32(futex_val | futex_waiters),
+                        &addr,
+                        AtomicWriteOrd::Relaxed,
+                    )?;
+
+                    // This ensures all loads afterwards get updated value of *addr.
+                    this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+                } else {
+                    // Check that the futex has been marked as contended.
+                    // Check that the top waiter is waiting using FUTEX_LOCK_PI.
+                    if (futex_val & futex_waiters) == 0 || this.futex_top_waiter_extra(addr_usize).is_none() {
+                        this.set_last_error(LibcError("EINVAL"))?;
+                        this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                        return interp_ok(());
+                    }
+                }
 
                 // Put ourselves into the wait queue.
                 this.futex_wait(