src/threads.rs

   1 //! Implements threads.
   2
   3 use std::cell::RefCell;
   4 use std::collections::hash_map::Entry;
   5
   6 use log::trace;
   7
   8 use rustc_data_structures::fx::FxHashMap;
   9 use rustc_index::vec::{Idx, IndexVec};
  10 use rustc_middle::mir;
  11 use rustc_middle::ty;
  12
  13 use crate::*;
  14
  15 /// A thread identifier.
  16 #[derive(Clone, Copy, Debug, PartialOrd, Ord, PartialEq, Eq, Hash)]
  17 pub struct ThreadId(usize);
  18
  19 impl Idx for ThreadId {
  20     fn new(idx: usize) -> Self {
  21         ThreadId(idx)
  22     }
  23     fn index(self) -> usize {
  24         self.0
  25     }
  26 }
  27
  28 impl From<u64> for ThreadId {
  29     fn from(id: u64) -> Self {
  30         Self(id as usize)
  31     }
  32 }
  33
  34 /// The state of a thread.
  35 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
  36 pub enum ThreadState {
  37     /// The thread is enabled and can be executed.
  38     Enabled,
  39     /// The thread tried to join the specified thread and is blocked until that
  40     /// thread terminates.
  41     Blocked(ThreadId),
  42     /// The thread has terminated its execution (we do not delete terminated
  43     /// threads.)
  44     Terminated,
  45 }
  46
  47 /// A thread.
  48 pub struct Thread<'mir, 'tcx> {
  49     state: ThreadState,
  50     /// The virtual call stack.
  51     stack: Vec<Frame<'mir, 'tcx, Tag, FrameData<'tcx>>>,
  52     /// Is the thread detached?
  53     ///
  54     /// A thread is detached if its join handle was destroyed and no other
  55     /// thread can join it.
  56     detached: bool,
  57 }
  58
  59 impl<'mir, 'tcx> Thread<'mir, 'tcx> {
  60     /// Check if the thread terminated. If yes, change the state to terminated
  61     /// and return `true`.
  62     fn check_terminated(&mut self) -> bool {
  63         if self.state == ThreadState::Enabled {
  64             if self.stack.is_empty() {
  65                 self.state = ThreadState::Terminated;
  66                 return true;
  67             }
  68         }
  69         false
  70     }
  71 }
  72
  73 impl<'mir, 'tcx> std::fmt::Debug for Thread<'mir, 'tcx> {
  74     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
  75         write!(f, "{:?}", self.state)
  76     }
  77 }
  78
  79 impl<'mir, 'tcx> Default for Thread<'mir, 'tcx> {
  80     fn default() -> Self {
  81         Self { state: ThreadState::Enabled, stack: Vec::new(), detached: false }
  82     }
  83 }
  84
  85 /// A set of threads.
  86 #[derive(Debug)]
  87 pub struct ThreadSet<'mir, 'tcx> {
  88     /// Identifier of the currently active thread.
  89     active_thread: ThreadId,
  90     /// Threads used in the program.
  91     ///
  92     /// Note that this vector also contains terminated threads.
  93     threads: IndexVec<ThreadId, Thread<'mir, 'tcx>>,
  94 }
  95
  96 impl<'mir, 'tcx> Default for ThreadSet<'mir, 'tcx> {
  97     fn default() -> Self {
  98         let mut threads = IndexVec::new();
  99         threads.push(Default::default());
 100         Self {
 101             active_thread: ThreadId::new(0),
 102             threads: threads,
 103         }
 104     }
 105 }
 106
 107 impl<'mir, 'tcx: 'mir> ThreadSet<'mir, 'tcx> {
 108     /// Borrow the stack of the active thread.
 109     fn active_thread_stack(&self) -> &[Frame<'mir, 'tcx, Tag, FrameData<'tcx>>] {
 110         &self.threads[self.active_thread].stack
 111     }
 112     /// Mutably borrow the stack of the active thread.
 113     fn active_thread_stack_mut(&mut self) -> &mut Vec<Frame<'mir, 'tcx, Tag, FrameData<'tcx>>> {
 114         &mut self.threads[self.active_thread].stack
 115     }
 116     /// Create a new thread and returns its id.
 117     fn create_thread(&mut self) -> ThreadId {
 118         let new_thread_id = ThreadId::new(self.threads.len());
 119         self.threads.push(Default::default());
 120         new_thread_id
 121     }
 122     /// Set an active thread and return the id of the thread that was active before.
 123     fn set_active_thread(&mut self, id: ThreadId) -> ThreadId {
 124         let active_thread_id = self.active_thread;
 125         self.active_thread = id;
 126         assert!(self.active_thread.index() < self.threads.len());
 127         active_thread_id
 128     }
 129     /// Get the id of the currently active thread.
 130     fn get_active_thread(&self) -> ThreadId {
 131         self.active_thread
 132     }
 133     /// Mark the thread as detached, which means that no other thread will try
 134     /// to join it and the thread is responsible for cleaning up.
 135     fn detach_thread(&mut self, id: ThreadId) {
 136         self.threads[id].detached = true;
 137     }
 138     /// Mark that the active thread tries to join the thread with `joined_thread_id`.
 139     fn join_thread(&mut self, joined_thread_id: ThreadId) {
 140         assert!(!self.threads[joined_thread_id].detached, "Bug: trying to join a detached thread.");
 141         assert_ne!(joined_thread_id, self.active_thread, "Bug: trying to join itself");
 142         assert!(
 143             self.threads
 144                 .iter()
 145                 .all(|thread| thread.state != ThreadState::Blocked(joined_thread_id)),
 146             "Bug: multiple threads try to join the same thread."
 147         );
 148         if self.threads[joined_thread_id].state != ThreadState::Terminated {
 149             // The joined thread is still running, we need to wait for it.
 150             self.threads[self.active_thread].state = ThreadState::Blocked(joined_thread_id);
 151             trace!(
 152                 "{:?} blocked on {:?} when trying to join",
 153                 self.active_thread,
 154                 joined_thread_id
 155             );
 156         }
 157     }
 158     /// Get ids of all threads ever allocated.
 159     fn get_all_thread_ids(&mut self) -> Vec<ThreadId> {
 160         (0..self.threads.len()).map(ThreadId::new).collect()
 161     }
 162     /// Decide which thread to run next.
 163     ///
 164     /// Returns `false` if all threads terminated.
 165     fn schedule(&mut self) -> InterpResult<'tcx, bool> {
 166         if self.threads[self.active_thread].check_terminated() {
 167             // Check if we need to unblock any threads.
 168             for (i, thread) in self.threads.iter_enumerated_mut() {
 169                 if thread.state == ThreadState::Blocked(self.active_thread) {
 170                     trace!("unblocking {:?} because {:?} terminated", i, self.active_thread);
 171                     thread.state = ThreadState::Enabled;
 172                 }
 173             }
 174         }
 175         if self.threads[self.active_thread].state == ThreadState::Enabled {
 176             return Ok(true);
 177         }
 178         if let Some(enabled_thread) =
 179             self.threads.iter().position(|thread| thread.state == ThreadState::Enabled)
 180         {
 181             self.active_thread = ThreadId::new(enabled_thread);
 182             return Ok(true);
 183         }
 184         if self.threads.iter().all(|thread| thread.state == ThreadState::Terminated) {
 185             Ok(false)
 186         } else {
 187             throw_machine_stop!(TerminationInfo::Abort(Some(format!("execution deadlocked"))))
 188         }
 189     }
 190 }
 191
 192 /// In Rust, a thread local variable is just a specially marked static. To
 193 /// ensure a property that each memory allocation has a globally unique
 194 /// allocation identifier, we create a fresh allocation id for each thread. This
 195 /// data structure keeps the track of the created allocation identifiers and
 196 /// their relation to the original static allocations.
 197 #[derive(Clone, Debug, Default)]
 198 pub struct ThreadLocalStorage {
 199     /// A map from a thread local allocation identifier to the static from which
 200     /// it was created.
 201     thread_local_origin: RefCell<FxHashMap<AllocId, AllocId>>,
 202     /// A map from a thread local static and thread id to the unique thread
 203     /// local allocation.
 204     thread_local_allocations: RefCell<FxHashMap<(AllocId, ThreadId), AllocId>>,
 205     /// The currently active thread.
 206     active_thread: Option<ThreadId>,
 207 }
 208
 209 impl ThreadLocalStorage {
 210     /// For static allocation identifier `original_id` get a thread local
 211     /// allocation identifier. If it is not allocated yet, allocate.
 212     pub fn get_or_register_allocation(&self, tcx: ty::TyCtxt<'_>, original_id: AllocId) -> AllocId {
 213         match self
 214             .thread_local_allocations
 215             .borrow_mut()
 216             .entry((original_id, self.active_thread.unwrap()))
 217         {
 218             Entry::Occupied(entry) => *entry.get(),
 219             Entry::Vacant(entry) => {
 220                 let fresh_id = tcx.alloc_map.lock().reserve();
 221                 entry.insert(fresh_id);
 222                 self.thread_local_origin.borrow_mut().insert(fresh_id, original_id);
 223                 trace!(
 224                     "get_or_register_allocation(original_id={:?}) -> {:?}",
 225                     original_id,
 226                     fresh_id
 227                 );
 228                 fresh_id
 229             }
 230         }
 231     }
 232     /// For thread local allocation identifier `alloc_id`, retrieve the original
 233     /// static allocation identifier from which it was created.
 234     pub fn resolve_allocation<'tcx>(
 235         &self,
 236         tcx: ty::TyCtxt<'tcx>,
 237         alloc_id: AllocId,
 238     ) -> Option<mir::interpret::GlobalAlloc<'tcx>> {
 239         trace!("resolve_allocation(alloc_id: {:?})", alloc_id);
 240         if let Some(original_id) = self.thread_local_origin.borrow().get(&alloc_id) {
 241             trace!("resolve_allocation(alloc_id: {:?}) -> {:?}", alloc_id, original_id);
 242             tcx.alloc_map.lock().get(*original_id)
 243         } else {
 244             tcx.alloc_map.lock().get(alloc_id)
 245         }
 246     }
 247     /// Set which thread is currently active.
 248     fn set_active_thread(&mut self, active_thread: ThreadId) {
 249         self.active_thread = Some(active_thread);
 250     }
 251 }
 252
 253 impl<'mir, 'tcx: 'mir> EvalContextExt<'mir, 'tcx> for crate::MiriEvalContext<'mir, 'tcx> {}
 254 pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx> {
 255     fn create_thread(&mut self) -> InterpResult<'tcx, ThreadId> {
 256         let this = self.eval_context_mut();
 257         Ok(this.machine.threads.create_thread())
 258     }
 259     fn detach_thread(&mut self, thread_id: ThreadId) -> InterpResult<'tcx> {
 260         let this = self.eval_context_mut();
 261         this.machine.threads.detach_thread(thread_id);
 262         Ok(())
 263     }
 264     fn join_thread(&mut self, joined_thread_id: ThreadId) -> InterpResult<'tcx> {
 265         let this = self.eval_context_mut();
 266         this.machine.threads.join_thread(joined_thread_id);
 267         Ok(())
 268     }
 269     fn set_active_thread(&mut self, thread_id: ThreadId) -> InterpResult<'tcx, ThreadId> {
 270         let this = self.eval_context_mut();
 271         this.memory.extra.tls.set_active_thread(thread_id);
 272         Ok(this.machine.threads.set_active_thread(thread_id))
 273     }
 274     fn get_active_thread(&self) -> InterpResult<'tcx, ThreadId> {
 275         let this = self.eval_context_ref();
 276         Ok(this.machine.threads.get_active_thread())
 277     }
 278     fn active_thread_stack(&self) -> &[Frame<'mir, 'tcx, Tag, FrameData<'tcx>>] {
 279         let this = self.eval_context_ref();
 280         this.machine.threads.active_thread_stack()
 281     }
 282     fn active_thread_stack_mut(&mut self) -> &mut Vec<Frame<'mir, 'tcx, Tag, FrameData<'tcx>>> {
 283         let this = self.eval_context_mut();
 284         this.machine.threads.active_thread_stack_mut()
 285     }
 286     fn get_all_thread_ids(&mut self) -> Vec<ThreadId> {
 287         let this = self.eval_context_mut();
 288         this.machine.threads.get_all_thread_ids()
 289     }
 290     /// Decide which thread to run next.
 291     ///
 292     /// Returns `false` if all threads terminated.
 293     fn schedule(&mut self) -> InterpResult<'tcx, bool> {
 294         let this = self.eval_context_mut();
 295         // Find the next thread to run.
 296         if this.machine.threads.schedule()? {
 297             let active_thread = this.machine.threads.get_active_thread();
 298             this.memory.extra.tls.set_active_thread(active_thread);
 299             Ok(true)
 300         } else {
 301             Ok(false)
 302         }
 303     }
 304 }