Implement futex_wait and futex_wake.

src/shims/posix/linux/sync.rs

@@ -1,5 +1,7 @@
+use crate::thread::Time;
 use crate::*;
 use rustc_target::abi::{Align, Size};
+use std::time::{Instant, SystemTime};
 
 /// Implementation of the SYS_futex syscall.
 pub fn futex<'tcx>(
@@ -38,23 +40,37 @@ pub fn futex<'tcx>(
     let futex_private = this.eval_libc_i32("FUTEX_PRIVATE_FLAG")?;
     let futex_wait = this.eval_libc_i32("FUTEX_WAIT")?;
     let futex_wake = this.eval_libc_i32("FUTEX_WAKE")?;
+    let futex_realtime = this.eval_libc_i32("FUTEX_CLOCK_REALTIME")?;
 
     // FUTEX_PRIVATE enables an optimization that stops it from working across processes.
     // Miri doesn't support that anyway, so we ignore that flag.
     match op & !futex_private {
         // FUTEX_WAIT: (int *addr, int op = FUTEX_WAIT, int val, const timespec *timeout)
         // Blocks the thread if *addr still equals val. Wakes up when FUTEX_WAKE is called on the same address,
         // or *timeout expires. `timeout == null` for an infinite timeout.
-        op if op == futex_wait => {
+        op if op & !futex_realtime == futex_wait => {
             if args.len() < 5 {
                 throw_ub_format!("incorrect number of arguments for FUTEX_WAIT syscall: got {}, expected at least 5", args.len());
             }
-            let timeout = this.read_scalar(args[4])?.check_init()?;
-            if !this.is_null(timeout)? {
-                // FIXME: Implement timeouts. The condvar waiting code is probably a good example to start with.
-                // Note that a triggered timeout should have this syscall return with -1 and errno set to ETIMEOUT.
-                throw_ub_format!("miri does not support timeouts for futex operations");
-            }
+            let timeout = args[4];
+            let timeout_time = if this.is_null(this.read_scalar(timeout)?.check_init()?)? {
+                None
+            } else {
+                let duration = match this.read_timespec(timeout)? {
+                    Some(duration) => duration,
+                    None => {
+                        let einval = this.eval_libc("EINVAL")?;
+                        this.set_last_error(einval)?;
+                        this.write_scalar(Scalar::from_i32(-1), dest)?;
+                        return Ok(());
+                    }
+                };
+                Some(if op & futex_realtime != 0 {
+                    Time::RealTime(SystemTime::now().checked_add(duration).unwrap())
+                } else {
+                    Time::Monotonic(Instant::now().checked_add(duration).unwrap())
+                })
+            };
             // Check the pointer for alignment and validity.
             // Atomic operations are only available for fully aligned values.
             this.memory.check_ptr_access(addr.into(), Size::from_bytes(4), Align::from_bytes(4).unwrap())?;
@@ -66,6 +82,22 @@ pub fn futex<'tcx>(
                 this.futex_wait(addr, thread);
                 // Succesfully waking up from FUTEX_WAIT always returns zero.
                 this.write_scalar(Scalar::from_i32(0), dest)?;
+                // Register a timeout callback if a timeout was specified.
+                // This callback will override the return value when the timeout triggers.
+                if let Some(timeout_time) = timeout_time {
+                    this.register_timeout_callback(
+                        thread,
+                        timeout_time,
+                        Box::new(move |this| {
+                            this.unblock_thread(thread);
+                            this.futex_remove_waiter(addr, thread);
+                            let etimedout = this.eval_libc("ETIMEDOUT")?;
+                            this.set_last_error(etimedout)?;
+                            this.write_scalar(Scalar::from_i32(-1), dest)?;
+                            Ok(())
+                        }),
+                    );
+                }
             } else {
                 // The futex value doesn't match the expected value, so we return failure
                 // right away without sleeping: -1 and errno set to EAGAIN.
@@ -83,6 +115,7 @@ pub fn futex<'tcx>(
             for _ in 0..val {
                 if let Some(thread) = this.futex_wake(addr) {
                     this.unblock_thread(thread);
+                    this.unregister_timeout_callback_if_exists(thread);
                     n += 1;
                 } else {
                     break;

src/sync.rs

@@ -430,4 +430,11 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx
         let waiters = &mut this.machine.threads.sync.futexes.get_mut(&addr.erase_tag())?.waiters;
         waiters.pop_front().map(|waiter| waiter.thread)
     }
+
+    fn futex_remove_waiter(&mut self, addr: Pointer<stacked_borrows::Tag>, thread: ThreadId) {
+        let this = self.eval_context_mut();
+        if let Some(futex) = this.machine.threads.sync.futexes.get_mut(&addr.erase_tag()) {
+            futex.waiters.retain(|waiter| waiter.thread != thread);
+        }
+    }
 }

tests/run-pass/concurrency/parking.rs

@@ -0,0 +1,37 @@
+// ignore-windows: Concurrency on Windows is not supported yet.
+// compile-flags: -Zmiri-disable-isolation
+
+use std::thread;
+use std::time::{Duration, Instant};
+
+// Normally, waiting in park/park_timeout may spuriously wake up early, but we
+// know Miri's timed synchronization primitives do not do that.
+
+fn park_timeout() {
+    let start = Instant::now();
+
+    thread::park_timeout(Duration::from_millis(200));
+
+    assert!((200..500).contains(&start.elapsed().as_millis()));
+}
+
+fn park_unpark() {
+    let t1 = thread::current();
+    let t2 = thread::spawn(move || {
+        thread::park();
+        thread::sleep(Duration::from_millis(200));
+        t1.unpark();
+    });
+
+    let start = Instant::now();
+
+    t2.thread().unpark();
+    thread::park();
+
+    assert!((200..500).contains(&start.elapsed().as_millis()));
+}
+
+fn main() {
+    park_timeout();
+    park_unpark();
+}

tests/run-pass/concurrency/parking.stderr

@@ -0,0 +1,2 @@
+warning: thread support is experimental. For example, Miri does not detect data races yet.
+