miri/concurrency/
thread.rs

1//! Implements threads.
2
3use std::mem;
4use std::sync::atomic::Ordering::Relaxed;
5use std::task::Poll;
6use std::time::{Duration, SystemTime};
7
8use either::Either;
9use rand::seq::IteratorRandom;
10use rustc_abi::ExternAbi;
11use rustc_const_eval::CTRL_C_RECEIVED;
12use rustc_data_structures::fx::FxHashMap;
13use rustc_hir::def_id::DefId;
14use rustc_index::{Idx, IndexVec};
15use rustc_middle::mir::Mutability;
16use rustc_middle::ty::layout::TyAndLayout;
17use rustc_span::Span;
18
19use crate::concurrency::GlobalDataRaceHandler;
20use crate::shims::tls;
21use crate::*;
22
23#[derive(Clone, Copy, Debug, PartialEq)]
24enum SchedulingAction {
25    /// Execute step on the active thread.
26    ExecuteStep,
27    /// Execute a timeout callback.
28    ExecuteTimeoutCallback,
29    /// Wait for a bit, until there is a timeout to be called.
30    Sleep(Duration),
31}
32
33/// What to do with TLS allocations from terminated threads
34#[derive(Clone, Copy, Debug, PartialEq)]
35pub enum TlsAllocAction {
36    /// Deallocate backing memory of thread-local statics as usual
37    Deallocate,
38    /// Skip deallocating backing memory of thread-local statics and consider all memory reachable
39    /// from them as "allowed to leak" (like global `static`s).
40    Leak,
41}
42
43/// The argument type for the "unblock" callback, indicating why the thread got unblocked.
44#[derive(Clone, Copy, Debug, PartialEq)]
45pub enum UnblockKind {
46    /// Operation completed successfully, thread continues normal execution.
47    Ready,
48    /// The operation did not complete within its specified duration.
49    TimedOut,
50}
51
52/// Type alias for unblock callbacks, i.e. machine callbacks invoked when
53/// a thread gets unblocked.
54pub type DynUnblockCallback<'tcx> = DynMachineCallback<'tcx, UnblockKind>;
55
56/// A thread identifier.
57#[derive(Clone, Copy, Debug, PartialOrd, Ord, PartialEq, Eq, Hash)]
58pub struct ThreadId(u32);
59
60impl ThreadId {
61    pub fn to_u32(self) -> u32 {
62        self.0
63    }
64
65    /// Create a new thread id from a `u32` without checking if this thread exists.
66    pub fn new_unchecked(id: u32) -> Self {
67        Self(id)
68    }
69
70    pub const MAIN_THREAD: ThreadId = ThreadId(0);
71}
72
73impl Idx for ThreadId {
74    fn new(idx: usize) -> Self {
75        ThreadId(u32::try_from(idx).unwrap())
76    }
77
78    fn index(self) -> usize {
79        usize::try_from(self.0).unwrap()
80    }
81}
82
83impl From<ThreadId> for u64 {
84    fn from(t: ThreadId) -> Self {
85        t.0.into()
86    }
87}
88
89/// Keeps track of what the thread is blocked on.
90#[derive(Debug, Copy, Clone, PartialEq, Eq)]
91pub enum BlockReason {
92    /// The thread tried to join the specified thread and is blocked until that
93    /// thread terminates.
94    Join(ThreadId),
95    /// Waiting for time to pass.
96    Sleep,
97    /// Blocked on a mutex.
98    Mutex,
99    /// Blocked on a condition variable.
100    Condvar(CondvarId),
101    /// Blocked on a reader-writer lock.
102    RwLock(RwLockId),
103    /// Blocked on a Futex variable.
104    Futex,
105    /// Blocked on an InitOnce.
106    InitOnce(InitOnceId),
107    /// Blocked on epoll.
108    Epoll,
109    /// Blocked on eventfd.
110    Eventfd,
111    /// Blocked on unnamed_socket.
112    UnnamedSocket,
113}
114
115/// The state of a thread.
116enum ThreadState<'tcx> {
117    /// The thread is enabled and can be executed.
118    Enabled,
119    /// The thread is blocked on something.
120    Blocked { reason: BlockReason, timeout: Option<Timeout>, callback: DynUnblockCallback<'tcx> },
121    /// The thread has terminated its execution. We do not delete terminated
122    /// threads (FIXME: why?).
123    Terminated,
124}
125
126impl<'tcx> std::fmt::Debug for ThreadState<'tcx> {
127    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
128        match self {
129            Self::Enabled => write!(f, "Enabled"),
130            Self::Blocked { reason, timeout, .. } =>
131                f.debug_struct("Blocked").field("reason", reason).field("timeout", timeout).finish(),
132            Self::Terminated => write!(f, "Terminated"),
133        }
134    }
135}
136
137impl<'tcx> ThreadState<'tcx> {
138    fn is_enabled(&self) -> bool {
139        matches!(self, ThreadState::Enabled)
140    }
141
142    fn is_terminated(&self) -> bool {
143        matches!(self, ThreadState::Terminated)
144    }
145
146    fn is_blocked_on(&self, reason: BlockReason) -> bool {
147        matches!(*self, ThreadState::Blocked { reason: actual_reason, .. } if actual_reason == reason)
148    }
149}
150
151/// The join status of a thread.
152#[derive(Debug, Copy, Clone, PartialEq, Eq)]
153enum ThreadJoinStatus {
154    /// The thread can be joined.
155    Joinable,
156    /// A thread is detached if its join handle was destroyed and no other
157    /// thread can join it.
158    Detached,
159    /// The thread was already joined by some thread and cannot be joined again.
160    Joined,
161}
162
163/// A thread.
164pub struct Thread<'tcx> {
165    state: ThreadState<'tcx>,
166
167    /// Name of the thread.
168    thread_name: Option<Vec<u8>>,
169
170    /// The virtual call stack.
171    stack: Vec<Frame<'tcx, Provenance, FrameExtra<'tcx>>>,
172
173    /// The function to call when the stack ran empty, to figure out what to do next.
174    /// Conceptually, this is the interpreter implementation of the things that happen 'after' the
175    /// Rust language entry point for this thread returns (usually implemented by the C or OS runtime).
176    /// (`None` is an error, it means the callback has not been set up yet or is actively running.)
177    pub(crate) on_stack_empty: Option<StackEmptyCallback<'tcx>>,
178
179    /// The index of the topmost user-relevant frame in `stack`. This field must contain
180    /// the value produced by `get_top_user_relevant_frame`.
181    /// The `None` state here represents
182    /// This field is a cache to reduce how often we call that method. The cache is manually
183    /// maintained inside `MiriMachine::after_stack_push` and `MiriMachine::after_stack_pop`.
184    top_user_relevant_frame: Option<usize>,
185
186    /// The join status.
187    join_status: ThreadJoinStatus,
188
189    /// Stack of active panic payloads for the current thread. Used for storing
190    /// the argument of the call to `miri_start_unwind` (the panic payload) when unwinding.
191    /// This is pointer-sized, and matches the `Payload` type in `src/libpanic_unwind/miri.rs`.
192    ///
193    /// In real unwinding, the payload gets passed as an argument to the landing pad,
194    /// which then forwards it to 'Resume'. However this argument is implicit in MIR,
195    /// so we have to store it out-of-band. When there are multiple active unwinds,
196    /// the innermost one is always caught first, so we can store them as a stack.
197    pub(crate) panic_payloads: Vec<ImmTy<'tcx>>,
198
199    /// Last OS error location in memory. It is a 32-bit integer.
200    pub(crate) last_error: Option<MPlaceTy<'tcx>>,
201}
202
203pub type StackEmptyCallback<'tcx> =
204    Box<dyn FnMut(&mut MiriInterpCx<'tcx>) -> InterpResult<'tcx, Poll<()>> + 'tcx>;
205
206impl<'tcx> Thread<'tcx> {
207    /// Get the name of the current thread if it was set.
208    fn thread_name(&self) -> Option<&[u8]> {
209        self.thread_name.as_deref()
210    }
211
212    /// Get the name of the current thread for display purposes; will include thread ID if not set.
213    fn thread_display_name(&self, id: ThreadId) -> String {
214        if let Some(ref thread_name) = self.thread_name {
215            String::from_utf8_lossy(thread_name).into_owned()
216        } else {
217            format!("unnamed-{}", id.index())
218        }
219    }
220
221    /// Return the top user-relevant frame, if there is one.
222    /// Note that the choice to return `None` here when there is no user-relevant frame is part of
223    /// justifying the optimization that only pushes of user-relevant frames require updating the
224    /// `top_user_relevant_frame` field.
225    fn compute_top_user_relevant_frame(&self) -> Option<usize> {
226        self.stack
227            .iter()
228            .enumerate()
229            .rev()
230            .find_map(|(idx, frame)| if frame.extra.is_user_relevant { Some(idx) } else { None })
231    }
232
233    /// Re-compute the top user-relevant frame from scratch.
234    pub fn recompute_top_user_relevant_frame(&mut self) {
235        self.top_user_relevant_frame = self.compute_top_user_relevant_frame();
236    }
237
238    /// Set the top user-relevant frame to the given value. Must be equal to what
239    /// `get_top_user_relevant_frame` would return!
240    pub fn set_top_user_relevant_frame(&mut self, frame_idx: usize) {
241        debug_assert_eq!(Some(frame_idx), self.compute_top_user_relevant_frame());
242        self.top_user_relevant_frame = Some(frame_idx);
243    }
244
245    /// Returns the topmost frame that is considered user-relevant, or the
246    /// top of the stack if there is no such frame, or `None` if the stack is empty.
247    pub fn top_user_relevant_frame(&self) -> Option<usize> {
248        debug_assert_eq!(self.top_user_relevant_frame, self.compute_top_user_relevant_frame());
249        // This can be called upon creation of an allocation. We create allocations while setting up
250        // parts of the Rust runtime when we do not have any stack frames yet, so we need to handle
251        // empty stacks.
252        self.top_user_relevant_frame.or_else(|| self.stack.len().checked_sub(1))
253    }
254
255    pub fn current_span(&self) -> Span {
256        self.top_user_relevant_frame()
257            .map(|frame_idx| self.stack[frame_idx].current_span())
258            .unwrap_or(rustc_span::DUMMY_SP)
259    }
260}
261
262impl<'tcx> std::fmt::Debug for Thread<'tcx> {
263    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
264        write!(
265            f,
266            "{}({:?}, {:?})",
267            String::from_utf8_lossy(self.thread_name().unwrap_or(b"<unnamed>")),
268            self.state,
269            self.join_status
270        )
271    }
272}
273
274impl<'tcx> Thread<'tcx> {
275    fn new(name: Option<&str>, on_stack_empty: Option<StackEmptyCallback<'tcx>>) -> Self {
276        Self {
277            state: ThreadState::Enabled,
278            thread_name: name.map(|name| Vec::from(name.as_bytes())),
279            stack: Vec::new(),
280            top_user_relevant_frame: None,
281            join_status: ThreadJoinStatus::Joinable,
282            panic_payloads: Vec::new(),
283            last_error: None,
284            on_stack_empty,
285        }
286    }
287}
288
289impl VisitProvenance for Thread<'_> {
290    fn visit_provenance(&self, visit: &mut VisitWith<'_>) {
291        let Thread {
292            panic_payloads: panic_payload,
293            last_error,
294            stack,
295            top_user_relevant_frame: _,
296            state: _,
297            thread_name: _,
298            join_status: _,
299            on_stack_empty: _, // we assume the closure captures no GC-relevant state
300        } = self;
301
302        for payload in panic_payload {
303            payload.visit_provenance(visit);
304        }
305        last_error.visit_provenance(visit);
306        for frame in stack {
307            frame.visit_provenance(visit)
308        }
309    }
310}
311
312impl VisitProvenance for Frame<'_, Provenance, FrameExtra<'_>> {
313    fn visit_provenance(&self, visit: &mut VisitWith<'_>) {
314        let Frame {
315            return_place,
316            locals,
317            extra,
318            // There are some private fields we cannot access; they contain no tags.
319            ..
320        } = self;
321
322        // Return place.
323        return_place.visit_provenance(visit);
324        // Locals.
325        for local in locals.iter() {
326            match local.as_mplace_or_imm() {
327                None => {}
328                Some(Either::Left((ptr, meta))) => {
329                    ptr.visit_provenance(visit);
330                    meta.visit_provenance(visit);
331                }
332                Some(Either::Right(imm)) => {
333                    imm.visit_provenance(visit);
334                }
335            }
336        }
337
338        extra.visit_provenance(visit);
339    }
340}
341
342/// The moment in time when a blocked thread should be woken up.
343#[derive(Debug)]
344enum Timeout {
345    Monotonic(Instant),
346    RealTime(SystemTime),
347}
348
349impl Timeout {
350    /// How long do we have to wait from now until the specified time?
351    fn get_wait_time(&self, clock: &MonotonicClock) -> Duration {
352        match self {
353            Timeout::Monotonic(instant) => instant.duration_since(clock.now()),
354            Timeout::RealTime(time) =>
355                time.duration_since(SystemTime::now()).unwrap_or(Duration::ZERO),
356        }
357    }
358
359    /// Will try to add `duration`, but if that overflows it may add less.
360    fn add_lossy(&self, duration: Duration) -> Self {
361        match self {
362            Timeout::Monotonic(i) => Timeout::Monotonic(i.add_lossy(duration)),
363            Timeout::RealTime(s) => {
364                // If this overflows, try adding just 1h and assume that will not overflow.
365                Timeout::RealTime(
366                    s.checked_add(duration)
367                        .unwrap_or_else(|| s.checked_add(Duration::from_secs(3600)).unwrap()),
368                )
369            }
370        }
371    }
372}
373
374/// The clock to use for the timeout you are asking for.
375#[derive(Debug, Copy, Clone)]
376pub enum TimeoutClock {
377    Monotonic,
378    RealTime,
379}
380
381/// Whether the timeout is relative or absolute.
382#[derive(Debug, Copy, Clone)]
383pub enum TimeoutAnchor {
384    Relative,
385    Absolute,
386}
387
388/// An error signaling that the requested thread doesn't exist.
389#[derive(Debug, Copy, Clone)]
390pub struct ThreadNotFound;
391
392/// A set of threads.
393#[derive(Debug)]
394pub struct ThreadManager<'tcx> {
395    /// Identifier of the currently active thread.
396    active_thread: ThreadId,
397    /// Threads used in the program.
398    ///
399    /// Note that this vector also contains terminated threads.
400    threads: IndexVec<ThreadId, Thread<'tcx>>,
401    /// A mapping from a thread-local static to the thread specific allocation.
402    thread_local_allocs: FxHashMap<(DefId, ThreadId), StrictPointer>,
403    /// A flag that indicates that we should change the active thread.
404    yield_active_thread: bool,
405    /// A flag that indicates that we should do round robin scheduling of threads else randomized scheduling is used.
406    fixed_scheduling: bool,
407}
408
409impl VisitProvenance for ThreadManager<'_> {
410    fn visit_provenance(&self, visit: &mut VisitWith<'_>) {
411        let ThreadManager {
412            threads,
413            thread_local_allocs,
414            active_thread: _,
415            yield_active_thread: _,
416            fixed_scheduling: _,
417        } = self;
418
419        for thread in threads {
420            thread.visit_provenance(visit);
421        }
422        for ptr in thread_local_allocs.values() {
423            ptr.visit_provenance(visit);
424        }
425    }
426}
427
428impl<'tcx> ThreadManager<'tcx> {
429    pub(crate) fn new(config: &MiriConfig) -> Self {
430        let mut threads = IndexVec::new();
431        // Create the main thread and add it to the list of threads.
432        threads.push(Thread::new(Some("main"), None));
433        Self {
434            active_thread: ThreadId::MAIN_THREAD,
435            threads,
436            thread_local_allocs: Default::default(),
437            yield_active_thread: false,
438            fixed_scheduling: config.fixed_scheduling,
439        }
440    }
441
442    pub(crate) fn init(
443        ecx: &mut MiriInterpCx<'tcx>,
444        on_main_stack_empty: StackEmptyCallback<'tcx>,
445    ) {
446        ecx.machine.threads.threads[ThreadId::MAIN_THREAD].on_stack_empty =
447            Some(on_main_stack_empty);
448        if ecx.tcx.sess.target.os.as_ref() != "windows" {
449            // The main thread can *not* be joined on except on windows.
450            ecx.machine.threads.threads[ThreadId::MAIN_THREAD].join_status =
451                ThreadJoinStatus::Detached;
452        }
453    }
454
455    pub fn thread_id_try_from(&self, id: impl TryInto<u32>) -> Result<ThreadId, ThreadNotFound> {
456        if let Ok(id) = id.try_into()
457            && usize::try_from(id).is_ok_and(|id| id < self.threads.len())
458        {
459            Ok(ThreadId(id))
460        } else {
461            Err(ThreadNotFound)
462        }
463    }
464
465    /// Check if we have an allocation for the given thread local static for the
466    /// active thread.
467    fn get_thread_local_alloc_id(&self, def_id: DefId) -> Option<StrictPointer> {
468        self.thread_local_allocs.get(&(def_id, self.active_thread)).cloned()
469    }
470
471    /// Set the pointer for the allocation of the given thread local
472    /// static for the active thread.
473    ///
474    /// Panics if a thread local is initialized twice for the same thread.
475    fn set_thread_local_alloc(&mut self, def_id: DefId, ptr: StrictPointer) {
476        self.thread_local_allocs.try_insert((def_id, self.active_thread), ptr).unwrap();
477    }
478
479    /// Borrow the stack of the active thread.
480    pub fn active_thread_stack(&self) -> &[Frame<'tcx, Provenance, FrameExtra<'tcx>>] {
481        &self.threads[self.active_thread].stack
482    }
483
484    /// Mutably borrow the stack of the active thread.
485    pub fn active_thread_stack_mut(
486        &mut self,
487    ) -> &mut Vec<Frame<'tcx, Provenance, FrameExtra<'tcx>>> {
488        &mut self.threads[self.active_thread].stack
489    }
490
491    pub fn all_stacks(
492        &self,
493    ) -> impl Iterator<Item = (ThreadId, &[Frame<'tcx, Provenance, FrameExtra<'tcx>>])> {
494        self.threads.iter_enumerated().map(|(id, t)| (id, &t.stack[..]))
495    }
496
497    /// Create a new thread and returns its id.
498    fn create_thread(&mut self, on_stack_empty: StackEmptyCallback<'tcx>) -> ThreadId {
499        let new_thread_id = ThreadId::new(self.threads.len());
500        self.threads.push(Thread::new(None, Some(on_stack_empty)));
501        new_thread_id
502    }
503
504    /// Set an active thread and return the id of the thread that was active before.
505    fn set_active_thread_id(&mut self, id: ThreadId) -> ThreadId {
506        assert!(id.index() < self.threads.len());
507        info!(
508            "---------- Now executing on thread `{}` (previous: `{}`) ----------------------------------------",
509            self.get_thread_display_name(id),
510            self.get_thread_display_name(self.active_thread)
511        );
512        std::mem::replace(&mut self.active_thread, id)
513    }
514
515    /// Get the id of the currently active thread.
516    pub fn active_thread(&self) -> ThreadId {
517        self.active_thread
518    }
519
520    /// Get the total number of threads that were ever spawn by this program.
521    pub fn get_total_thread_count(&self) -> usize {
522        self.threads.len()
523    }
524
525    /// Get the total of threads that are currently live, i.e., not yet terminated.
526    /// (They might be blocked.)
527    pub fn get_live_thread_count(&self) -> usize {
528        self.threads.iter().filter(|t| !t.state.is_terminated()).count()
529    }
530
531    /// Has the given thread terminated?
532    fn has_terminated(&self, thread_id: ThreadId) -> bool {
533        self.threads[thread_id].state.is_terminated()
534    }
535
536    /// Have all threads terminated?
537    fn have_all_terminated(&self) -> bool {
538        self.threads.iter().all(|thread| thread.state.is_terminated())
539    }
540
541    /// Enable the thread for execution. The thread must be terminated.
542    fn enable_thread(&mut self, thread_id: ThreadId) {
543        assert!(self.has_terminated(thread_id));
544        self.threads[thread_id].state = ThreadState::Enabled;
545    }
546
547    /// Get a mutable borrow of the currently active thread.
548    pub fn active_thread_mut(&mut self) -> &mut Thread<'tcx> {
549        &mut self.threads[self.active_thread]
550    }
551
552    /// Get a shared borrow of the currently active thread.
553    pub fn active_thread_ref(&self) -> &Thread<'tcx> {
554        &self.threads[self.active_thread]
555    }
556
557    /// Mark the thread as detached, which means that no other thread will try
558    /// to join it and the thread is responsible for cleaning up.
559    ///
560    /// `allow_terminated_joined` allows detaching joined threads that have already terminated.
561    /// This matches Windows's behavior for `CloseHandle`.
562    ///
563    /// See <https://docs.microsoft.com/en-us/windows/win32/procthread/thread-handles-and-identifiers>:
564    /// > The handle is valid until closed, even after the thread it represents has been terminated.
565    fn detach_thread(&mut self, id: ThreadId, allow_terminated_joined: bool) -> InterpResult<'tcx> {
566        trace!("detaching {:?}", id);
567
568        let is_ub = if allow_terminated_joined && self.threads[id].state.is_terminated() {
569            // "Detached" in particular means "not yet joined". Redundant detaching is still UB.
570            self.threads[id].join_status == ThreadJoinStatus::Detached
571        } else {
572            self.threads[id].join_status != ThreadJoinStatus::Joinable
573        };
574        if is_ub {
575            throw_ub_format!("trying to detach thread that was already detached or joined");
576        }
577
578        self.threads[id].join_status = ThreadJoinStatus::Detached;
579        interp_ok(())
580    }
581
582    /// Mark that the active thread tries to join the thread with `joined_thread_id`.
583    fn join_thread(
584        &mut self,
585        joined_thread_id: ThreadId,
586        data_race_handler: &mut GlobalDataRaceHandler,
587    ) -> InterpResult<'tcx> {
588        if self.threads[joined_thread_id].join_status == ThreadJoinStatus::Detached {
589            // On Windows this corresponds to joining on a closed handle.
590            throw_ub_format!("trying to join a detached thread");
591        }
592
593        fn after_join<'tcx>(
594            threads: &mut ThreadManager<'_>,
595            joined_thread_id: ThreadId,
596            data_race_handler: &mut GlobalDataRaceHandler,
597        ) -> InterpResult<'tcx> {
598            match data_race_handler {
599                GlobalDataRaceHandler::None => {}
600                GlobalDataRaceHandler::Vclocks(data_race) =>
601                    data_race.thread_joined(threads, joined_thread_id),
602                GlobalDataRaceHandler::Genmc(genmc_ctx) =>
603                    genmc_ctx.handle_thread_join(threads.active_thread, joined_thread_id)?,
604            }
605            interp_ok(())
606        }
607
608        // Mark the joined thread as being joined so that we detect if other
609        // threads try to join it.
610        self.threads[joined_thread_id].join_status = ThreadJoinStatus::Joined;
611        if !self.threads[joined_thread_id].state.is_terminated() {
612            trace!(
613                "{:?} blocked on {:?} when trying to join",
614                self.active_thread, joined_thread_id
615            );
616            // The joined thread is still running, we need to wait for it.
617            // Unce we get unblocked, perform the appropriate synchronization.
618            self.block_thread(
619                BlockReason::Join(joined_thread_id),
620                None,
621                callback!(
622                    @capture<'tcx> {
623                        joined_thread_id: ThreadId,
624                    }
625                    |this, unblock: UnblockKind| {
626                        assert_eq!(unblock, UnblockKind::Ready);
627                        after_join(&mut this.machine.threads, joined_thread_id, &mut this.machine.data_race)
628                    }
629                ),
630            );
631        } else {
632            // The thread has already terminated - establish happens-before
633            after_join(self, joined_thread_id, data_race_handler)?;
634        }
635        interp_ok(())
636    }
637
638    /// Mark that the active thread tries to exclusively join the thread with `joined_thread_id`.
639    /// If the thread is already joined by another thread, it will throw UB
640    fn join_thread_exclusive(
641        &mut self,
642        joined_thread_id: ThreadId,
643        data_race_handler: &mut GlobalDataRaceHandler,
644    ) -> InterpResult<'tcx> {
645        if self.threads[joined_thread_id].join_status == ThreadJoinStatus::Joined {
646            throw_ub_format!("trying to join an already joined thread");
647        }
648
649        if joined_thread_id == self.active_thread {
650            throw_ub_format!("trying to join itself");
651        }
652
653        // Sanity check `join_status`.
654        assert!(
655            self.threads
656                .iter()
657                .all(|thread| { !thread.state.is_blocked_on(BlockReason::Join(joined_thread_id)) }),
658            "this thread already has threads waiting for its termination"
659        );
660
661        self.join_thread(joined_thread_id, data_race_handler)
662    }
663
664    /// Set the name of the given thread.
665    pub fn set_thread_name(&mut self, thread: ThreadId, new_thread_name: Vec<u8>) {
666        self.threads[thread].thread_name = Some(new_thread_name);
667    }
668
669    /// Get the name of the given thread.
670    pub fn get_thread_name(&self, thread: ThreadId) -> Option<&[u8]> {
671        self.threads[thread].thread_name()
672    }
673
674    pub fn get_thread_display_name(&self, thread: ThreadId) -> String {
675        self.threads[thread].thread_display_name(thread)
676    }
677
678    /// Put the thread into the blocked state.
679    fn block_thread(
680        &mut self,
681        reason: BlockReason,
682        timeout: Option<Timeout>,
683        callback: DynUnblockCallback<'tcx>,
684    ) {
685        let state = &mut self.threads[self.active_thread].state;
686        assert!(state.is_enabled());
687        *state = ThreadState::Blocked { reason, timeout, callback }
688    }
689
690    /// Change the active thread to some enabled thread.
691    fn yield_active_thread(&mut self) {
692        // We do not yield immediately, as swapping out the current stack while executing a MIR statement
693        // could lead to all sorts of confusion.
694        // We should only switch stacks between steps.
695        self.yield_active_thread = true;
696    }
697
698    /// Get the wait time for the next timeout, or `None` if no timeout is pending.
699    fn next_callback_wait_time(&self, clock: &MonotonicClock) -> Option<Duration> {
700        self.threads
701            .iter()
702            .filter_map(|t| {
703                match &t.state {
704                    ThreadState::Blocked { timeout: Some(timeout), .. } =>
705                        Some(timeout.get_wait_time(clock)),
706                    _ => None,
707                }
708            })
709            .min()
710    }
711}
712
713impl<'tcx> EvalContextPrivExt<'tcx> for MiriInterpCx<'tcx> {}
714trait EvalContextPrivExt<'tcx>: MiriInterpCxExt<'tcx> {
715    /// Execute a timeout callback on the callback's thread.
716    #[inline]
717    fn run_timeout_callback(&mut self) -> InterpResult<'tcx> {
718        let this = self.eval_context_mut();
719        let mut found_callback = None;
720        // Find a blocked thread that has timed out.
721        for (id, thread) in this.machine.threads.threads.iter_enumerated_mut() {
722            match &thread.state {
723                ThreadState::Blocked { timeout: Some(timeout), .. }
724                    if timeout.get_wait_time(&this.machine.monotonic_clock) == Duration::ZERO =>
725                {
726                    let old_state = mem::replace(&mut thread.state, ThreadState::Enabled);
727                    let ThreadState::Blocked { callback, .. } = old_state else { unreachable!() };
728                    found_callback = Some((id, callback));
729                    // Run the fallback (after the loop because borrow-checking).
730                    break;
731                }
732                _ => {}
733            }
734        }
735        if let Some((thread, callback)) = found_callback {
736            // This back-and-forth with `set_active_thread` is here because of two
737            // design decisions:
738            // 1. Make the caller and not the callback responsible for changing
739            //    thread.
740            // 2. Make the scheduler the only place that can change the active
741            //    thread.
742            let old_thread = this.machine.threads.set_active_thread_id(thread);
743            callback.call(this, UnblockKind::TimedOut)?;
744            this.machine.threads.set_active_thread_id(old_thread);
745        }
746        // found_callback can remain None if the computer's clock
747        // was shifted after calling the scheduler and before the call
748        // to get_ready_callback (see issue
749        // https://github.com/rust-lang/miri/issues/1763). In this case,
750        // just do nothing, which effectively just returns to the
751        // scheduler.
752        interp_ok(())
753    }
754
755    #[inline]
756    fn run_on_stack_empty(&mut self) -> InterpResult<'tcx, Poll<()>> {
757        let this = self.eval_context_mut();
758        // Inform GenMC that a thread has finished all user code. GenMC needs to know this for scheduling.
759        if let Some(genmc_ctx) = this.machine.data_race.as_genmc_ref() {
760            let thread_id = this.active_thread();
761            genmc_ctx.handle_thread_stack_empty(thread_id);
762        }
763        let mut callback = this
764            .active_thread_mut()
765            .on_stack_empty
766            .take()
767            .expect("`on_stack_empty` not set up, or already running");
768        let res = callback(this)?;
769        this.active_thread_mut().on_stack_empty = Some(callback);
770        interp_ok(res)
771    }
772
773    /// Decide which action to take next and on which thread.
774    ///
775    /// The currently implemented scheduling policy is the one that is commonly
776    /// used in stateless model checkers such as Loom: run the active thread as
777    /// long as we can and switch only when we have to (the active thread was
778    /// blocked, terminated, or has explicitly asked to be preempted).
779    ///
780    /// If GenMC mode is active, the scheduling is instead handled by GenMC.
781    fn schedule(&mut self) -> InterpResult<'tcx, SchedulingAction> {
782        let this = self.eval_context_mut();
783        // In GenMC mode, we let GenMC do the scheduling
784        if let Some(genmc_ctx) = this.machine.data_race.as_genmc_ref() {
785            let next_thread_id = genmc_ctx.schedule_thread(this)?;
786
787            let thread_manager = &mut this.machine.threads;
788            thread_manager.active_thread = next_thread_id;
789            thread_manager.yield_active_thread = false;
790
791            assert!(thread_manager.threads[thread_manager.active_thread].state.is_enabled());
792            return interp_ok(SchedulingAction::ExecuteStep);
793        }
794
795        // We are not in GenMC mode, so we control the schedule
796        let thread_manager = &mut this.machine.threads;
797        let clock = &this.machine.monotonic_clock;
798        let rng = this.machine.rng.get_mut();
799        // This thread and the program can keep going.
800        if thread_manager.threads[thread_manager.active_thread].state.is_enabled()
801            && !thread_manager.yield_active_thread
802        {
803            // The currently active thread is still enabled, just continue with it.
804            return interp_ok(SchedulingAction::ExecuteStep);
805        }
806        // The active thread yielded or got terminated. Let's see if there are any timeouts to take
807        // care of. We do this *before* running any other thread, to ensure that timeouts "in the
808        // past" fire before any other thread can take an action. This ensures that for
809        // `pthread_cond_timedwait`, "an error is returned if [...] the absolute time specified by
810        // abstime has already been passed at the time of the call".
811        // <https://pubs.opengroup.org/onlinepubs/9699919799/functions/pthread_cond_timedwait.html>
812        let potential_sleep_time = thread_manager.next_callback_wait_time(clock);
813        if potential_sleep_time == Some(Duration::ZERO) {
814            return interp_ok(SchedulingAction::ExecuteTimeoutCallback);
815        }
816        // No callbacks immediately scheduled, pick a regular thread to execute.
817        // The active thread blocked or yielded. So we go search for another enabled thread.
818        // We build the list of threads by starting with the threads after the current one, followed by
819        // the threads before the current one and then the current thread itself (i.e., this iterator acts
820        // like `threads.rotate_left(self.active_thread.index() + 1)`. This ensures that if we pick the first
821        // eligible thread, we do regular round-robin scheduling, and all threads get a chance to take a step.
822        let mut threads_iter = thread_manager
823            .threads
824            .iter_enumerated()
825            .skip(thread_manager.active_thread.index() + 1)
826            .chain(
827                thread_manager
828                    .threads
829                    .iter_enumerated()
830                    .take(thread_manager.active_thread.index() + 1),
831            )
832            .filter(|(_id, thread)| thread.state.is_enabled());
833        // Pick a new thread, and switch to it.
834        let new_thread = if thread_manager.fixed_scheduling {
835            threads_iter.next()
836        } else {
837            threads_iter.choose(rng)
838        };
839
840        if let Some((id, _thread)) = new_thread {
841            if thread_manager.active_thread != id {
842                info!(
843                    "---------- Now executing on thread `{}` (previous: `{}`) ----------------------------------------",
844                    thread_manager.get_thread_display_name(id),
845                    thread_manager.get_thread_display_name(thread_manager.active_thread)
846                );
847                thread_manager.active_thread = id;
848            }
849        }
850        // This completes the `yield`, if any was requested.
851        thread_manager.yield_active_thread = false;
852
853        if thread_manager.threads[thread_manager.active_thread].state.is_enabled() {
854            return interp_ok(SchedulingAction::ExecuteStep);
855        }
856        // We have not found a thread to execute.
857        if thread_manager.threads.iter().all(|thread| thread.state.is_terminated()) {
858            unreachable!("all threads terminated without the main thread terminating?!");
859        } else if let Some(sleep_time) = potential_sleep_time {
860            // All threads are currently blocked, but we have unexecuted
861            // timeout_callbacks, which may unblock some of the threads. Hence,
862            // sleep until the first callback.
863            interp_ok(SchedulingAction::Sleep(sleep_time))
864        } else {
865            throw_machine_stop!(TerminationInfo::Deadlock);
866        }
867    }
868}
869
870// Public interface to thread management.
871impl<'tcx> EvalContextExt<'tcx> for crate::MiriInterpCx<'tcx> {}
872pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
873    #[inline]
874    fn thread_id_try_from(&self, id: impl TryInto<u32>) -> Result<ThreadId, ThreadNotFound> {
875        self.eval_context_ref().machine.threads.thread_id_try_from(id)
876    }
877
878    /// Get a thread-specific allocation id for the given thread-local static.
879    /// If needed, allocate a new one.
880    fn get_or_create_thread_local_alloc(
881        &mut self,
882        def_id: DefId,
883    ) -> InterpResult<'tcx, StrictPointer> {
884        let this = self.eval_context_mut();
885        let tcx = this.tcx;
886        if let Some(old_alloc) = this.machine.threads.get_thread_local_alloc_id(def_id) {
887            // We already have a thread-specific allocation id for this
888            // thread-local static.
889            interp_ok(old_alloc)
890        } else {
891            // We need to allocate a thread-specific allocation id for this
892            // thread-local static.
893            // First, we compute the initial value for this static.
894            if tcx.is_foreign_item(def_id) {
895                throw_unsup_format!("foreign thread-local statics are not supported");
896            }
897            let alloc = this.ctfe_query(|tcx| tcx.eval_static_initializer(def_id))?;
898            // We make a full copy of this allocation.
899            let mut alloc = alloc.inner().adjust_from_tcx(
900                &this.tcx,
901                |bytes, align| {
902                    interp_ok(MiriAllocBytes::from_bytes(std::borrow::Cow::Borrowed(bytes), align))
903                },
904                |ptr| this.global_root_pointer(ptr),
905            )?;
906            // This allocation will be deallocated when the thread dies, so it is not in read-only memory.
907            alloc.mutability = Mutability::Mut;
908            // Create a fresh allocation with this content.
909            let ptr = this.insert_allocation(alloc, MiriMemoryKind::Tls.into())?;
910            this.machine.threads.set_thread_local_alloc(def_id, ptr);
911            interp_ok(ptr)
912        }
913    }
914
915    /// Start a regular (non-main) thread.
916    #[inline]
917    fn start_regular_thread(
918        &mut self,
919        thread: Option<MPlaceTy<'tcx>>,
920        start_routine: Pointer,
921        start_abi: ExternAbi,
922        func_arg: ImmTy<'tcx>,
923        ret_layout: TyAndLayout<'tcx>,
924    ) -> InterpResult<'tcx, ThreadId> {
925        let this = self.eval_context_mut();
926
927        // Create the new thread
928        let new_thread_id = this.machine.threads.create_thread({
929            let mut state = tls::TlsDtorsState::default();
930            Box::new(move |m| state.on_stack_empty(m))
931        });
932        let current_span = this.machine.current_span();
933        match &mut this.machine.data_race {
934            GlobalDataRaceHandler::None => {}
935            GlobalDataRaceHandler::Vclocks(data_race) =>
936                data_race.thread_created(&this.machine.threads, new_thread_id, current_span),
937            GlobalDataRaceHandler::Genmc(genmc_ctx) =>
938                genmc_ctx.handle_thread_create(&this.machine.threads, new_thread_id)?,
939        }
940        // Write the current thread-id, switch to the next thread later
941        // to treat this write operation as occurring on the current thread.
942        if let Some(thread_info_place) = thread {
943            this.write_scalar(
944                Scalar::from_uint(new_thread_id.to_u32(), thread_info_place.layout.size),
945                &thread_info_place,
946            )?;
947        }
948
949        // Finally switch to new thread so that we can push the first stackframe.
950        // After this all accesses will be treated as occurring in the new thread.
951        let old_thread_id = this.machine.threads.set_active_thread_id(new_thread_id);
952
953        // The child inherits its parent's cpu affinity.
954        if let Some(cpuset) = this.machine.thread_cpu_affinity.get(&old_thread_id).cloned() {
955            this.machine.thread_cpu_affinity.insert(new_thread_id, cpuset);
956        }
957
958        // Perform the function pointer load in the new thread frame.
959        let instance = this.get_ptr_fn(start_routine)?.as_instance()?;
960
961        // Note: the returned value is currently ignored (see the FIXME in
962        // pthread_join in shims/unix/thread.rs) because the Rust standard library does not use
963        // it.
964        let ret_place = this.allocate(ret_layout, MiriMemoryKind::Machine.into())?;
965
966        this.call_function(
967            instance,
968            start_abi,
969            &[func_arg],
970            Some(&ret_place),
971            StackPopCleanup::Root { cleanup: true },
972        )?;
973
974        // Restore the old active thread frame.
975        this.machine.threads.set_active_thread_id(old_thread_id);
976
977        interp_ok(new_thread_id)
978    }
979
980    /// Handles thread termination of the active thread: wakes up threads joining on this one,
981    /// and deals with the thread's thread-local statics according to `tls_alloc_action`.
982    ///
983    /// This is called by the eval loop when a thread's on_stack_empty returns `Ready`.
984    fn terminate_active_thread(&mut self, tls_alloc_action: TlsAllocAction) -> InterpResult<'tcx> {
985        let this = self.eval_context_mut();
986
987        // Mark thread as terminated.
988        let thread = this.active_thread_mut();
989        assert!(thread.stack.is_empty(), "only threads with an empty stack can be terminated");
990        thread.state = ThreadState::Terminated;
991        match &mut this.machine.data_race {
992            GlobalDataRaceHandler::None => {}
993            GlobalDataRaceHandler::Vclocks(data_race) =>
994                data_race.thread_terminated(&this.machine.threads),
995            GlobalDataRaceHandler::Genmc(genmc_ctx) =>
996                genmc_ctx.handle_thread_finish(&this.machine.threads)?,
997        }
998        // Deallocate TLS.
999        let gone_thread = this.active_thread();
1000        {
1001            let mut free_tls_statics = Vec::new();
1002            this.machine.threads.thread_local_allocs.retain(|&(_def_id, thread), &mut alloc_id| {
1003                if thread != gone_thread {
1004                    // A different thread, keep this static around.
1005                    return true;
1006                }
1007                // Delete this static from the map and from memory.
1008                // We cannot free directly here as we cannot use `?` in this context.
1009                free_tls_statics.push(alloc_id);
1010                false
1011            });
1012            // Now free the TLS statics.
1013            for ptr in free_tls_statics {
1014                match tls_alloc_action {
1015                    TlsAllocAction::Deallocate =>
1016                        this.deallocate_ptr(ptr.into(), None, MiriMemoryKind::Tls.into())?,
1017                    TlsAllocAction::Leak =>
1018                        if let Some(alloc) = ptr.provenance.get_alloc_id() {
1019                            trace!(
1020                                "Thread-local static leaked and stored as static root: {:?}",
1021                                alloc
1022                            );
1023                            this.machine.static_roots.push(alloc);
1024                        },
1025                }
1026            }
1027        }
1028        // Unblock joining threads.
1029        let unblock_reason = BlockReason::Join(gone_thread);
1030        let threads = &this.machine.threads.threads;
1031        let joining_threads = threads
1032            .iter_enumerated()
1033            .filter(|(_, thread)| thread.state.is_blocked_on(unblock_reason))
1034            .map(|(id, _)| id)
1035            .collect::<Vec<_>>();
1036        for thread in joining_threads {
1037            this.unblock_thread(thread, unblock_reason)?;
1038        }
1039
1040        interp_ok(())
1041    }
1042
1043    /// Block the current thread, with an optional timeout.
1044    /// The callback will be invoked when the thread gets unblocked.
1045    #[inline]
1046    fn block_thread(
1047        &mut self,
1048        reason: BlockReason,
1049        timeout: Option<(TimeoutClock, TimeoutAnchor, Duration)>,
1050        callback: DynUnblockCallback<'tcx>,
1051    ) {
1052        let this = self.eval_context_mut();
1053        let timeout = timeout.map(|(clock, anchor, duration)| {
1054            let anchor = match clock {
1055                TimeoutClock::RealTime => {
1056                    assert!(
1057                        this.machine.communicate(),
1058                        "cannot have `RealTime` timeout with isolation enabled!"
1059                    );
1060                    Timeout::RealTime(match anchor {
1061                        TimeoutAnchor::Absolute => SystemTime::UNIX_EPOCH,
1062                        TimeoutAnchor::Relative => SystemTime::now(),
1063                    })
1064                }
1065                TimeoutClock::Monotonic =>
1066                    Timeout::Monotonic(match anchor {
1067                        TimeoutAnchor::Absolute => this.machine.monotonic_clock.epoch(),
1068                        TimeoutAnchor::Relative => this.machine.monotonic_clock.now(),
1069                    }),
1070            };
1071            anchor.add_lossy(duration)
1072        });
1073        this.machine.threads.block_thread(reason, timeout, callback);
1074    }
1075
1076    /// Put the blocked thread into the enabled state.
1077    /// Sanity-checks that the thread previously was blocked for the right reason.
1078    fn unblock_thread(&mut self, thread: ThreadId, reason: BlockReason) -> InterpResult<'tcx> {
1079        let this = self.eval_context_mut();
1080        let old_state =
1081            mem::replace(&mut this.machine.threads.threads[thread].state, ThreadState::Enabled);
1082        let callback = match old_state {
1083            ThreadState::Blocked { reason: actual_reason, callback, .. } => {
1084                assert_eq!(
1085                    reason, actual_reason,
1086                    "unblock_thread: thread was blocked for the wrong reason"
1087                );
1088                callback
1089            }
1090            _ => panic!("unblock_thread: thread was not blocked"),
1091        };
1092        // The callback must be executed in the previously blocked thread.
1093        let old_thread = this.machine.threads.set_active_thread_id(thread);
1094        callback.call(this, UnblockKind::Ready)?;
1095        this.machine.threads.set_active_thread_id(old_thread);
1096        interp_ok(())
1097    }
1098
1099    #[inline]
1100    fn detach_thread(
1101        &mut self,
1102        thread_id: ThreadId,
1103        allow_terminated_joined: bool,
1104    ) -> InterpResult<'tcx> {
1105        let this = self.eval_context_mut();
1106        this.machine.threads.detach_thread(thread_id, allow_terminated_joined)
1107    }
1108
1109    #[inline]
1110    fn join_thread(&mut self, joined_thread_id: ThreadId) -> InterpResult<'tcx> {
1111        let this = self.eval_context_mut();
1112        this.machine.threads.join_thread(joined_thread_id, &mut this.machine.data_race)?;
1113        interp_ok(())
1114    }
1115
1116    #[inline]
1117    fn join_thread_exclusive(&mut self, joined_thread_id: ThreadId) -> InterpResult<'tcx> {
1118        let this = self.eval_context_mut();
1119        this.machine
1120            .threads
1121            .join_thread_exclusive(joined_thread_id, &mut this.machine.data_race)?;
1122        interp_ok(())
1123    }
1124
1125    #[inline]
1126    fn active_thread(&self) -> ThreadId {
1127        let this = self.eval_context_ref();
1128        this.machine.threads.active_thread()
1129    }
1130
1131    #[inline]
1132    fn active_thread_mut(&mut self) -> &mut Thread<'tcx> {
1133        let this = self.eval_context_mut();
1134        this.machine.threads.active_thread_mut()
1135    }
1136
1137    #[inline]
1138    fn active_thread_ref(&self) -> &Thread<'tcx> {
1139        let this = self.eval_context_ref();
1140        this.machine.threads.active_thread_ref()
1141    }
1142
1143    #[inline]
1144    fn get_total_thread_count(&self) -> usize {
1145        let this = self.eval_context_ref();
1146        this.machine.threads.get_total_thread_count()
1147    }
1148
1149    #[inline]
1150    fn have_all_terminated(&self) -> bool {
1151        let this = self.eval_context_ref();
1152        this.machine.threads.have_all_terminated()
1153    }
1154
1155    #[inline]
1156    fn enable_thread(&mut self, thread_id: ThreadId) {
1157        let this = self.eval_context_mut();
1158        this.machine.threads.enable_thread(thread_id);
1159    }
1160
1161    #[inline]
1162    fn active_thread_stack<'a>(&'a self) -> &'a [Frame<'tcx, Provenance, FrameExtra<'tcx>>] {
1163        let this = self.eval_context_ref();
1164        this.machine.threads.active_thread_stack()
1165    }
1166
1167    #[inline]
1168    fn active_thread_stack_mut<'a>(
1169        &'a mut self,
1170    ) -> &'a mut Vec<Frame<'tcx, Provenance, FrameExtra<'tcx>>> {
1171        let this = self.eval_context_mut();
1172        this.machine.threads.active_thread_stack_mut()
1173    }
1174
1175    /// Set the name of the current thread. The buffer must not include the null terminator.
1176    #[inline]
1177    fn set_thread_name(&mut self, thread: ThreadId, new_thread_name: Vec<u8>) {
1178        self.eval_context_mut().machine.threads.set_thread_name(thread, new_thread_name);
1179    }
1180
1181    #[inline]
1182    fn get_thread_name<'c>(&'c self, thread: ThreadId) -> Option<&'c [u8]>
1183    where
1184        'tcx: 'c,
1185    {
1186        self.eval_context_ref().machine.threads.get_thread_name(thread)
1187    }
1188
1189    #[inline]
1190    fn yield_active_thread(&mut self) {
1191        self.eval_context_mut().machine.threads.yield_active_thread();
1192    }
1193
1194    #[inline]
1195    fn maybe_preempt_active_thread(&mut self) {
1196        use rand::Rng as _;
1197
1198        let this = self.eval_context_mut();
1199        if !this.machine.threads.fixed_scheduling
1200            && this.machine.rng.get_mut().random_bool(this.machine.preemption_rate)
1201        {
1202            this.yield_active_thread();
1203        }
1204    }
1205
1206    /// Run the core interpreter loop. Returns only when an interrupt occurs (an error or program
1207    /// termination).
1208    fn run_threads(&mut self) -> InterpResult<'tcx, !> {
1209        let this = self.eval_context_mut();
1210        loop {
1211            if CTRL_C_RECEIVED.load(Relaxed) {
1212                this.machine.handle_abnormal_termination();
1213                throw_machine_stop!(TerminationInfo::Interrupted);
1214            }
1215            match this.schedule()? {
1216                SchedulingAction::ExecuteStep => {
1217                    if !this.step()? {
1218                        // See if this thread can do something else.
1219                        match this.run_on_stack_empty()? {
1220                            Poll::Pending => {} // keep going
1221                            Poll::Ready(()) =>
1222                                this.terminate_active_thread(TlsAllocAction::Deallocate)?,
1223                        }
1224                    }
1225                }
1226                SchedulingAction::ExecuteTimeoutCallback => {
1227                    this.run_timeout_callback()?;
1228                }
1229                SchedulingAction::Sleep(duration) => {
1230                    this.machine.monotonic_clock.sleep(duration);
1231                }
1232            }
1233        }
1234    }
1235}