use std::collections::{hash_map::Entry, VecDeque};
use std::num::NonZeroU32;
use std::ops::Not;
use log::trace;
use rustc_data_structures::fx::FxHashMap;
use rustc_index::{Idx, IndexVec};
use rustc_middle::ty::layout::TyAndLayout;
use super::init_once::InitOnce;
use super::vector_clock::VClock;
use crate::*;
pub trait SyncId {
fn from_u32(id: u32) -> Self;
fn to_u32(&self) -> u32;
}
macro_rules! declare_id {
($name: ident) => {
#[derive(Clone, Copy, Debug, PartialOrd, Ord, PartialEq, Eq, Hash)]
pub struct $name(NonZeroU32);
impl SyncId for $name {
fn from_u32(id: u32) -> Self {
Self(NonZeroU32::new(id).unwrap())
}
fn to_u32(&self) -> u32 {
self.0.get()
}
}
impl Idx for $name {
fn new(idx: usize) -> Self {
let shifted_idx = u32::try_from(idx).unwrap().checked_add(1).unwrap();
$name(NonZeroU32::new(shifted_idx).unwrap())
}
fn index(self) -> usize {
usize::try_from(self.0.get() - 1).unwrap()
}
}
impl $name {
pub fn to_u32_scalar(&self) -> Scalar<Provenance> {
Scalar::from_u32(self.0.get())
}
}
};
}
declare_id!(MutexId);
#[derive(Default, Debug)]
struct Mutex {
owner: Option<ThreadId>,
lock_count: usize,
queue: VecDeque<ThreadId>,
data_race: VClock,
}
declare_id!(RwLockId);
#[derive(Default, Debug)]
struct RwLock {
writer: Option<ThreadId>,
readers: FxHashMap<ThreadId, usize>,
writer_queue: VecDeque<ThreadId>,
reader_queue: VecDeque<ThreadId>,
data_race: VClock,
data_race_reader: VClock,
}
declare_id!(CondvarId);
#[derive(Debug, Copy, Clone)]
pub enum RwLockMode {
Read,
Write,
}
#[derive(Debug)]
pub enum CondvarLock {
Mutex(MutexId),
RwLock { id: RwLockId, mode: RwLockMode },
}
#[derive(Debug)]
struct CondvarWaiter {
thread: ThreadId,
lock: CondvarLock,
}
#[derive(Default, Debug)]
struct Condvar {
waiters: VecDeque<CondvarWaiter>,
data_race: VClock,
}
#[derive(Default, Debug)]
struct Futex {
waiters: VecDeque<FutexWaiter>,
data_race: VClock,
}
#[derive(Debug)]
struct FutexWaiter {
thread: ThreadId,
bitset: u32,
}
#[derive(Default, Debug)]
pub(crate) struct SynchronizationState<'mir, 'tcx> {
mutexes: IndexVec<MutexId, Mutex>,
rwlocks: IndexVec<RwLockId, RwLock>,
condvars: IndexVec<CondvarId, Condvar>,
futexes: FxHashMap<u64, Futex>,
pub(super) init_onces: IndexVec<InitOnceId, InitOnce<'mir, 'tcx>>,
}
impl<'mir, 'tcx> VisitTags for SynchronizationState<'mir, 'tcx> {
fn visit_tags(&self, visit: &mut dyn FnMut(BorTag)) {
for init_once in self.init_onces.iter() {
init_once.visit_tags(visit);
}
}
}
impl<'mir, 'tcx: 'mir> EvalContextExtPriv<'mir, 'tcx> for crate::MiriInterpCx<'mir, 'tcx> {}
pub(super) trait EvalContextExtPriv<'mir, 'tcx: 'mir>:
crate::MiriInterpCxExt<'mir, 'tcx>
{
#[inline]
fn get_or_create_id<Id: SyncId>(
&mut self,
next_id: Id,
lock_op: &OpTy<'tcx, Provenance>,
lock_layout: TyAndLayout<'tcx>,
offset: u64,
) -> InterpResult<'tcx, Option<Id>> {
let this = self.eval_context_mut();
let value_place =
this.deref_pointer_and_offset(lock_op, offset, lock_layout, this.machine.layouts.u32)?;
let (old, success) = this
.atomic_compare_exchange_scalar(
&value_place,
&ImmTy::from_uint(0u32, this.machine.layouts.u32),
Scalar::from_u32(next_id.to_u32()),
AtomicRwOrd::Relaxed, AtomicReadOrd::Relaxed,
false,
)?
.to_scalar_pair();
Ok(if success.to_bool().expect("compare_exchange's second return value is a bool") {
None
} else {
Some(Id::from_u32(old.to_u32().expect("layout is u32")))
})
}
#[inline]
fn rwlock_dequeue_and_lock_reader(&mut self, id: RwLockId) -> bool {
let this = self.eval_context_mut();
if let Some(reader) = this.machine.threads.sync.rwlocks[id].reader_queue.pop_front() {
this.unblock_thread(reader);
this.rwlock_reader_lock(id, reader);
true
} else {
false
}
}
#[inline]
fn rwlock_dequeue_and_lock_writer(&mut self, id: RwLockId) -> bool {
let this = self.eval_context_mut();
if let Some(writer) = this.machine.threads.sync.rwlocks[id].writer_queue.pop_front() {
this.unblock_thread(writer);
this.rwlock_writer_lock(id, writer);
true
} else {
false
}
}
#[inline]
fn mutex_dequeue_and_lock(&mut self, id: MutexId) -> bool {
let this = self.eval_context_mut();
if let Some(thread) = this.machine.threads.sync.mutexes[id].queue.pop_front() {
this.unblock_thread(thread);
this.mutex_lock(id, thread);
true
} else {
false
}
}
}
impl<'mir, 'tcx: 'mir> EvalContextExt<'mir, 'tcx> for crate::MiriInterpCx<'mir, 'tcx> {}
pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {
fn mutex_get_or_create_id(
&mut self,
lock_op: &OpTy<'tcx, Provenance>,
lock_layout: TyAndLayout<'tcx>,
offset: u64,
) -> InterpResult<'tcx, MutexId> {
let this = self.eval_context_mut();
this.mutex_get_or_create(|ecx, next_id| {
ecx.get_or_create_id(next_id, lock_op, lock_layout, offset)
})
}
fn rwlock_get_or_create_id(
&mut self,
lock_op: &OpTy<'tcx, Provenance>,
lock_layout: TyAndLayout<'tcx>,
offset: u64,
) -> InterpResult<'tcx, RwLockId> {
let this = self.eval_context_mut();
this.rwlock_get_or_create(|ecx, next_id| {
ecx.get_or_create_id(next_id, lock_op, lock_layout, offset)
})
}
fn condvar_get_or_create_id(
&mut self,
lock_op: &OpTy<'tcx, Provenance>,
lock_layout: TyAndLayout<'tcx>,
offset: u64,
) -> InterpResult<'tcx, CondvarId> {
let this = self.eval_context_mut();
this.condvar_get_or_create(|ecx, next_id| {
ecx.get_or_create_id(next_id, lock_op, lock_layout, offset)
})
}
#[inline]
fn mutex_get_or_create<F>(&mut self, existing: F) -> InterpResult<'tcx, MutexId>
where
F: FnOnce(&mut MiriInterpCx<'mir, 'tcx>, MutexId) -> InterpResult<'tcx, Option<MutexId>>,
{
let this = self.eval_context_mut();
let next_index = this.machine.threads.sync.mutexes.next_index();
if let Some(old) = existing(this, next_index)? {
Ok(old)
} else {
let new_index = this.machine.threads.sync.mutexes.push(Default::default());
assert_eq!(next_index, new_index);
Ok(new_index)
}
}
#[inline]
fn mutex_get_owner(&mut self, id: MutexId) -> ThreadId {
let this = self.eval_context_ref();
this.machine.threads.sync.mutexes[id].owner.unwrap()
}
#[inline]
fn mutex_is_locked(&self, id: MutexId) -> bool {
let this = self.eval_context_ref();
this.machine.threads.sync.mutexes[id].owner.is_some()
}
fn mutex_lock(&mut self, id: MutexId, thread: ThreadId) {
let this = self.eval_context_mut();
let mutex = &mut this.machine.threads.sync.mutexes[id];
if let Some(current_owner) = mutex.owner {
assert_eq!(thread, current_owner, "mutex already locked by another thread");
assert!(
mutex.lock_count > 0,
"invariant violation: lock_count == 0 iff the thread is unlocked"
);
} else {
mutex.owner = Some(thread);
}
mutex.lock_count = mutex.lock_count.checked_add(1).unwrap();
if let Some(data_race) = &this.machine.data_race {
data_race.validate_lock_acquire(&mutex.data_race, thread);
}
}
fn mutex_unlock(&mut self, id: MutexId, expected_owner: ThreadId) -> Option<usize> {
let this = self.eval_context_mut();
let current_span = this.machine.current_span();
let mutex = &mut this.machine.threads.sync.mutexes[id];
if let Some(current_owner) = mutex.owner {
if current_owner != expected_owner {
return None;
}
let old_lock_count = mutex.lock_count;
mutex.lock_count = old_lock_count
.checked_sub(1)
.expect("invariant violation: lock_count == 0 iff the thread is unlocked");
if mutex.lock_count == 0 {
mutex.owner = None;
if let Some(data_race) = &this.machine.data_race {
data_race.validate_lock_release(
&mut mutex.data_race,
current_owner,
current_span,
);
}
this.mutex_dequeue_and_lock(id);
}
Some(old_lock_count)
} else {
None
}
}
#[inline]
fn mutex_enqueue_and_block(&mut self, id: MutexId, thread: ThreadId) {
let this = self.eval_context_mut();
assert!(this.mutex_is_locked(id), "queing on unlocked mutex");
this.machine.threads.sync.mutexes[id].queue.push_back(thread);
this.block_thread(thread);
}
#[inline]
fn rwlock_get_or_create<F>(&mut self, existing: F) -> InterpResult<'tcx, RwLockId>
where
F: FnOnce(&mut MiriInterpCx<'mir, 'tcx>, RwLockId) -> InterpResult<'tcx, Option<RwLockId>>,
{
let this = self.eval_context_mut();
let next_index = this.machine.threads.sync.rwlocks.next_index();
if let Some(old) = existing(this, next_index)? {
Ok(old)
} else {
let new_index = this.machine.threads.sync.rwlocks.push(Default::default());
assert_eq!(next_index, new_index);
Ok(new_index)
}
}
#[inline]
fn rwlock_is_locked(&self, id: RwLockId) -> bool {
let this = self.eval_context_ref();
let rwlock = &this.machine.threads.sync.rwlocks[id];
trace!(
"rwlock_is_locked: {:?} writer is {:?} and there are {} reader threads (some of which could hold multiple read locks)",
id,
rwlock.writer,
rwlock.readers.len(),
);
rwlock.writer.is_some() || rwlock.readers.is_empty().not()
}
#[inline]
fn rwlock_is_write_locked(&self, id: RwLockId) -> bool {
let this = self.eval_context_ref();
let rwlock = &this.machine.threads.sync.rwlocks[id];
trace!("rwlock_is_write_locked: {:?} writer is {:?}", id, rwlock.writer);
rwlock.writer.is_some()
}
fn rwlock_reader_lock(&mut self, id: RwLockId, reader: ThreadId) {
let this = self.eval_context_mut();
assert!(!this.rwlock_is_write_locked(id), "the lock is write locked");
trace!("rwlock_reader_lock: {:?} now also held (one more time) by {:?}", id, reader);
let rwlock = &mut this.machine.threads.sync.rwlocks[id];
let count = rwlock.readers.entry(reader).or_insert(0);
*count = count.checked_add(1).expect("the reader counter overflowed");
if let Some(data_race) = &this.machine.data_race {
data_race.validate_lock_acquire(&rwlock.data_race, reader);
}
}
fn rwlock_reader_unlock(&mut self, id: RwLockId, reader: ThreadId) -> bool {
let this = self.eval_context_mut();
let current_span = this.machine.current_span();
let rwlock = &mut this.machine.threads.sync.rwlocks[id];
match rwlock.readers.entry(reader) {
Entry::Occupied(mut entry) => {
let count = entry.get_mut();
assert!(*count > 0, "rwlock locked with count == 0");
*count -= 1;
if *count == 0 {
trace!("rwlock_reader_unlock: {:?} no longer held by {:?}", id, reader);
entry.remove();
} else {
trace!("rwlock_reader_unlock: {:?} held one less time by {:?}", id, reader);
}
}
Entry::Vacant(_) => return false, }
if let Some(data_race) = &this.machine.data_race {
data_race.validate_lock_release_shared(
&mut rwlock.data_race_reader,
reader,
current_span,
);
}
if this.rwlock_is_locked(id).not() {
let rwlock = &mut this.machine.threads.sync.rwlocks[id];
rwlock.data_race.clone_from(&rwlock.data_race_reader);
this.rwlock_dequeue_and_lock_writer(id);
}
true
}
#[inline]
fn rwlock_enqueue_and_block_reader(&mut self, id: RwLockId, reader: ThreadId) {
let this = self.eval_context_mut();
assert!(this.rwlock_is_write_locked(id), "read-queueing on not write locked rwlock");
this.machine.threads.sync.rwlocks[id].reader_queue.push_back(reader);
this.block_thread(reader);
}
#[inline]
fn rwlock_writer_lock(&mut self, id: RwLockId, writer: ThreadId) {
let this = self.eval_context_mut();
assert!(!this.rwlock_is_locked(id), "the rwlock is already locked");
trace!("rwlock_writer_lock: {:?} now held by {:?}", id, writer);
let rwlock = &mut this.machine.threads.sync.rwlocks[id];
rwlock.writer = Some(writer);
if let Some(data_race) = &this.machine.data_race {
data_race.validate_lock_acquire(&rwlock.data_race, writer);
}
}
#[inline]
fn rwlock_writer_unlock(&mut self, id: RwLockId, expected_writer: ThreadId) -> bool {
let this = self.eval_context_mut();
let current_span = this.machine.current_span();
let rwlock = &mut this.machine.threads.sync.rwlocks[id];
if let Some(current_writer) = rwlock.writer {
if current_writer != expected_writer {
return false;
}
rwlock.writer = None;
trace!("rwlock_writer_unlock: {:?} unlocked by {:?}", id, expected_writer);
if let Some(data_race) = &this.machine.data_race {
data_race.validate_lock_release(
&mut rwlock.data_race,
current_writer,
current_span,
);
data_race.validate_lock_release(
&mut rwlock.data_race_reader,
current_writer,
current_span,
);
}
if this.rwlock_dequeue_and_lock_writer(id) {
} else {
while this.rwlock_dequeue_and_lock_reader(id) {
}
}
true
} else {
false
}
}
#[inline]
fn rwlock_enqueue_and_block_writer(&mut self, id: RwLockId, writer: ThreadId) {
let this = self.eval_context_mut();
assert!(this.rwlock_is_locked(id), "write-queueing on unlocked rwlock");
this.machine.threads.sync.rwlocks[id].writer_queue.push_back(writer);
this.block_thread(writer);
}
#[inline]
fn condvar_get_or_create<F>(&mut self, existing: F) -> InterpResult<'tcx, CondvarId>
where
F: FnOnce(
&mut MiriInterpCx<'mir, 'tcx>,
CondvarId,
) -> InterpResult<'tcx, Option<CondvarId>>,
{
let this = self.eval_context_mut();
let next_index = this.machine.threads.sync.condvars.next_index();
if let Some(old) = existing(this, next_index)? {
Ok(old)
} else {
let new_index = this.machine.threads.sync.condvars.push(Default::default());
assert_eq!(next_index, new_index);
Ok(new_index)
}
}
#[inline]
fn condvar_is_awaited(&mut self, id: CondvarId) -> bool {
let this = self.eval_context_mut();
!this.machine.threads.sync.condvars[id].waiters.is_empty()
}
fn condvar_wait(&mut self, id: CondvarId, thread: ThreadId, lock: CondvarLock) {
let this = self.eval_context_mut();
let waiters = &mut this.machine.threads.sync.condvars[id].waiters;
assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
waiters.push_back(CondvarWaiter { thread, lock });
}
fn condvar_signal(&mut self, id: CondvarId) -> Option<(ThreadId, CondvarLock)> {
let this = self.eval_context_mut();
let current_thread = this.get_active_thread();
let current_span = this.machine.current_span();
let condvar = &mut this.machine.threads.sync.condvars[id];
let data_race = &this.machine.data_race;
if let Some(data_race) = data_race {
data_race.validate_lock_release(&mut condvar.data_race, current_thread, current_span);
}
condvar.waiters.pop_front().map(|waiter| {
if let Some(data_race) = data_race {
data_race.validate_lock_acquire(&condvar.data_race, waiter.thread);
}
(waiter.thread, waiter.lock)
})
}
#[inline]
fn condvar_remove_waiter(&mut self, id: CondvarId, thread: ThreadId) {
let this = self.eval_context_mut();
this.machine.threads.sync.condvars[id].waiters.retain(|waiter| waiter.thread != thread);
}
fn futex_wait(&mut self, addr: u64, thread: ThreadId, bitset: u32) {
let this = self.eval_context_mut();
let futex = &mut this.machine.threads.sync.futexes.entry(addr).or_default();
let waiters = &mut futex.waiters;
assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
waiters.push_back(FutexWaiter { thread, bitset });
}
fn futex_wake(&mut self, addr: u64, bitset: u32) -> Option<ThreadId> {
let this = self.eval_context_mut();
let current_thread = this.get_active_thread();
let current_span = this.machine.current_span();
let futex = &mut this.machine.threads.sync.futexes.get_mut(&addr)?;
let data_race = &this.machine.data_race;
if let Some(data_race) = data_race {
data_race.validate_lock_release(&mut futex.data_race, current_thread, current_span);
}
futex.waiters.iter().position(|w| w.bitset & bitset != 0).map(|i| {
let waiter = futex.waiters.remove(i).unwrap();
if let Some(data_race) = data_race {
data_race.validate_lock_acquire(&futex.data_race, waiter.thread);
}
waiter.thread
})
}
fn futex_remove_waiter(&mut self, addr: u64, thread: ThreadId) {
let this = self.eval_context_mut();
if let Some(futex) = this.machine.threads.sync.futexes.get_mut(&addr) {
futex.waiters.retain(|waiter| waiter.thread != thread);
}
}
}