src/sync.rs

   1 use std::collections::{hash_map::Entry, HashMap, VecDeque};
   2 use std::convert::TryFrom;
   3 use std::num::NonZeroU32;
   4 use std::ops::Not;
   5
   6 use log::trace;
   7
   8 use rustc_index::vec::{Idx, IndexVec};
   9
  10 use crate::*;
  11
  12 /// We cannot use the `newtype_index!` macro because we have to use 0 as a
  13 /// sentinel value meaning that the identifier is not assigned. This is because
  14 /// the pthreads static initializers initialize memory with zeros (see the
  15 /// `src/shims/sync.rs` file).
  16 macro_rules! declare_id {
  17     ($name: ident) => {
  18         /// 0 is used to indicate that the id was not yet assigned and,
  19         /// therefore, is not a valid identifier.
  20         #[derive(Clone, Copy, Debug, PartialOrd, Ord, PartialEq, Eq, Hash)]
  21         pub struct $name(NonZeroU32);
  22
  23         impl $name {
  24             // Panics if `id == 0`.
  25             pub fn from_u32(id: u32) -> Self {
  26                 Self(NonZeroU32::new(id).unwrap())
  27             }
  28         }
  29
  30         impl Idx for $name {
  31             fn new(idx: usize) -> Self {
  32                 // We use 0 as a sentinel value (see the comment above) and,
  33                 // therefore, need to shift by one when converting from an index
  34                 // into a vector.
  35                 let shifted_idx = u32::try_from(idx).unwrap().checked_add(1).unwrap();
  36                 $name(NonZeroU32::new(shifted_idx).unwrap())
  37             }
  38             fn index(self) -> usize {
  39                 // See the comment in `Self::new`.
  40                 // (This cannot underflow because self is NonZeroU32.)
  41                 usize::try_from(self.0.get() - 1).unwrap()
  42             }
  43         }
  44
  45         impl $name {
  46             pub fn to_u32_scalar<'tcx>(&self) -> Scalar<Tag> {
  47                 Scalar::from_u32(self.0.get())
  48             }
  49         }
  50     };
  51 }
  52
  53 declare_id!(MutexId);
  54
  55 /// The mutex state.
  56 #[derive(Default, Debug)]
  57 struct Mutex {
  58     /// The thread that currently owns the lock.
  59     owner: Option<ThreadId>,
  60     /// How many times the mutex was locked by the owner.
  61     lock_count: usize,
  62     /// The queue of threads waiting for this mutex.
  63     queue: VecDeque<ThreadId>,
  64     /// Data race handle, this tracks the happens-before
  65     /// relationship between each mutex access. It is
  66     /// released to during unlock and acquired from during
  67     /// locking, and therefore stores the clock of the last
  68     /// thread to release this mutex.
  69     data_race: VClock
  70 }
  71
  72 declare_id!(RwLockId);
  73
  74 /// The read-write lock state.
  75 #[derive(Default, Debug)]
  76 struct RwLock {
  77     /// The writer thread that currently owns the lock.
  78     writer: Option<ThreadId>,
  79     /// The readers that currently own the lock and how many times they acquired
  80     /// the lock.
  81     readers: HashMap<ThreadId, usize>,
  82     /// The queue of writer threads waiting for this lock.
  83     writer_queue: VecDeque<ThreadId>,
  84     /// The queue of reader threads waiting for this lock.
  85     reader_queue: VecDeque<ThreadId>,
  86     /// Data race handle for writers, tracks the happens-before
  87     /// ordering between each write access to a rwlock and is updated
  88     /// after a sequence of concurrent readers to track the happens-
  89     /// before ordering between the set of previous readers and
  90     /// the current writer.
  91     /// Contains the clock of the last thread to release a writer
  92     /// lock or the joined clock of the set of last threads to release
  93     /// shared reader locks.
  94     data_race: VClock,
  95     /// Data race handle for readers, this is temporary storage
  96     /// for the combined happens-before ordering for between all
  97     /// concurrent readers and the next writer, and the value
  98     /// is stored to the main data_race variable once all
  99     /// readers are finished.
 100     /// Has to be stored separately since reader lock acquires
 101     /// must load the clock of the last write and must not
 102     /// add happens-before orderings between shared reader
 103     /// locks.
 104     data_race_reader: VClock,
 105 }
 106
 107 declare_id!(CondvarId);
 108
 109 /// A thread waiting on a conditional variable.
 110 #[derive(Debug)]
 111 struct CondvarWaiter {
 112     /// The thread that is waiting on this variable.
 113     thread: ThreadId,
 114     /// The mutex on which the thread is waiting.
 115     mutex: MutexId,
 116 }
 117
 118 /// The conditional variable state.
 119 #[derive(Default, Debug)]
 120 struct Condvar {
 121     waiters: VecDeque<CondvarWaiter>,
 122     /// Tracks the happens-before relationship
 123     /// between a cond-var signal and a cond-var
 124     /// wait during a non-suprious signal event.
 125     /// Contains the clock of the last thread to
 126     /// perform a futex-signal.
 127     data_race: VClock,
 128 }
 129
 130 /// The futex state.
 131 #[derive(Default, Debug)]
 132 struct Futex {
 133     waiters: VecDeque<FutexWaiter>,
 134     /// Tracks the happens-before relationship
 135     /// between a futex-wake and a futex-wait
 136     /// during a non-spurious wake event.
 137     /// Contains the clock of the last thread to
 138     /// perform a futex-wake.
 139     data_race: VClock,
 140 }
 141
 142 /// A thread waiting on a futex.
 143 #[derive(Debug)]
 144 struct FutexWaiter {
 145     /// The thread that is waiting on this futex.
 146     thread: ThreadId,
 147 }
 148
 149 /// The state of all synchronization variables.
 150 #[derive(Default, Debug)]
 151 pub(super) struct SynchronizationState {
 152     mutexes: IndexVec<MutexId, Mutex>,
 153     rwlocks: IndexVec<RwLockId, RwLock>,
 154     condvars: IndexVec<CondvarId, Condvar>,
 155     futexes: HashMap<Pointer, Futex>,
 156 }
 157
 158 // Private extension trait for local helper methods
 159 impl<'mir, 'tcx: 'mir> EvalContextExtPriv<'mir, 'tcx> for crate::MiriEvalContext<'mir, 'tcx> {}
 160 trait EvalContextExtPriv<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx> {
 161     /// Take a reader out of the queue waiting for the lock.
 162     /// Returns `true` if some thread got the rwlock.
 163     #[inline]
 164     fn rwlock_dequeue_and_lock_reader(&mut self, id: RwLockId) -> bool {
 165         let this = self.eval_context_mut();
 166         if let Some(reader) = this.machine.threads.sync.rwlocks[id].reader_queue.pop_front() {
 167             this.unblock_thread(reader);
 168             this.rwlock_reader_lock(id, reader);
 169             true
 170         } else {
 171             false
 172         }
 173     }
 174
 175     /// Take the writer out of the queue waiting for the lock.
 176     /// Returns `true` if some thread got the rwlock.
 177     #[inline]
 178     fn rwlock_dequeue_and_lock_writer(&mut self, id: RwLockId) -> bool {
 179         let this = self.eval_context_mut();
 180         if let Some(writer) = this.machine.threads.sync.rwlocks[id].writer_queue.pop_front() {
 181             this.unblock_thread(writer);
 182             this.rwlock_writer_lock(id, writer);
 183             true
 184         } else {
 185             false
 186         }
 187     }
 188
 189     /// Take a thread out of the queue waiting for the mutex, and lock
 190     /// the mutex for it. Returns `true` if some thread has the mutex now.
 191     #[inline]
 192     fn mutex_dequeue_and_lock(&mut self, id: MutexId) -> bool {
 193         let this = self.eval_context_mut();
 194         if let Some(thread) = this.machine.threads.sync.mutexes[id].queue.pop_front() {
 195             this.unblock_thread(thread);
 196             this.mutex_lock(id, thread);
 197             true
 198         } else {
 199             false
 200         }
 201     }
 202 }
 203
 204 // Public interface to synchronization primitives. Please note that in most
 205 // cases, the function calls are infallible and it is the client's (shim
 206 // implementation's) responsibility to detect and deal with erroneous
 207 // situations.
 208 impl<'mir, 'tcx: 'mir> EvalContextExt<'mir, 'tcx> for crate::MiriEvalContext<'mir, 'tcx> {}
 209 pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx> {
 210     #[inline]
 211     /// Create state for a new mutex.
 212     fn mutex_create(&mut self) -> MutexId {
 213         let this = self.eval_context_mut();
 214         this.machine.threads.sync.mutexes.push(Default::default())
 215     }
 216
 217     #[inline]
 218     /// Get the id of the thread that currently owns this lock.
 219     fn mutex_get_owner(&mut self, id: MutexId) -> ThreadId {
 220         let this = self.eval_context_ref();
 221         this.machine.threads.sync.mutexes[id].owner.unwrap()
 222     }
 223
 224     #[inline]
 225     /// Check if locked.
 226     fn mutex_is_locked(&self, id: MutexId) -> bool {
 227         let this = self.eval_context_ref();
 228         this.machine.threads.sync.mutexes[id].owner.is_some()
 229     }
 230
 231     /// Lock by setting the mutex owner and increasing the lock count.
 232     fn mutex_lock(&mut self, id: MutexId, thread: ThreadId) {
 233         let this = self.eval_context_mut();
 234         let mutex = &mut this.machine.threads.sync.mutexes[id];
 235         if let Some(current_owner) = mutex.owner {
 236             assert_eq!(thread, current_owner, "mutex already locked by another thread");
 237             assert!(
 238                 mutex.lock_count > 0,
 239                 "invariant violation: lock_count == 0 iff the thread is unlocked"
 240             );
 241         } else {
 242             mutex.owner = Some(thread);
 243         }
 244         mutex.lock_count = mutex.lock_count.checked_add(1).unwrap();
 245         if let Some(data_race) = &this.memory.extra.data_race {
 246             data_race.validate_lock_acquire(&mutex.data_race, thread);
 247         }
 248     }
 249
 250     /// Try unlocking by decreasing the lock count and returning the old lock
 251     /// count. If the lock count reaches 0, release the lock and potentially
 252     /// give to a new owner. If the lock was not locked by `expected_owner`,
 253     /// return `None`.
 254     fn mutex_unlock(
 255         &mut self,
 256         id: MutexId,
 257         expected_owner: ThreadId,
 258     ) -> Option<usize> {
 259         let this = self.eval_context_mut();
 260         let mutex = &mut this.machine.threads.sync.mutexes[id];
 261         if let Some(current_owner) = mutex.owner {
 262             // Mutex is locked.
 263             if current_owner != expected_owner {
 264                 // Only the owner can unlock the mutex.
 265                 return None;
 266             }
 267             let old_lock_count = mutex.lock_count;
 268             mutex.lock_count = old_lock_count
 269                 .checked_sub(1)
 270                 .expect("invariant violation: lock_count == 0 iff the thread is unlocked");
 271             if mutex.lock_count == 0 {
 272                 mutex.owner = None;
 273                 // The mutex is completely unlocked. Try transfering ownership
 274                 // to another thread.
 275                 if let Some(data_race) = &this.memory.extra.data_race {
 276                     data_race.validate_lock_release(&mut mutex.data_race, current_owner);
 277                 }
 278                 this.mutex_dequeue_and_lock(id);
 279             }
 280             Some(old_lock_count)
 281         } else {
 282             // Mutex is not locked.
 283             None
 284         }
 285     }
 286
 287     #[inline]
 288     /// Put the thread into the queue waiting for the mutex.
 289     fn mutex_enqueue_and_block(&mut self, id: MutexId, thread: ThreadId) {
 290         let this = self.eval_context_mut();
 291         assert!(this.mutex_is_locked(id), "queing on unlocked mutex");
 292         this.machine.threads.sync.mutexes[id].queue.push_back(thread);
 293         this.block_thread(thread);
 294     }
 295
 296     #[inline]
 297     /// Create state for a new read write lock.
 298     fn rwlock_create(&mut self) -> RwLockId {
 299         let this = self.eval_context_mut();
 300         this.machine.threads.sync.rwlocks.push(Default::default())
 301     }
 302
 303     #[inline]
 304     /// Check if locked.
 305     fn rwlock_is_locked(&self, id: RwLockId) -> bool {
 306         let this = self.eval_context_ref();
 307         let rwlock = &this.machine.threads.sync.rwlocks[id];
 308         trace!(
 309             "rwlock_is_locked: {:?} writer is {:?} and there are {} reader threads (some of which could hold multiple read locks)",
 310             id, rwlock.writer, rwlock.readers.len(),
 311         );
 312         rwlock.writer.is_some()|| rwlock.readers.is_empty().not()
 313     }
 314
 315     #[inline]
 316     /// Check if write locked.
 317     fn rwlock_is_write_locked(&self, id: RwLockId) -> bool {
 318         let this = self.eval_context_ref();
 319         let rwlock = &this.machine.threads.sync.rwlocks[id];
 320         trace!("rwlock_is_write_locked: {:?} writer is {:?}", id, rwlock.writer);
 321         rwlock.writer.is_some()
 322     }
 323
 324     /// Read-lock the lock by adding the `reader` the list of threads that own
 325     /// this lock.
 326     fn rwlock_reader_lock(&mut self, id: RwLockId, reader: ThreadId) {
 327         let this = self.eval_context_mut();
 328         assert!(!this.rwlock_is_write_locked(id), "the lock is write locked");
 329         trace!("rwlock_reader_lock: {:?} now also held (one more time) by {:?}", id, reader);
 330         let rwlock = &mut this.machine.threads.sync.rwlocks[id];
 331         let count = rwlock.readers.entry(reader).or_insert(0);
 332         *count = count.checked_add(1).expect("the reader counter overflowed");
 333         if let Some(data_race) = &this.memory.extra.data_race {
 334             data_race.validate_lock_acquire(&rwlock.data_race, reader);
 335         }
 336     }
 337
 338     /// Try read-unlock the lock for `reader` and potentially give the lock to a new owner.
 339     /// Returns `true` if succeeded, `false` if this `reader` did not hold the lock.
 340     fn rwlock_reader_unlock(&mut self, id: RwLockId, reader: ThreadId) -> bool {
 341         let this = self.eval_context_mut();
 342         let rwlock = &mut this.machine.threads.sync.rwlocks[id];
 343         match rwlock.readers.entry(reader) {
 344             Entry::Occupied(mut entry) => {
 345                 let count = entry.get_mut();
 346                 assert!(*count > 0, "rwlock locked with count == 0");
 347                 *count -= 1;
 348                 if *count == 0 {
 349                     trace!("rwlock_reader_unlock: {:?} no longer held by {:?}", id, reader);
 350                     entry.remove();
 351                 } else {
 352                     trace!("rwlock_reader_unlock: {:?} held one less time by {:?}", id, reader);
 353                 }
 354             }
 355             Entry::Vacant(_) => return false, // we did not even own this lock
 356         }
 357         if let Some(data_race) = &this.memory.extra.data_race {
 358             data_race.validate_lock_release_shared(&mut rwlock.data_race_reader, reader);
 359         }
 360
 361         // The thread was a reader. If the lock is not held any more, give it to a writer.
 362         if this.rwlock_is_locked(id).not() {
 363
 364             // All the readers are finished, so set the writer data-race handle to the value
 365             //  of the union of all reader data race handles, since the set of readers
 366             //  happen-before the writers
 367             let rwlock = &mut this.machine.threads.sync.rwlocks[id];
 368             rwlock.data_race.clone_from(&rwlock.data_race_reader);
 369             this.rwlock_dequeue_and_lock_writer(id);
 370         }
 371         true
 372     }
 373
 374     #[inline]
 375     /// Put the reader in the queue waiting for the lock and block it.
 376     fn rwlock_enqueue_and_block_reader(
 377         &mut self,
 378         id: RwLockId,
 379         reader: ThreadId,
 380     ) {
 381         let this = self.eval_context_mut();
 382         assert!(this.rwlock_is_write_locked(id), "read-queueing on not write locked rwlock");
 383         this.machine.threads.sync.rwlocks[id].reader_queue.push_back(reader);
 384         this.block_thread(reader);
 385     }
 386
 387     #[inline]
 388     /// Lock by setting the writer that owns the lock.
 389     fn rwlock_writer_lock(&mut self, id: RwLockId, writer: ThreadId) {
 390         let this = self.eval_context_mut();
 391         assert!(!this.rwlock_is_locked(id), "the rwlock is already locked");
 392         trace!("rwlock_writer_lock: {:?} now held by {:?}", id, writer);
 393         let rwlock = &mut this.machine.threads.sync.rwlocks[id];
 394         rwlock.writer = Some(writer);
 395         if let Some(data_race) = &this.memory.extra.data_race {
 396             data_race.validate_lock_acquire(&rwlock.data_race, writer);
 397         }
 398     }
 399
 400     #[inline]
 401     /// Try to unlock by removing the writer.
 402     fn rwlock_writer_unlock(&mut self, id: RwLockId, expected_writer: ThreadId) -> bool {
 403         let this = self.eval_context_mut();
 404         let rwlock = &mut this.machine.threads.sync.rwlocks[id];
 405         if let Some(current_writer) = rwlock.writer {
 406             if current_writer != expected_writer {
 407                 // Only the owner can unlock the rwlock.
 408                 return false;
 409             }
 410             rwlock.writer = None;
 411             trace!("rwlock_writer_unlock: {:?} unlocked by {:?}", id, expected_writer);
 412             // Release memory to both reader and writer vector clocks
 413             //  since this writer happens-before both the union of readers once they are finished
 414             //  and the next writer
 415             if let Some(data_race) = &this.memory.extra.data_race {
 416                 data_race.validate_lock_release(&mut rwlock.data_race, current_writer);
 417                 data_race.validate_lock_release(&mut rwlock.data_race_reader, current_writer);
 418             }
 419             // The thread was a writer.
 420             //
 421             // We are prioritizing writers here against the readers. As a
 422             // result, not only readers can starve writers, but also writers can
 423             // starve readers.
 424             if this.rwlock_dequeue_and_lock_writer(id) {
 425                 // Someone got the write lock, nice.
 426             } else {
 427                 // Give the lock to all readers.
 428                 while this.rwlock_dequeue_and_lock_reader(id) {
 429                     // Rinse and repeat.
 430                 }
 431             }
 432             true
 433         } else {
 434             false
 435         }
 436     }
 437
 438     #[inline]
 439     /// Put the writer in the queue waiting for the lock.
 440     fn rwlock_enqueue_and_block_writer(
 441         &mut self,
 442         id: RwLockId,
 443         writer: ThreadId,
 444     ) {
 445         let this = self.eval_context_mut();
 446         assert!(this.rwlock_is_locked(id), "write-queueing on unlocked rwlock");
 447         this.machine.threads.sync.rwlocks[id].writer_queue.push_back(writer);
 448         this.block_thread(writer);
 449     }
 450
 451     #[inline]
 452     /// Create state for a new conditional variable.
 453     fn condvar_create(&mut self) -> CondvarId {
 454         let this = self.eval_context_mut();
 455         this.machine.threads.sync.condvars.push(Default::default())
 456     }
 457
 458     #[inline]
 459     /// Is the conditional variable awaited?
 460     fn condvar_is_awaited(&mut self, id: CondvarId) -> bool {
 461         let this = self.eval_context_mut();
 462         !this.machine.threads.sync.condvars[id].waiters.is_empty()
 463     }
 464
 465     /// Mark that the thread is waiting on the conditional variable.
 466     fn condvar_wait(&mut self, id: CondvarId, thread: ThreadId, mutex: MutexId) {
 467         let this = self.eval_context_mut();
 468         let waiters = &mut this.machine.threads.sync.condvars[id].waiters;
 469         assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
 470         waiters.push_back(CondvarWaiter { thread, mutex });
 471     }
 472
 473     /// Wake up some thread (if there is any) sleeping on the conditional
 474     /// variable.
 475     fn condvar_signal(&mut self, id: CondvarId) -> Option<(ThreadId, MutexId)> {
 476         let this = self.eval_context_mut();
 477         let current_thread = this.get_active_thread();
 478         let condvar = &mut this.machine.threads.sync.condvars[id];
 479         let data_race = &this.memory.extra.data_race;
 480
 481         // Each condvar signal happens-before the end of the condvar wake
 482         if let Some(data_race) = data_race {
 483             data_race.validate_lock_release(&mut condvar.data_race, current_thread);
 484         }
 485         condvar.waiters
 486             .pop_front()
 487             .map(|waiter| {
 488                 if let Some(data_race) = data_race {
 489                     data_race.validate_lock_acquire(&mut condvar.data_race, waiter.thread);
 490                 }
 491                 (waiter.thread, waiter.mutex)
 492             })
 493     }
 494
 495     #[inline]
 496     /// Remove the thread from the queue of threads waiting on this conditional variable.
 497     fn condvar_remove_waiter(&mut self, id: CondvarId, thread: ThreadId) {
 498         let this = self.eval_context_mut();
 499         this.machine.threads.sync.condvars[id].waiters.retain(|waiter| waiter.thread != thread);
 500     }
 501
 502     fn futex_wait(&mut self, addr: Pointer<stacked_borrows::Tag>, thread: ThreadId) {
 503         let this = self.eval_context_mut();
 504         let futex = &mut this.machine.threads.sync.futexes.entry(addr.erase_tag()).or_default();
 505         let waiters = &mut futex.waiters;
 506         assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
 507         waiters.push_back(FutexWaiter { thread });
 508     }
 509
 510     fn futex_wake(&mut self, addr: Pointer<stacked_borrows::Tag>) -> Option<ThreadId> {
 511         let this = self.eval_context_mut();
 512         let current_thread = this.get_active_thread();
 513         let futex = &mut this.machine.threads.sync.futexes.get_mut(&addr.erase_tag())?;
 514         let data_race =  &this.memory.extra.data_race;
 515
 516         // Each futex-wake happens-before the end of the futex wait
 517         if let Some(data_race) = data_race {
 518             data_race.validate_lock_release(&mut futex.data_race, current_thread);
 519         }
 520         let res = futex.waiters.pop_front().map(|waiter| {
 521             if let Some(data_race) = data_race {
 522                 data_race.validate_lock_acquire(&futex.data_race, waiter.thread);
 523             }
 524             waiter.thread
 525         });
 526         res
 527     }
 528
 529     fn futex_remove_waiter(&mut self, addr: Pointer<stacked_borrows::Tag>, thread: ThreadId) {
 530         let this = self.eval_context_mut();
 531         if let Some(futex) = this.machine.threads.sync.futexes.get_mut(&addr.erase_tag()) {
 532             futex.waiters.retain(|waiter| waiter.thread != thread);
 533         }
 534     }
 535 }