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     /// List of threads that just terminated. TODO: Cleanup.
  96     terminated_threads: Vec<ThreadId>,
  97 }
  98
  99 impl<'mir, 'tcx> Default for ThreadSet<'mir, 'tcx> {
 100     fn default() -> Self {
 101         let mut threads = IndexVec::new();
 102         threads.push(Default::default());
 103         Self {
 104             active_thread: ThreadId::new(0),
 105             threads: threads,
 106             terminated_threads: Default::default(),
 107         }
 108     }
 109 }
 110
 111 impl<'mir, 'tcx: 'mir> ThreadSet<'mir, 'tcx> {
 112     /// Borrow the stack of the active thread.
 113     fn active_thread_stack(&self) -> &[Frame<'mir, 'tcx, Tag, FrameData<'tcx>>] {
 114         &self.threads[self.active_thread].stack
 115     }
 116     /// Mutably borrow the stack of the active thread.
 117     fn active_thread_stack_mut(&mut self) -> &mut Vec<Frame<'mir, 'tcx, Tag, FrameData<'tcx>>> {
 118         &mut self.threads[self.active_thread].stack
 119     }
 120     /// Create a new thread and returns its id.
 121     fn create_thread(&mut self) -> ThreadId {
 122         let new_thread_id = ThreadId::new(self.threads.len());
 123         self.threads.push(Default::default());
 124         new_thread_id
 125     }
 126     /// Set an active thread and return the id of the thread that was active before.
 127     fn set_active_thread(&mut self, id: ThreadId) -> ThreadId {
 128         let active_thread_id = self.active_thread;
 129         self.active_thread = id;
 130         assert!(self.active_thread.index() < self.threads.len());
 131         active_thread_id
 132     }
 133     /// Get the id of the currently active thread.
 134     fn get_active_thread(&self) -> ThreadId {
 135         self.active_thread
 136     }
 137     /// Mark the thread as detached, which means that no other thread will try
 138     /// to join it and the thread is responsible for cleaning up.
 139     fn detach_thread(&mut self, id: ThreadId) {
 140         self.threads[id].detached = true;
 141     }
 142     /// Mark that the active thread tries to join the thread with `joined_thread_id`.
 143     fn join_thread(&mut self, joined_thread_id: ThreadId) {
 144         assert!(!self.threads[joined_thread_id].detached, "Bug: trying to join a detached thread.");
 145         assert_ne!(joined_thread_id, self.active_thread, "Bug: trying to join itself");
 146         assert!(
 147             self.threads
 148                 .iter()
 149                 .all(|thread| thread.state != ThreadState::Blocked(joined_thread_id)),
 150             "Bug: multiple threads try to join the same thread."
 151         );
 152         if self.threads[joined_thread_id].state != ThreadState::Terminated {
 153             // The joined thread is still running, we need to wait for it.
 154             self.threads[self.active_thread].state = ThreadState::Blocked(joined_thread_id);
 155             trace!(
 156                 "{:?} blocked on {:?} when trying to join",
 157                 self.active_thread,
 158                 joined_thread_id
 159             );
 160         }
 161     }
 162     /// Get ids of all threads ever allocated.
 163     fn get_all_thread_ids(&mut self) -> Vec<ThreadId> {
 164         (0..self.threads.len()).map(ThreadId::new).collect()
 165     }
 166     /// Decide which thread to run next.
 167     ///
 168     /// Returns `false` if all threads terminated.
 169     fn schedule(&mut self) -> InterpResult<'tcx, bool> {
 170         if self.threads[self.active_thread].check_terminated() {
 171             // Check if we need to unblock any threads.
 172             for (i, thread) in self.threads.iter_enumerated_mut() {
 173                 if thread.state == ThreadState::Blocked(self.active_thread) {
 174                     trace!("unblocking {:?} because {:?} terminated", i, self.active_thread);
 175                     thread.state = ThreadState::Enabled;
 176                 }
 177             }
 178         }
 179         if self.threads[self.active_thread].state == ThreadState::Enabled {
 180             return Ok(true);
 181         }
 182         if let Some(enabled_thread) =
 183             self.threads.iter().position(|thread| thread.state == ThreadState::Enabled)
 184         {
 185             self.active_thread = ThreadId::new(enabled_thread);
 186             return Ok(true);
 187         }
 188         if self.threads.iter().all(|thread| thread.state == ThreadState::Terminated) {
 189             Ok(false)
 190         } else {
 191             throw_machine_stop!(TerminationInfo::Abort(Some(format!("execution deadlocked"))))
 192         }
 193     }
 194 }
 195
 196 /// In Rust, a thread local variable is just a specially marked static. To
 197 /// ensure a property that each memory allocation has a globally unique
 198 /// allocation identifier, we create a fresh allocation id for each thread. This
 199 /// data structure keeps the track of the created allocation identifiers and
 200 /// their relation to the original static allocations.
 201 #[derive(Clone, Debug, Default)]
 202 pub struct ThreadLocalStorage {
 203     /// A map from a thread local allocation identifier to the static from which
 204     /// it was created.
 205     thread_local_origin: RefCell<FxHashMap<AllocId, AllocId>>,
 206     /// A map from a thread local static and thread id to the unique thread
 207     /// local allocation.
 208     thread_local_allocations: RefCell<FxHashMap<(AllocId, ThreadId), AllocId>>,
 209     /// The currently active thread.
 210     active_thread: Option<ThreadId>,
 211 }
 212
 213 impl ThreadLocalStorage {
 214     /// For static allocation identifier `original_id` get a thread local
 215     /// allocation identifier. If it is not allocated yet, allocate.
 216     pub fn get_or_register_allocation(&self, tcx: ty::TyCtxt<'_>, original_id: AllocId) -> AllocId {
 217         match self
 218             .thread_local_allocations
 219             .borrow_mut()
 220             .entry((original_id, self.active_thread.unwrap()))
 221         {
 222             Entry::Occupied(entry) => *entry.get(),
 223             Entry::Vacant(entry) => {
 224                 let fresh_id = tcx.alloc_map.lock().reserve();
 225                 entry.insert(fresh_id);
 226                 self.thread_local_origin.borrow_mut().insert(fresh_id, original_id);
 227                 trace!(
 228                     "get_or_register_allocation(original_id={:?}) -> {:?}",
 229                     original_id,
 230                     fresh_id
 231                 );
 232                 fresh_id
 233             }
 234         }
 235     }
 236     /// For thread local allocation identifier `alloc_id`, retrieve the original
 237     /// static allocation identifier from which it was created.
 238     pub fn resolve_allocation<'tcx>(
 239         &self,
 240         tcx: ty::TyCtxt<'tcx>,
 241         alloc_id: AllocId,
 242     ) -> Option<mir::interpret::GlobalAlloc<'tcx>> {
 243         trace!("resolve_allocation(alloc_id: {:?})", alloc_id);
 244         if let Some(original_id) = self.thread_local_origin.borrow().get(&alloc_id) {
 245             trace!("resolve_allocation(alloc_id: {:?}) -> {:?}", alloc_id, original_id);
 246             tcx.alloc_map.lock().get(*original_id)
 247         } else {
 248             tcx.alloc_map.lock().get(alloc_id)
 249         }
 250     }
 251     /// Set which thread is currently active.
 252     fn set_active_thread(&mut self, active_thread: ThreadId) {
 253         self.active_thread = Some(active_thread);
 254     }
 255 }
 256
 257 impl<'mir, 'tcx: 'mir> EvalContextExt<'mir, 'tcx> for crate::MiriEvalContext<'mir, 'tcx> {}
 258 pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx> {
 259     fn create_thread(&mut self) -> InterpResult<'tcx, ThreadId> {
 260         let this = self.eval_context_mut();
 261         Ok(this.machine.threads.create_thread())
 262     }
 263     fn detach_thread(&mut self, thread_id: ThreadId) -> InterpResult<'tcx> {
 264         let this = self.eval_context_mut();
 265         this.machine.threads.detach_thread(thread_id);
 266         Ok(())
 267     }
 268     fn join_thread(&mut self, joined_thread_id: ThreadId) -> InterpResult<'tcx> {
 269         let this = self.eval_context_mut();
 270         this.machine.threads.join_thread(joined_thread_id);
 271         Ok(())
 272     }
 273     fn set_active_thread(&mut self, thread_id: ThreadId) -> InterpResult<'tcx, ThreadId> {
 274         let this = self.eval_context_mut();
 275         this.memory.extra.tls.set_active_thread(thread_id);
 276         Ok(this.machine.threads.set_active_thread(thread_id))
 277     }
 278     fn get_active_thread(&self) -> InterpResult<'tcx, ThreadId> {
 279         let this = self.eval_context_ref();
 280         Ok(this.machine.threads.get_active_thread())
 281     }
 282     fn active_thread_stack(&self) -> &[Frame<'mir, 'tcx, Tag, FrameData<'tcx>>] {
 283         let this = self.eval_context_ref();
 284         this.machine.threads.active_thread_stack()
 285     }
 286     fn active_thread_stack_mut(&mut self) -> &mut Vec<Frame<'mir, 'tcx, Tag, FrameData<'tcx>>> {
 287         let this = self.eval_context_mut();
 288         this.machine.threads.active_thread_stack_mut()
 289     }
 290     fn get_all_thread_ids(&mut self) -> Vec<ThreadId> {
 291         let this = self.eval_context_mut();
 292         this.machine.threads.get_all_thread_ids()
 293     }
 294     /// Decide which thread to run next.
 295     ///
 296     /// Returns `false` if all threads terminated.
 297     fn schedule(&mut self) -> InterpResult<'tcx, bool> {
 298         let this = self.eval_context_mut();
 299         // Find the next thread to run.
 300         if this.machine.threads.schedule()? {
 301             let active_thread = this.machine.threads.get_active_thread();
 302             this.memory.extra.tls.set_active_thread(active_thread);
 303             Ok(true)
 304         } else {
 305             Ok(false)
 306         }
 307     }
 308 }