Implements miri shims for FUTEX_LOCK_PI, FUTEX_UNLOCK_PI

Author: ruihe774

src/tools/miri/src/concurrency/sync.rs

@@ -159,6 +159,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.
@@ -783,6 +785,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,
     ) {
@@ -791,7 +794,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,
@@ -848,4 +851,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
+        }
+    }
 }

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

@@ -1,3 +1,6 @@
+use std::time::Duration;
+
+use crate::shims::unix::env::EvalContextExt;
 use crate::*;
 
 /// Implementation of the SYS_futex syscall.
@@ -15,19 +18,18 @@ pub fn futex<'tcx>(
     // may or may not be left out from the `syscall()` call.
     // Therefore we don't use `check_arg_count` here, but only check for the
     // number of arguments to fall within a range.
-    let [addr, op, val, ..] = args else {
+    let [addr, op, ..] = args else {
         throw_ub_format!(
             "incorrect number of arguments for `futex` syscall: got {}, expected at least 3",
             args.len()
         );
     };
 
-    // The first three arguments (after the syscall number itself) are the same to all futex operations:
-    //     (int *addr, int op, int val).
+    // The first two arguments (after the syscall number itself) are the same to all futex operations:
+    //     (int *addr, int op).
     // We checked above that these definitely exist.
     let addr = this.read_pointer(addr)?;
     let op = this.read_scalar(op)?.to_i32()?;
-    let val = this.read_scalar(val)?.to_i32()?;
 
     // This is a vararg function so we have to bring our own type for this pointer.
     let addr = this.ptr_to_mplace(addr, this.machine.layouts.i32);
@@ -39,6 +41,50 @@ pub fn futex<'tcx>(
     let futex_wake = this.eval_libc_i32("FUTEX_WAKE");
     let futex_wake_bitset = this.eval_libc_i32("FUTEX_WAKE_BITSET");
     let futex_realtime = this.eval_libc_i32("FUTEX_CLOCK_REALTIME");
+    let futex_lock_pi = this.eval_libc_i32("FUTEX_LOCK_PI");
+    let futex_unlock_pi = this.eval_libc_i32("FUTEX_UNLOCK_PI");
+    let futex_waiters = this.eval_libc_u32("FUTEX_WAITERS");
+    let futex_tid_mask = this.eval_libc_u32("FUTEX_TID_MASK");
+
+    // Ok(None) for EINVAL set, Ok(Some(None)) for no timeout (infinity), Ok(Some(Some(...))) for a timeout.
+    // Forgive me, I don't want to create an enum for this return value.
+    fn read_timeout<'tcx>(
+        this: &mut MiriInterpCx<'tcx>,
+        arg3: &OpTy<'tcx>,
+        use_realtime_clock: bool,
+        use_absolute_time: bool,
+        dest: &MPlaceTy<'tcx>,
+    ) -> InterpResult<'tcx, Option<Option<(TimeoutClock, TimeoutAnchor, Duration)>>> {
+        let timeout = this.deref_pointer_as(arg3, this.libc_ty_layout("timespec"))?;
+        interp_ok(Some(if this.ptr_is_null(timeout.ptr())? {
+            None
+        } else {
+            let duration = match this.read_timespec(&timeout)? {
+                Some(duration) => duration,
+                None => {
+                    this.set_last_error(LibcError("EINVAL"))?;
+                    this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                    return interp_ok(None);
+                }
+            };
+            let timeout_clock = if use_realtime_clock {
+                this.check_no_isolation(
+                    "`futex` syscall with `op=FUTEX_WAIT` and non-null timeout with `FUTEX_CLOCK_REALTIME`",
+                )?;
+                TimeoutClock::RealTime
+            } else {
+                TimeoutClock::Monotonic
+            };
+            let timeout_anchor = if use_absolute_time {
+                // FUTEX_WAIT_BITSET uses an absolute timestamp.
+                TimeoutAnchor::Absolute
+            } else {
+                // FUTEX_WAIT uses a relative timestamp.
+                TimeoutAnchor::Relative
+            };
+            Some((timeout_clock, timeout_anchor, duration))
+        }))
+    }
 
     // FUTEX_PRIVATE enables an optimization that stops it from working across processes.
     // Miri doesn't support that anyway, so we ignore that flag.
@@ -74,41 +120,25 @@ pub fn futex<'tcx>(
                 u32::MAX
             };
 
+            // We ensured at least 4 arguments above so these work.
+            let val = this.read_scalar(&args[2])?.to_i32()?;
+            let Some(timeout) = read_timeout(
+                this,
+                &args[3],
+                op & futex_realtime == futex_realtime,
+                wait_bitset,
+                dest,
+            )?
+            else {
+                return interp_ok(());
+            };
+
             if bitset == 0 {
                 this.set_last_error(LibcError("EINVAL"))?;
                 this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
                 return interp_ok(());
             }
 
-            let timeout = this.deref_pointer_as(&args[3], this.libc_ty_layout("timespec"))?;
-            let timeout = if this.ptr_is_null(timeout.ptr())? {
-                None
-            } else {
-                let duration = match this.read_timespec(&timeout)? {
-                    Some(duration) => duration,
-                    None => {
-                        this.set_last_error(LibcError("EINVAL"))?;
-                        this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
-                        return interp_ok(());
-                    }
-                };
-                let timeout_clock = if op & futex_realtime == futex_realtime {
-                    this.check_no_isolation(
-                        "`futex` syscall with `op=FUTEX_WAIT` and non-null timeout with `FUTEX_CLOCK_REALTIME`",
-                    )?;
-                    TimeoutClock::RealTime
-                } else {
-                    TimeoutClock::Monotonic
-                };
-                let timeout_anchor = if wait_bitset {
-                    // FUTEX_WAIT_BITSET uses an absolute timestamp.
-                    TimeoutAnchor::Absolute
-                } else {
-                    // FUTEX_WAIT uses a relative timestamp.
-                    TimeoutAnchor::Relative
-                };
-                Some((timeout_clock, timeout_anchor, duration))
-            };
             // There may be a concurrent thread changing the value of addr
             // and then invoking the FUTEX_WAKE syscall. It is critical that the
             // effects of this and the other thread are correctly observed,
@@ -164,6 +194,7 @@ pub fn futex<'tcx>(
                     timeout,
                     Scalar::from_target_isize(0, this), // retval_succ
                     Scalar::from_target_isize(-1, this), // retval_timeout
+                    None,
                     dest.clone(),
                     this.eval_libc("ETIMEDOUT"),
                 );
@@ -182,6 +213,15 @@ pub fn futex<'tcx>(
         // FUTEX_WAKE_BITSET: (int *addr, int op = FUTEX_WAKE, int val, const timespect *_unused, int *_unused, unsigned int bitset)
         // Same as FUTEX_WAKE, but allows you to specify a bitset to select which threads to wake up.
         op if op == futex_wake || op == futex_wake_bitset => {
+            if args.len() < 3 {
+                throw_ub_format!(
+                    "incorrect number of arguments for `futex` syscall with `op=FUTEX_WAKE`: got {}, expected at least 3",
+                    args.len()
+                );
+            }
+
+            let val = this.read_scalar(&args[2])?.to_i32()?;
+
             let bitset = if op == futex_wake_bitset {
                 let [_, _, _, timeout, uaddr2, bitset, ..] = args else {
                     throw_ub_format!(
@@ -215,6 +255,114 @@ pub fn futex<'tcx>(
             }
             this.write_scalar(Scalar::from_target_isize(n, this), dest)?;
         }
+        op if op == futex_lock_pi => {
+            if args.len() < 4 {
+                throw_ub_format!(
+                    "incorrect number of arguments for `futex` syscall with `op=FUTEX_LOCK_PI`: got {}, expected at least 4",
+                    args.len()
+                );
+            }
+
+            // FUTEX_LOCK_PI uses absolute CLOCK_REALTIME timestamp.
+            let Some(timeout) = read_timeout(this, &args[3], true, true, dest)? else {
+                return interp_ok(());
+            };
+
+            // The tid of the owner is store to *addr.
+            // N.B. it is not the same as posix thread id.
+            let tid = this.linux_gettid()?.to_u32()?;
+
+            // The same as above. This makes modifications visible to us.
+            this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+            // For bitand working properly, we read it as a u32.
+            let futex_val = this.read_scalar_atomic(&addr, AtomicReadOrd::Relaxed)?.to_u32()?;
+
+            if futex_val == 0 {
+                // 0 means unlocked - then lock it.
+                this.write_scalar_atomic(Scalar::from_u32(tid), &addr, AtomicWriteOrd::Relaxed)?;
+
+                // This ensures all loads afterwards get updated value of *addr.
+                this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+                // FUTEX_LOCK_PI returns 0 on success.
+                this.write_scalar(Scalar::from_target_isize(0, this), dest)?;
+            } else if (futex_val & futex_tid_mask) == tid {
+                // Locked by self.
+                this.set_last_error(LibcError("EDEADLK"))?;
+                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)?;
+
+                // Put ourselves into the wait queue.
+                this.futex_wait(
+                    addr_usize,
+                    u32::MAX,
+                    timeout,
+                    Scalar::from_target_isize(0, this), // retval_succ
+                    Scalar::from_target_isize(-1, this), // retval_timeout
+                    Some(tid),
+                    dest.clone(),
+                    this.eval_libc("ETIMEDOUT"),
+                );
+            }
+        }
+        op if op == futex_unlock_pi => {
+            let tid = this.linux_gettid()?.to_u32()?;
+
+            // This ensures all modifications happen before.
+            this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+            // Check we owns lock.
+            let futex_val = this.read_scalar_atomic(&addr, AtomicReadOrd::Relaxed)?.to_u32()?;
+            if (futex_val & futex_tid_mask) != tid {
+                this.set_last_error(LibcError("EINVAL"))?;
+                this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                return interp_ok(());
+            }
+
+            let waiter_count = this.futex_waiter_count(addr_usize);
+            let new_futex_val = if waiter_count == 0 {
+                0
+            } else {
+                let Some(new_tid) = this.futex_top_waiter_extra(addr_usize) else {
+                    // The waiter is not waiting using FUTEX_LOCK_PI.
+                    this.set_last_error(LibcError("EINVAL"))?;
+                    this.write_scalar(Scalar::from_target_isize(-1, this), dest)?;
+                    return interp_ok(());
+                };
+
+                if waiter_count > 1 {
+                    // Still contended after we unlocks.
+                    new_tid | futex_waiters
+                } else {
+                    new_tid
+                }
+            };
+
+            // Update locked state.
+            this.write_scalar_atomic(Scalar::from_u32(new_futex_val), &addr, AtomicWriteOrd::Relaxed)?;
+
+            // This ensures all loads afterwards get updated value of *addr.
+            // There are no preemptions so no one can wake after we set the futex to unlocked
+            // and before we use futex_wake to wake one waiter.
+            this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+            // Unlocking wakes zero or one waiters.
+            let _ = this.futex_wake(addr_usize, u32::MAX)?;
+
+            // FUTEX_UNLOCK_PI returns 0 on success.
+            this.write_scalar(Scalar::from_target_isize(0, this), dest)?;
+        }
         op => throw_unsup_format!("Miri does not support `futex` syscall with op={}", op),
     }
 

src/tools/miri/src/shims/windows/sync.rs

@@ -181,6 +181,7 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
                 timeout,
                 Scalar::from_i32(1), // retval_succ
                 Scalar::from_i32(0), // retval_timeout
+                None,
                 dest.clone(),
                 this.eval_windows("c", "ERROR_TIMEOUT"),
             );