implement thread parking on Windows

Author: RalfJung

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

@@ -214,11 +214,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/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
 

tests/pass/threadleak_ignored.rs

@@ -1,4 +1,3 @@
-//@ignore-target-windows: Channels on Windows are not supported yet.
 // FIXME: disallow preemption to work around https://github.com/rust-lang/rust/issues/55005
 //@compile-flags: -Zmiri-ignore-leaks -Zmiri-preemption-rate=0