Implements miri shims for FUTEX_LOCK_PI, FUTEX_UNLOCK_PI

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/concurrency/sync.rs

@@ -128,6 +128,8 @@ struct FutexWaiter {
     thread: ThreadId,
     /// The bitset used by FUTEX_*_BITSET, or u32::MAX for other operations.
     bitset: u32,
+    /// Extra info stored for this waiter.
+    extra: Option<u32>,
 }
 
 /// The state of all synchronization objects.
@@ -695,6 +697,7 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
         timeout: Option<(TimeoutClock, TimeoutAnchor, Duration)>,
         retval_succ: Scalar,
         retval_timeout: Scalar,
+        waiter_extra: Option<u32>,
         dest: MPlaceTy<'tcx>,
         errno_timeout: Scalar,
     ) {
@@ -703,7 +706,7 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
         let futex = &mut this.machine.sync.futexes.entry(addr).or_default();
         let waiters = &mut futex.waiters;
         assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
-        waiters.push_back(FutexWaiter { thread, bitset });
+        waiters.push_back(FutexWaiter { thread, bitset, extra: waiter_extra });
         this.block_thread(
             BlockReason::Futex { addr },
             timeout,
@@ -760,4 +763,22 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
         this.unblock_thread(waiter.thread, BlockReason::Futex { addr })?;
         interp_ok(true)
     }
+
+    /// Returns the extra info of the top waiter.
+    /// The caller must ensure there are waiters.
+    fn futex_top_waiter_extra(&mut self, addr: u64) -> Option<u32> {
+        let this = self.eval_context_mut();
+        let futex = &mut this.machine.sync.futexes.get(&addr)?;
+        futex.waiters.front().unwrap().extra
+    }
+
+    /// Returns the number of waiters
+    fn futex_waiter_count(&mut self, addr: u64) -> usize {
+        let this = self.eval_context_mut();
+        if let Some(futex) = &mut this.machine.sync.futexes.get(&addr) {
+            futex.waiters.len()
+        } else {
+            0
+        }
+    }
 }