Auto merge of #3471 - RalfJung:blocked, r=RalfJung

bors · bors · commit 1129faca2bc7 · 2024-04-16T10:17:07.000Z
threads: keep track of why we are blocked, and sanity-check that when waking up

Also remove support for condvars blocked on rwlocks, as that was no longer used.
diff --git a/src/tools/miri/src/concurrency/init_once.rs b/src/tools/miri/src/concurrency/init_once.rs
@@ -77,7 +77,7 @@ trait EvalContextExtPriv<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
         let this = self.eval_context_mut();
         let current_thread = this.get_active_thread();
 
-        this.unblock_thread(waiter.thread);
+        this.unblock_thread(waiter.thread, BlockReason::InitOnce(id));
 
         // Call callback, with the woken-up thread as `current`.
         this.set_active_thread(waiter.thread);
@@ -142,7 +142,7 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
         let init_once = &mut this.machine.threads.sync.init_onces[id];
         assert_ne!(init_once.status, InitOnceStatus::Complete, "queueing on complete init once");
         init_once.waiters.push_back(InitOnceWaiter { thread, callback });
-        this.block_thread(thread);
+        this.block_thread(thread, BlockReason::InitOnce(id));
     }
 
     /// Begin initializing this InitOnce. Must only be called after checking that it is currently
diff --git a/src/tools/miri/src/concurrency/sync.rs b/src/tools/miri/src/concurrency/sync.rs
@@ -115,25 +115,13 @@ struct RwLock {
 
 declare_id!(CondvarId);
 
-#[derive(Debug, Copy, Clone)]
-pub enum RwLockMode {
-    Read,
-    Write,
-}
-
-#[derive(Debug)]
-pub enum CondvarLock {
-    Mutex(MutexId),
-    RwLock { id: RwLockId, mode: RwLockMode },
-}
-
 /// A thread waiting on a conditional variable.
 #[derive(Debug)]
 struct CondvarWaiter {
     /// The thread that is waiting on this variable.
     thread: ThreadId,
-    /// The mutex or rwlock on which the thread is waiting.
-    lock: CondvarLock,
+    /// The mutex on which the thread is waiting.
+    lock: MutexId,
 }
 
 /// The conditional variable state.
@@ -232,7 +220,7 @@ pub(super) trait EvalContextExtPriv<'mir, 'tcx: 'mir>:
     fn rwlock_dequeue_and_lock_reader(&mut self, id: RwLockId) -> bool {
         let this = self.eval_context_mut();
         if let Some(reader) = this.machine.threads.sync.rwlocks[id].reader_queue.pop_front() {
-            this.unblock_thread(reader);
+            this.unblock_thread(reader, BlockReason::RwLock(id));
             this.rwlock_reader_lock(id, reader);
             true
         } else {
@@ -246,7 +234,7 @@ pub(super) trait EvalContextExtPriv<'mir, 'tcx: 'mir>:
     fn rwlock_dequeue_and_lock_writer(&mut self, id: RwLockId) -> bool {
         let this = self.eval_context_mut();
         if let Some(writer) = this.machine.threads.sync.rwlocks[id].writer_queue.pop_front() {
-            this.unblock_thread(writer);
+            this.unblock_thread(writer, BlockReason::RwLock(id));
             this.rwlock_writer_lock(id, writer);
             true
         } else {
@@ -260,7 +248,7 @@ pub(super) trait EvalContextExtPriv<'mir, 'tcx: 'mir>:
     fn mutex_dequeue_and_lock(&mut self, id: MutexId) -> bool {
         let this = self.eval_context_mut();
         if let Some(thread) = this.machine.threads.sync.mutexes[id].queue.pop_front() {
-            this.unblock_thread(thread);
+            this.unblock_thread(thread, BlockReason::Mutex(id));
             this.mutex_lock(id, thread);
             true
         } else {
@@ -406,7 +394,7 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
         let this = self.eval_context_mut();
         assert!(this.mutex_is_locked(id), "queing on unlocked mutex");
         this.machine.threads.sync.mutexes[id].queue.push_back(thread);
-        this.block_thread(thread);
+        this.block_thread(thread, BlockReason::Mutex(id));
     }
 
     /// Provides the closure with the next RwLockId. Creates that RwLock if the closure returns None,
@@ -511,7 +499,7 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
         let this = self.eval_context_mut();
         assert!(this.rwlock_is_write_locked(id), "read-queueing on not write locked rwlock");
         this.machine.threads.sync.rwlocks[id].reader_queue.push_back(reader);
-        this.block_thread(reader);
+        this.block_thread(reader, BlockReason::RwLock(id));
     }
 
     /// Lock by setting the writer that owns the lock.
@@ -573,7 +561,7 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
         let this = self.eval_context_mut();
         assert!(this.rwlock_is_locked(id), "write-queueing on unlocked rwlock");
         this.machine.threads.sync.rwlocks[id].writer_queue.push_back(writer);
-        this.block_thread(writer);
+        this.block_thread(writer, BlockReason::RwLock(id));
     }
 
     /// Provides the closure with the next CondvarId. Creates that Condvar if the closure returns None,
@@ -605,7 +593,7 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
     }
 
     /// Mark that the thread is waiting on the conditional variable.
-    fn condvar_wait(&mut self, id: CondvarId, thread: ThreadId, lock: CondvarLock) {
+    fn condvar_wait(&mut self, id: CondvarId, thread: ThreadId, lock: MutexId) {
         let this = self.eval_context_mut();
         let waiters = &mut this.machine.threads.sync.condvars[id].waiters;
         assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
@@ -614,7 +602,7 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
 
     /// Wake up some thread (if there is any) sleeping on the conditional
     /// variable.
-    fn condvar_signal(&mut self, id: CondvarId) -> Option<(ThreadId, CondvarLock)> {
+    fn condvar_signal(&mut self, id: CondvarId) -> Option<(ThreadId, MutexId)> {
         let this = self.eval_context_mut();
         let current_thread = this.get_active_thread();
         let current_span = this.machine.current_span();
diff --git a/src/tools/miri/src/concurrency/thread.rs b/src/tools/miri/src/concurrency/thread.rs
@@ -88,18 +88,33 @@ impl From<ThreadId> for u64 {
     }
 }
 
+/// Keeps track of what the thread is blocked on.
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum BlockReason {
+    /// The thread tried to join the specified thread and is blocked until that
+    /// thread terminates.
+    Join(ThreadId),
+    /// Waiting for time to pass.
+    Sleep,
+    /// Blocked on a mutex.
+    Mutex(MutexId),
+    /// Blocked on a condition variable.
+    Condvar(CondvarId),
+    /// Blocked on a reader-writer lock.
+    RwLock(RwLockId),
+    /// Blocled on a Futex variable.
+    Futex { addr: u64 },
+    /// Blocked on an InitOnce.
+    InitOnce(InitOnceId),
+}
+
 /// The state of a thread.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum ThreadState {
     /// The thread is enabled and can be executed.
     Enabled,
-    /// The thread tried to join the specified thread and is blocked until that
-    /// thread terminates.
-    BlockedOnJoin(ThreadId),
-    /// The thread is blocked on some synchronization primitive. It is the
-    /// responsibility of the synchronization primitives to track threads that
-    /// are blocked by them.
-    BlockedOnSync,
+    /// The thread is blocked on something.
+    Blocked(BlockReason),
     /// The thread has terminated its execution. We do not delete terminated
     /// threads (FIXME: why?).
     Terminated,
@@ -296,17 +311,17 @@ impl VisitProvenance for Frame<'_, '_, Provenance, FrameExtra<'_>> {
 
 /// A specific moment in time.
 #[derive(Debug)]
-pub enum Time {
+pub enum CallbackTime {
     Monotonic(Instant),
     RealTime(SystemTime),
 }
 
-impl Time {
+impl CallbackTime {
     /// How long do we have to wait from now until the specified time?
     fn get_wait_time(&self, clock: &Clock) -> Duration {
         match self {
-            Time::Monotonic(instant) => instant.duration_since(clock.now()),
-            Time::RealTime(time) =>
+            CallbackTime::Monotonic(instant) => instant.duration_since(clock.now()),
+            CallbackTime::RealTime(time) =>
                 time.duration_since(SystemTime::now()).unwrap_or(Duration::new(0, 0)),
         }
     }
@@ -318,7 +333,7 @@ impl Time {
 /// conditional variable, the signal handler deletes the callback.
 struct TimeoutCallbackInfo<'mir, 'tcx> {
     /// The callback should be called no earlier than this time.
-    call_time: Time,
+    call_time: CallbackTime,
     /// The called function.
     callback: TimeoutCallback<'mir, 'tcx>,
 }
@@ -539,7 +554,8 @@ impl<'mir, 'tcx: 'mir> ThreadManager<'mir, 'tcx> {
         self.threads[joined_thread_id].join_status = ThreadJoinStatus::Joined;
         if self.threads[joined_thread_id].state != ThreadState::Terminated {
             // The joined thread is still running, we need to wait for it.
-            self.active_thread_mut().state = ThreadState::BlockedOnJoin(joined_thread_id);
+            self.active_thread_mut().state =
+                ThreadState::Blocked(BlockReason::Join(joined_thread_id));
             trace!(
                 "{:?} blocked on {:?} when trying to join",
                 self.active_thread,
@@ -569,10 +585,11 @@ impl<'mir, 'tcx: 'mir> ThreadManager<'mir, 'tcx> {
             throw_ub_format!("trying to join itself");
         }
 
+        // Sanity check `join_status`.
         assert!(
-            self.threads
-                .iter()
-                .all(|thread| thread.state != ThreadState::BlockedOnJoin(joined_thread_id)),
+            self.threads.iter().all(|thread| {
+                thread.state != ThreadState::Blocked(BlockReason::Join(joined_thread_id))
+            }),
             "this thread already has threads waiting for its termination"
         );
 
@@ -594,16 +611,17 @@ impl<'mir, 'tcx: 'mir> ThreadManager<'mir, 'tcx> {
     }
 
     /// Put the thread into the blocked state.
-    fn block_thread(&mut self, thread: ThreadId) {
+    fn block_thread(&mut self, thread: ThreadId, reason: BlockReason) {
         let state = &mut self.threads[thread].state;
         assert_eq!(*state, ThreadState::Enabled);
-        *state = ThreadState::BlockedOnSync;
+        *state = ThreadState::Blocked(reason);
     }
 
     /// Put the blocked thread into the enabled state.
-    fn unblock_thread(&mut self, thread: ThreadId) {
+    /// Sanity-checks that the thread previously was blocked for the right reason.
+    fn unblock_thread(&mut self, thread: ThreadId, reason: BlockReason) {
         let state = &mut self.threads[thread].state;
-        assert_eq!(*state, ThreadState::BlockedOnSync);
+        assert_eq!(*state, ThreadState::Blocked(reason));
         *state = ThreadState::Enabled;
     }
 
@@ -622,7 +640,7 @@ impl<'mir, 'tcx: 'mir> ThreadManager<'mir, 'tcx> {
     fn register_timeout_callback(
         &mut self,
         thread: ThreadId,
-        call_time: Time,
+        call_time: CallbackTime,
         callback: TimeoutCallback<'mir, 'tcx>,
     ) {
         self.timeout_callbacks
@@ -683,7 +701,7 @@ impl<'mir, 'tcx: 'mir> ThreadManager<'mir, 'tcx> {
         // Check if we need to unblock any threads.
         let mut joined_threads = vec![]; // store which threads joined, we'll need it
         for (i, thread) in self.threads.iter_enumerated_mut() {
-            if thread.state == ThreadState::BlockedOnJoin(self.active_thread) {
+            if thread.state == ThreadState::Blocked(BlockReason::Join(self.active_thread)) {
                 // The thread has terminated, mark happens-before edge to joining thread
                 if data_race.is_some() {
                     joined_threads.push(i);
@@ -999,13 +1017,13 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
     }
 
     #[inline]
-    fn block_thread(&mut self, thread: ThreadId) {
-        self.eval_context_mut().machine.threads.block_thread(thread);
+    fn block_thread(&mut self, thread: ThreadId, reason: BlockReason) {
+        self.eval_context_mut().machine.threads.block_thread(thread, reason);
     }
 
     #[inline]
-    fn unblock_thread(&mut self, thread: ThreadId) {
-        self.eval_context_mut().machine.threads.unblock_thread(thread);
+    fn unblock_thread(&mut self, thread: ThreadId, reason: BlockReason) {
+        self.eval_context_mut().machine.threads.unblock_thread(thread, reason);
     }
 
     #[inline]
@@ -1027,11 +1045,11 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
     fn register_timeout_callback(
         &mut self,
         thread: ThreadId,
-        call_time: Time,
+        call_time: CallbackTime,
         callback: TimeoutCallback<'mir, 'tcx>,
     ) {
         let this = self.eval_context_mut();
-        if !this.machine.communicate() && matches!(call_time, Time::RealTime(..)) {
+        if !this.machine.communicate() && matches!(call_time, CallbackTime::RealTime(..)) {
             panic!("cannot have `RealTime` callback with isolation enabled!")
         }
         this.machine.threads.register_timeout_callback(thread, call_time, callback);
diff --git a/src/tools/miri/src/lib.rs b/src/tools/miri/src/lib.rs
@@ -116,7 +116,9 @@ pub use crate::concurrency::{
     data_race::{AtomicFenceOrd, AtomicReadOrd, AtomicRwOrd, AtomicWriteOrd, EvalContextExt as _},
     init_once::{EvalContextExt as _, InitOnceId},
     sync::{CondvarId, EvalContextExt as _, MutexId, RwLockId, SyncId},
-    thread::{EvalContextExt as _, StackEmptyCallback, ThreadId, ThreadManager, Time},
+    thread::{
+        BlockReason, CallbackTime, EvalContextExt as _, StackEmptyCallback, ThreadId, ThreadManager,
+    },
 };
 pub use crate::diagnostics::{
     report_error, EvalContextExt as _, NonHaltingDiagnostic, TerminationInfo,
diff --git a/src/tools/miri/src/shims/time.rs b/src/tools/miri/src/shims/time.rs
@@ -236,11 +236,11 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
             .unwrap_or_else(|| now.checked_add(Duration::from_secs(3600)).unwrap());
 
         let active_thread = this.get_active_thread();
-        this.block_thread(active_thread);
+        this.block_thread(active_thread, BlockReason::Sleep);
 
         this.register_timeout_callback(
             active_thread,
-            Time::Monotonic(timeout_time),
+            CallbackTime::Monotonic(timeout_time),
             Box::new(UnblockCallback { thread_to_unblock: active_thread }),
         );
 
@@ -259,11 +259,11 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
         let timeout_time = this.machine.clock.now().checked_add(duration).unwrap();
 
         let active_thread = this.get_active_thread();
-        this.block_thread(active_thread);
+        this.block_thread(active_thread, BlockReason::Sleep);
 
         this.register_timeout_callback(
             active_thread,
-            Time::Monotonic(timeout_time),
+            CallbackTime::Monotonic(timeout_time),
             Box::new(UnblockCallback { thread_to_unblock: active_thread }),
         );
 
@@ -281,7 +281,7 @@ impl VisitProvenance for UnblockCallback {
 
 impl<'mir, 'tcx: 'mir> MachineCallback<'mir, 'tcx> for UnblockCallback {
     fn call(&self, ecx: &mut MiriInterpCx<'mir, 'tcx>) -> InterpResult<'tcx> {
-        ecx.unblock_thread(self.thread_to_unblock);
+        ecx.unblock_thread(self.thread_to_unblock, BlockReason::Sleep);
         Ok(())
     }
 }
diff --git a/src/tools/miri/src/shims/unix/linux/sync.rs b/src/tools/miri/src/shims/unix/linux/sync.rs
@@ -107,16 +107,22 @@ pub fn futex<'tcx>(
                 Some(if wait_bitset {
                     // FUTEX_WAIT_BITSET uses an absolute timestamp.
                     if realtime {
-                        Time::RealTime(SystemTime::UNIX_EPOCH.checked_add(duration).unwrap())
+                        CallbackTime::RealTime(
+                            SystemTime::UNIX_EPOCH.checked_add(duration).unwrap(),
+                        )
                     } else {
-                        Time::Monotonic(this.machine.clock.anchor().checked_add(duration).unwrap())
+                        CallbackTime::Monotonic(
+                            this.machine.clock.anchor().checked_add(duration).unwrap(),
+                        )
                     }
                 } else {
                     // FUTEX_WAIT uses a relative timestamp.
                     if realtime {
-                        Time::RealTime(SystemTime::now().checked_add(duration).unwrap())
+                        CallbackTime::RealTime(SystemTime::now().checked_add(duration).unwrap())
                     } else {
-                        Time::Monotonic(this.machine.clock.now().checked_add(duration).unwrap())
+                        CallbackTime::Monotonic(
+                            this.machine.clock.now().checked_add(duration).unwrap(),
+                        )
                     }
                 })
             };
@@ -169,7 +175,7 @@ pub fn futex<'tcx>(
             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);
+                this.block_thread(thread, BlockReason::Futex { addr: addr_usize });
                 this.futex_wait(addr_usize, thread, bitset);
                 // Succesfully waking up from FUTEX_WAIT always returns zero.
                 this.write_scalar(Scalar::from_target_isize(0, this), dest)?;
@@ -191,7 +197,10 @@ pub fn futex<'tcx>(
 
                     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.unblock_thread(
+                                self.thread,
+                                BlockReason::Futex { addr: self.addr_usize },
+                            );
                             this.futex_remove_waiter(self.addr_usize, self.thread);
                             let etimedout = this.eval_libc("ETIMEDOUT");
                             this.set_last_error(etimedout)?;
@@ -249,7 +258,7 @@ pub fn futex<'tcx>(
             #[allow(clippy::arithmetic_side_effects)]
             for _ in 0..val {
                 if let Some(thread) = this.futex_wake(addr_usize, bitset) {
-                    this.unblock_thread(thread);
+                    this.unblock_thread(thread, BlockReason::Futex { addr: addr_usize });
                     this.unregister_timeout_callback_if_exists(thread);
                     n += 1;
                 } else {
diff --git a/src/tools/miri/src/shims/unix/sync.rs b/src/tools/miri/src/shims/unix/sync.rs
diff --git a/src/tools/miri/src/shims/windows/sync.rs b/src/tools/miri/src/shims/windows/sync.rs
