Auto merge of #2630 - RalfJung:windows-parking, r=RalfJung

Implement thread parking for Windows

Cc #2628

Based on code by `@DrMeepster.` However I adjusted `WakeByAddressSingle`: I don't think the futex value is compared *again* after the thread is woken up. I see nothing in the Windows docs indicating such a comparison, and the Linux futex does not behave like that either. So we only check the value before sleeping, same as on Linux.

src/lib.rs

@@ -98,8 +98,8 @@ pub use crate::eval::{
 pub use crate::helpers::{CurrentSpan, EvalContextExt as _};
 pub use crate::intptrcast::ProvenanceMode;
 pub use crate::machine::{
-    AllocExtra, FrameData, MiriInterpCx, MiriInterpCxExt, MiriMachine, MiriMemoryKind, Provenance,
-    ProvenanceExtra, PAGE_SIZE, STACK_ADDR, STACK_SIZE,
+    AllocExtra, FrameData, MiriInterpCx, MiriInterpCxExt, MiriMachine, MiriMemoryKind,
+    PrimitiveLayouts, Provenance, ProvenanceExtra, PAGE_SIZE, STACK_ADDR, STACK_SIZE,
 };
 pub use crate::mono_hash_map::MonoHashMap;
 pub use crate::operator::EvalContextExt as _;

src/machine.rs

@@ -276,10 +276,14 @@ pub struct PrimitiveLayouts<'tcx> {
     pub i8: TyAndLayout<'tcx>,
     pub i16: TyAndLayout<'tcx>,
     pub i32: TyAndLayout<'tcx>,
+    pub i64: TyAndLayout<'tcx>,
+    pub i128: TyAndLayout<'tcx>,
     pub isize: TyAndLayout<'tcx>,
     pub u8: TyAndLayout<'tcx>,
     pub u16: TyAndLayout<'tcx>,
     pub u32: TyAndLayout<'tcx>,
+    pub u64: TyAndLayout<'tcx>,
+    pub u128: TyAndLayout<'tcx>,
     pub usize: TyAndLayout<'tcx>,
     pub bool: TyAndLayout<'tcx>,
     pub mut_raw_ptr: TyAndLayout<'tcx>,   // *mut ()
@@ -296,16 +300,42 @@ impl<'mir, 'tcx: 'mir> PrimitiveLayouts<'tcx> {
             i8: layout_cx.layout_of(tcx.types.i8)?,
             i16: layout_cx.layout_of(tcx.types.i16)?,
             i32: layout_cx.layout_of(tcx.types.i32)?,
+            i64: layout_cx.layout_of(tcx.types.i64)?,
+            i128: layout_cx.layout_of(tcx.types.i128)?,
             isize: layout_cx.layout_of(tcx.types.isize)?,
             u8: layout_cx.layout_of(tcx.types.u8)?,
             u16: layout_cx.layout_of(tcx.types.u16)?,
             u32: layout_cx.layout_of(tcx.types.u32)?,
+            u64: layout_cx.layout_of(tcx.types.u64)?,
+            u128: layout_cx.layout_of(tcx.types.u128)?,
             usize: layout_cx.layout_of(tcx.types.usize)?,
             bool: layout_cx.layout_of(tcx.types.bool)?,
             mut_raw_ptr: layout_cx.layout_of(mut_raw_ptr)?,
             const_raw_ptr: layout_cx.layout_of(const_raw_ptr)?,
         })
     }
+
+    pub fn uint(&self, size: Size) -> Option<TyAndLayout<'tcx>> {
+        match size.bits() {
+            8 => Some(self.u8),
+            16 => Some(self.u16),
+            32 => Some(self.u32),
+            64 => Some(self.u64),
+            128 => Some(self.u128),
+            _ => None,
+        }
+    }
+
+    pub fn int(&self, size: Size) -> Option<TyAndLayout<'tcx>> {
+        match size.bits() {
+            8 => Some(self.i8),
+            16 => Some(self.i16),
+            32 => Some(self.i32),
+            64 => Some(self.i64),
+            128 => Some(self.i128),
+            _ => None,
+        }
+    }
 }
 
 /// The machine itself.

src/shims/unix/linux/sync.rs

@@ -1,7 +1,7 @@
+use std::time::SystemTime;
+
 use crate::concurrency::thread::{MachineCallback, Time};
 use crate::*;
-use rustc_target::abi::{Align, Size};
-use std::time::SystemTime;
 
 /// Implementation of the SYS_futex syscall.
 /// `args` is the arguments *after* the syscall number.
@@ -28,13 +28,14 @@ pub fn futex<'tcx>(
     // The first three arguments (after the syscall number itself) are the same to all futex operations:
     //     (int *addr, int op, int val).
     // We checked above that these definitely exist.
-    let addr = this.read_immediate(&args[0])?;
+    let addr = this.read_pointer(&args[0])?;
     let op = this.read_scalar(&args[1])?.to_i32()?;
     let val = this.read_scalar(&args[2])?.to_i32()?;
 
     let thread = this.get_active_thread();
-    let addr_scalar = addr.to_scalar();
-    let addr_usize = addr_scalar.to_machine_usize(this)?;
+    // This is a vararg function so we have to bring our own type for this pointer.
+    let addr = MPlaceTy::from_aligned_ptr(addr, this.machine.layouts.i32);
+    let addr_usize = addr.ptr.addr().bytes();
 
     let futex_private = this.eval_libc_i32("FUTEX_PRIVATE_FLAG")?;
     let futex_wait = this.eval_libc_i32("FUTEX_WAIT")?;
@@ -117,15 +118,6 @@ pub fn futex<'tcx>(
                     }
                 })
             };
-            // Check the pointer for alignment and validity.
-            // The API requires `addr` to be a 4-byte aligned pointer, and will
-            // use the 4 bytes at the given address as an (atomic) i32.
-            this.check_ptr_access_align(
-                addr_scalar.to_pointer(this)?,
-                Size::from_bytes(4),
-                Align::from_bytes(4).unwrap(),
-                CheckInAllocMsg::MemoryAccessTest,
-            )?;
             // 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,
@@ -172,14 +164,7 @@ pub fn futex<'tcx>(
             this.atomic_fence(AtomicFenceOrd::SeqCst)?;
             // Read an `i32` through the pointer, regardless of any wrapper types.
             // 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_at_offset_atomic(
-                    &addr.into(),
-                    0,
-                    this.machine.layouts.i32,
-                    AtomicReadOrd::Relaxed,
-                )?
-                .to_i32()?;
+            let futex_val = this.read_scalar_atomic(&addr, AtomicReadOrd::Relaxed)?.to_i32()?;
             if val == futex_val {
                 // The value still matches, so we block the thread make it wait for FUTEX_WAKE.
                 this.block_thread(thread);
@@ -214,11 +199,10 @@ pub fn futex<'tcx>(
                         }
                     }
 
-                    let dest = dest.clone();
                     this.register_timeout_callback(
                         thread,
                         timeout_time,
-                        Box::new(Callback { thread, addr_usize, dest }),
+                        Box::new(Callback { thread, addr_usize, dest: dest.clone() }),
                     );
                 }
             } else {

src/shims/windows/dlsym.rs

@@ -6,12 +6,15 @@ use log::trace;
 
 use crate::helpers::check_arg_count;
 use crate::shims::windows::handle::{EvalContextExt as _, Handle, PseudoHandle};
+use crate::shims::windows::sync::EvalContextExt as _;
 use crate::*;
 
 #[derive(Debug, Copy, Clone)]
 pub enum Dlsym {
     NtWriteFile,
     SetThreadDescription,
+    WaitOnAddress,
+    WakeByAddressSingle,
 }
 
 impl Dlsym {
@@ -22,6 +25,8 @@ impl Dlsym {
             "GetSystemTimePreciseAsFileTime" => None,
             "NtWriteFile" => Some(Dlsym::NtWriteFile),
             "SetThreadDescription" => Some(Dlsym::SetThreadDescription),
+            "WaitOnAddress" => Some(Dlsym::WaitOnAddress),
+            "WakeByAddressSingle" => Some(Dlsym::WakeByAddressSingle),
             _ => throw_unsup_format!("unsupported Windows dlsym: {}", name),
         })
     }
@@ -127,6 +132,16 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
 
                 this.write_null(dest)?;
             }
+            Dlsym::WaitOnAddress => {
+                let [ptr_op, compare_op, size_op, timeout_op] = check_arg_count(args)?;
+
+                this.WaitOnAddress(ptr_op, compare_op, size_op, timeout_op, dest)?;
+            }
+            Dlsym::WakeByAddressSingle => {
+                let [ptr_op] = check_arg_count(args)?;
+
+                this.WakeByAddressSingle(ptr_op)?;
+            }
         }
 
         trace!("{:?}", this.dump_place(**dest));

src/shims/windows/sync.rs

@@ -1,3 +1,7 @@
+use std::time::Duration;
+
+use rustc_target::abi::Size;
+
 use crate::concurrency::init_once::InitOnceStatus;
 use crate::concurrency::thread::MachineCallback;
 use crate::*;
@@ -6,7 +10,6 @@ const SRWLOCK_ID_OFFSET: u64 = 0;
 const INIT_ONCE_ID_OFFSET: u64 = 0;
 
 impl<'mir, 'tcx> EvalContextExt<'mir, 'tcx> for crate::MiriInterpCx<'mir, 'tcx> {}
-
 #[allow(non_snake_case)]
 pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
     fn AcquireSRWLockExclusive(&mut self, lock_op: &OpTy<'tcx, Provenance>) -> InterpResult<'tcx> {
@@ -221,4 +224,107 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
 
         this.eval_windows("c", "TRUE")
     }
+
+    fn WaitOnAddress(
+        &mut self,
+        ptr_op: &OpTy<'tcx, Provenance>,
+        compare_op: &OpTy<'tcx, Provenance>,
+        size_op: &OpTy<'tcx, Provenance>,
+        timeout_op: &OpTy<'tcx, Provenance>,
+        dest: &PlaceTy<'tcx, Provenance>,
+    ) -> InterpResult<'tcx> {
+        let this = self.eval_context_mut();
+
+        let ptr = this.read_pointer(ptr_op)?;
+        let compare = this.read_pointer(compare_op)?;
+        let size = this.read_scalar(size_op)?.to_machine_usize(this)?;
+        let timeout_ms = this.read_scalar(timeout_op)?.to_u32()?;
+
+        let thread = this.get_active_thread();
+        let addr = ptr.addr().bytes();
+
+        if size > 8 || !size.is_power_of_two() {
+            let invalid_param = this.eval_windows("c", "ERROR_INVALID_PARAMETER")?;
+            this.set_last_error(invalid_param)?;
+            this.write_scalar(Scalar::from_i32(0), dest)?;
+            return Ok(());
+        };
+        let size = Size::from_bytes(size);
+
+        let timeout_time = if timeout_ms == this.eval_windows("c", "INFINITE")?.to_u32()? {
+            None
+        } else {
+            this.check_no_isolation("`WaitOnAddress` with non-infinite timeout")?;
+
+            let duration = Duration::from_millis(timeout_ms.into());
+            Some(Time::Monotonic(this.machine.clock.now().checked_add(duration).unwrap()))
+        };
+
+        // See the Linux futex implementation for why this fence exists.
+        this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+        let layout = this.machine.layouts.uint(size).unwrap();
+        let futex_val = this
+            .read_scalar_atomic(&MPlaceTy::from_aligned_ptr(ptr, layout), AtomicReadOrd::Relaxed)?;
+        let compare_val = this.read_scalar(&MPlaceTy::from_aligned_ptr(compare, layout).into())?;
+
+        if futex_val == compare_val {
+            // If the values are the same, we have to block.
+            this.block_thread(thread);
+            this.futex_wait(addr, thread, u32::MAX);
+
+            if let Some(timeout_time) = timeout_time {
+                struct Callback<'tcx> {
+                    thread: ThreadId,
+                    addr: u64,
+                    dest: PlaceTy<'tcx, Provenance>,
+                }
+
+                impl<'tcx> VisitTags for Callback<'tcx> {
+                    fn visit_tags(&self, visit: &mut dyn FnMut(SbTag)) {
+                        let Callback { thread: _, addr: _, dest } = self;
+                        dest.visit_tags(visit);
+                    }
+                }
+
+                impl<'mir, 'tcx: 'mir> MachineCallback<'mir, 'tcx> for Callback<'tcx> {
+                    fn call(&self, this: &mut MiriInterpCx<'mir, 'tcx>) -> InterpResult<'tcx> {
+                        this.unblock_thread(self.thread);
+                        this.futex_remove_waiter(self.addr, self.thread);
+                        let error_timeout = this.eval_windows("c", "ERROR_TIMEOUT")?;
+                        this.set_last_error(error_timeout)?;
+                        this.write_scalar(Scalar::from_i32(0), &self.dest)?;
+
+                        Ok(())
+                    }
+                }
+
+                this.register_timeout_callback(
+                    thread,
+                    timeout_time,
+                    Box::new(Callback { thread, addr, dest: dest.clone() }),
+                );
+            }
+        }
+
+        this.write_scalar(Scalar::from_i32(1), dest)?;
+
+        Ok(())
+    }
+
+    fn WakeByAddressSingle(&mut self, ptr_op: &OpTy<'tcx, Provenance>) -> InterpResult<'tcx> {
+        let this = self.eval_context_mut();
+
+        let ptr = this.read_pointer(ptr_op)?;
+
+        // See the Linux futex implementation for why this fence exists.
+        this.atomic_fence(AtomicFenceOrd::SeqCst)?;
+
+        if let Some(thread) = this.futex_wake(ptr.addr().bytes(), u32::MAX) {
+            this.unblock_thread(thread);
+            this.unregister_timeout_callback_if_exists(thread);
+        }
+
+        Ok(())
+    }
 }

tests/pass-dep/shims/libc-misc.rs

@@ -87,7 +87,7 @@ fn test_posix_realpath_errors() {
     assert_eq!(e.kind(), ErrorKind::NotFound);
 }
 
-#[cfg(any(target_os = "linux"))]
+#[cfg(target_os = "linux")]
 fn test_posix_fadvise() {
     use std::convert::TryInto;
     use std::io::Write;
@@ -115,7 +115,7 @@ fn test_posix_fadvise() {
     assert_eq!(result, 0);
 }
 
-#[cfg(any(target_os = "linux"))]
+#[cfg(target_os = "linux")]
 fn test_sync_file_range() {
     use std::io::Write;
 
@@ -181,7 +181,7 @@ fn test_thread_local_errno() {
 }
 
 /// Tests whether clock support exists at all
-#[cfg(any(target_os = "linux"))]
+#[cfg(target_os = "linux")]
 fn test_clocks() {
     let mut tp = std::mem::MaybeUninit::<libc::timespec>::uninit();
     let is_error = unsafe { libc::clock_gettime(libc::CLOCK_REALTIME, tp.as_mut_ptr()) };
@@ -283,23 +283,21 @@ fn test_posix_mkstemp() {
 }
 
 fn main() {
-    #[cfg(any(target_os = "linux"))]
-    test_posix_fadvise();
-
     test_posix_gettimeofday();
     test_posix_mkstemp();
 
     test_posix_realpath_alloc();
     test_posix_realpath_noalloc();
     test_posix_realpath_errors();
 
-    #[cfg(any(target_os = "linux"))]
-    test_sync_file_range();
-
     test_thread_local_errno();
 
-    #[cfg(any(target_os = "linux"))]
-    test_clocks();
-
     test_isatty();
+
+    #[cfg(target_os = "linux")]
+    {
+        test_posix_fadvise();
+        test_sync_file_range();
+        test_clocks();
+    }
 }

tests/pass-dep/shims/pthreads.rs

@@ -10,7 +10,7 @@ fn main() {
     test_rwlock_libc_static_initializer();
     test_named_thread_truncation();
 
-    #[cfg(any(target_os = "linux"))]
+    #[cfg(target_os = "linux")]
     test_mutex_libc_static_initializer_recursive();
 }
 

tests/pass/concurrency/channels.rs

@@ -1,4 +1,3 @@
-//@ignore-target-windows: Channels on Windows are not supported yet.
 //@compile-flags: -Zmiri-strict-provenance
 
 use std::sync::mpsc::{channel, sync_channel};

tests/pass/concurrency/spin_loops_nopreempt.rs

@@ -1,4 +1,3 @@
-//@ignore-target-windows: Channels on Windows are not supported yet.
 // This specifically tests behavior *without* preemption.
 //@compile-flags: -Zmiri-preemption-rate=0