use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicPtr, AtomicU64, AtomicUsize, Ordering};
use crate::server::record::RecordCell;
use crate::runtime::sync::{
MutexInfo, PriorityInheritanceMutex, PriorityInheritanceMutexGuard, mutex_report,
};
use super::PvDatabase;
struct LockSet {
id: u64,
lock: PriorityInheritanceMutex<()>,
mutex_seq: u64,
owner: AtomicU64,
depth: AtomicUsize,
many_holds: AtomicUsize,
held: std::cell::UnsafeCell<Option<PriorityInheritanceMutexGuard<'static, ()>>>,
}
unsafe impl Sync for LockSet {}
type Set = &'static LockSet;
pub(crate) struct LockRecord {
plock_set: AtomicPtr<LockSet>,
}
impl LockRecord {
fn new(set: Set) -> Arc<Self> {
Arc::new(Self {
plock_set: AtomicPtr::new(Self::as_ptr(set)),
})
}
pub(crate) fn bootstrap() -> Arc<Self> {
Self::new(bootstrap_set())
}
pub(crate) fn is_bootstrap(&self) -> bool {
std::ptr::eq(self.set(), bootstrap_set())
}
#[inline]
pub(crate) fn acquire(&self) -> SetGuard {
let set = self.set();
if set.owner.load(Ordering::Acquire) == thread_key() {
return set.reenter(false);
}
self.acquire_fresh()
}
#[cold]
#[inline(never)]
fn acquire_fresh(&self) -> SetGuard {
loop {
let set = self.set();
let guard = set.acquire(false);
if std::ptr::eq(self.set(), set) {
return guard;
}
drop(guard);
}
}
pub(crate) fn unheld<R>(&self, f: impl FnOnce() -> R) -> R {
self.set().unheld(f)
}
fn as_ptr(set: Set) -> *mut LockSet {
set as *const LockSet as *mut LockSet
}
fn set(&self) -> Set {
let p = self.plock_set.load(Ordering::Acquire);
unsafe { &*p }
}
fn store(&self, set: Set) {
self.plock_set.store(Self::as_ptr(set), Ordering::Release);
}
}
const FIRST_SET_ID: u64 = 2;
static SET_MUTEX_SEQ: std::sync::Mutex<u64> = std::sync::Mutex::new(0);
fn new_set(id: u64) -> Set {
let mut seq = SET_MUTEX_SEQ.lock().unwrap_or_else(|e| e.into_inner());
let mutex_seq = *seq;
*seq += 1;
let set: Set = Box::leak(Box::new(LockSet {
id,
lock: PriorityInheritanceMutex::new(()),
mutex_seq,
owner: AtomicU64::new(0),
depth: AtomicUsize::new(0),
many_holds: AtomicUsize::new(0),
held: std::cell::UnsafeCell::new(None),
}));
drop(seq);
set
}
fn bootstrap_set() -> Set {
static BOOTSTRAP: std::sync::OnceLock<Set> = std::sync::OnceLock::new();
BOOTSTRAP.get_or_init(|| new_set(0))
}
fn thread_key() -> u64 {
static NEXT: AtomicU64 = AtomicU64::new(1);
thread_local! {
static KEY: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}
KEY.with(|k| match k.get() {
0 => {
let minted = NEXT.fetch_add(1, Ordering::Relaxed);
k.set(minted);
minted
}
key => key,
})
}
impl LockSet {
fn acquire(&'static self, many: bool) -> SetGuard {
let me = thread_key();
if self.owner.load(Ordering::Acquire) == me {
return self.reenter(many);
}
self.lock_fresh(me);
if many {
self.many_holds.fetch_add(1, Ordering::Relaxed);
}
SetGuard {
set: self,
outermost: true,
many,
}
}
#[inline]
fn reenter(&'static self, many: bool) -> SetGuard {
self.depth
.store(self.depth.load(Ordering::Relaxed) + 1, Ordering::Relaxed);
if many {
self.many_holds.fetch_add(1, Ordering::Relaxed);
}
SetGuard {
set: self,
outermost: false,
many,
}
}
fn lock_fresh(&'static self, me: u64) {
debug_assert!(
!super::registration_gate_held(),
"a lock set (L1) was taken while this thread holds the registration \
gate (L46); record data is behind L1 and must be read before the \
gate is taken, or through a set this thread already holds"
);
debug_assert!(
!super::map_read_held(),
"a lock set (L1) was taken while this thread holds a read guard on \
the records map or the alias table; clone the record handle out \
under the guard and drop it before locking the record"
);
let guard = self.lock.lock();
unsafe { *self.held.get() = Some(guard) };
self.owner.store(me, Ordering::Release);
self.depth.store(1, Ordering::Relaxed);
}
fn unlock_outermost(&'static self) {
self.depth.store(0, Ordering::Relaxed);
self.owner.store(0, Ordering::Release);
let guard = unsafe { (*self.held.get()).take() };
drop(guard);
}
fn unheld<R>(&'static self, f: impl FnOnce() -> R) -> R {
let me = thread_key();
assert_eq!(
self.owner.load(Ordering::Acquire),
me,
"a lock set was given up by a thread that does not hold it"
);
assert_eq!(
self.depth.load(Ordering::Relaxed),
1,
"a lock set was given up with a nested hold live"
);
assert_eq!(
self.many_holds.load(Ordering::Relaxed),
0,
"a lock set was given up from inside a many-lock transaction"
);
self.unlock_outermost();
let r = f();
self.lock_fresh(me);
r
}
fn is_locked(&self) -> bool {
self.lock.try_lock().is_none()
}
}
pub(crate) struct SetGuard {
set: Set,
outermost: bool,
many: bool,
}
impl SetGuard {
#[inline]
pub(crate) fn holds(&self, record: &LockRecord) -> bool {
std::ptr::eq(record.set(), self.set)
}
}
impl Drop for SetGuard {
fn drop(&mut self) {
if self.many {
self.set.many_holds.fetch_sub(1, Ordering::Relaxed);
}
if self.outermost {
self.set.unlock_outermost();
} else {
self.set.depth.store(
self.set.depth.load(Ordering::Relaxed) - 1,
Ordering::Relaxed,
);
}
}
}
fn hold_sorted(sets: &[Set]) -> Vec<SetGuard> {
let mut sets: Vec<Set> = sets.to_vec();
sets.sort_unstable_by_key(|set| set.id);
sets.dedup_by_key(|set| set.id);
sets.into_iter().map(|set| set.acquire(false)).collect()
}
struct Registry {
active: BTreeMap<u64, SetState>,
free: VecDeque<Set>,
of_record: HashMap<String, Arc<LockRecord>>,
next_id: u64,
revision: u64,
built: bool,
}
struct SetState {
set: Set,
members: BTreeSet<String>,
}
impl Default for Registry {
fn default() -> Self {
Self {
active: BTreeMap::new(),
free: VecDeque::new(),
of_record: HashMap::new(),
next_id: FIRST_SET_ID - 1,
revision: 0,
built: false,
}
}
}
impl Registry {
fn make_set(&mut self) -> Set {
if let Some(set) = self.free.pop_front() {
debug_assert_eq!(
set.many_holds.load(Ordering::Relaxed),
0,
"C asserts refcount==0 for a set on the free list (dbLock.c:571)"
);
return set;
}
self.next_id += 1;
self.revision += 1;
new_set(self.next_id)
}
fn real_set_of(&self, name: &str) -> Option<Set> {
self.of_record
.get(name)
.map(|lr| lr.set())
.filter(|set| !std::ptr::eq(*set, bootstrap_set()))
}
fn set_of_registered(&self, name: &str) -> Option<Set> {
let set = self.real_set_of(name);
debug_assert!(
set.is_some() || !self.of_record.contains_key(name),
"registered record {name} is on the bootstrap set after iocInit"
);
set
}
fn lock_record_of(&mut self, name: &str) -> Arc<LockRecord> {
if let Some(lr) = self.of_record.get(name) {
return lr.clone();
}
let set = self.make_set();
let lr = LockRecord::new(set);
self.of_record.insert(name.to_string(), lr.clone());
self.active.insert(
set.id,
SetState {
set,
members: BTreeSet::from([name.to_string()]),
},
);
lr
}
fn adopt(&mut self, name: &str, lr: &Arc<LockRecord>) {
match self.of_record.get(name) {
Some(existing) if Arc::ptr_eq(existing, lr) => {}
Some(existing) => {
lr.store(existing.set());
self.of_record.insert(name.to_string(), lr.clone());
self.revision += 1;
}
None => {
self.of_record.insert(name.to_string(), lr.clone());
if self.built {
self.mint_for(name, lr);
}
}
}
}
fn mint_for(&mut self, name: &str, lr: &Arc<LockRecord>) {
if !lr.is_bootstrap() {
return;
}
let set = self.make_set();
lr.store(set);
self.active.insert(
set.id,
SetState {
set,
members: BTreeSet::from([name.to_string()]),
},
);
self.of_record
.entry(name.to_string())
.or_insert_with(|| lr.clone());
}
fn place(&mut self, names: impl IntoIterator<Item = String>, set: Set) {
self.revision += 1;
for name in names {
match self.of_record.get(&name) {
Some(lr) => lr.store(set),
None => {
self.of_record.insert(name, LockRecord::new(set));
}
}
}
}
fn merge(&mut self, first: &str, second: &str) {
let (Some(a), Some(b)) = (self.real_set_of(first), self.real_set_of(second)) else {
return;
};
let (a, b) = (a.id, b.id);
if a == b {
return;
}
let moved = self
.active
.remove(&b)
.expect("every id in of_record names an active set");
let survivor = self.active[&a].set;
self.place(moved.members.iter().cloned(), survivor);
let target = self
.active
.get_mut(&a)
.expect("every id in of_record names an active set");
target.members.extend(moved.members);
self.free.push_back(moved.set);
}
fn repartition(
&mut self,
seed: &str,
adjacency: &HashMap<String, BTreeSet<String>>,
held: &[Set],
) {
let Some(seed_set) = self.of_record.get(seed).map(|lr| lr.set()) else {
return;
};
let seed_id = seed_set.id;
let mut affected: BTreeSet<String> = BTreeSet::new();
let mut work: Vec<String> = vec![seed.to_string()];
while let Some(name) = work.pop() {
if !affected.insert(name.clone()) {
continue;
}
if let Some(targets) = adjacency.get(&name) {
work.extend(targets.iter().cloned());
}
if let Some(id) = self.real_set_of(&name).map(|set| set.id) {
work.extend(self.active[&id].members.iter().cloned());
}
}
let touched: BTreeSet<u64> = affected
.iter()
.filter_map(|name| self.real_set_of(name).map(|set| set.id))
.collect();
debug_assert!(
affected.iter().all(|name| {
self.of_record
.get(name)
.is_none_or(|lr| held.iter().any(|set| std::ptr::eq(*set, lr.set())))
}),
"repartition reached a record behind a set the caller does not hold"
);
for name in &affected {
if !adjacency.contains_key(name) {
self.of_record.remove(name);
}
}
let seed_present = adjacency.contains_key(seed);
let mut components: Vec<BTreeSet<String>> = Vec::new();
let mut placed: BTreeSet<String> = BTreeSet::new();
let starts = seed_present
.then(|| seed.to_string())
.into_iter()
.chain(affected.iter().cloned());
for start in starts {
if placed.contains(&start) || !adjacency.contains_key(&start) {
continue;
}
let mut component: BTreeSet<String> = BTreeSet::new();
let mut walk = vec![start];
while let Some(name) = walk.pop() {
if !component.insert(name.clone()) {
continue;
}
if let Some(targets) = adjacency.get(&name) {
walk.extend(targets.iter().cloned());
}
}
placed.extend(component.iter().cloned());
components.push(component);
}
let mut keeps: BTreeSet<u64> = BTreeSet::new();
if seed_present {
keeps.insert(seed_id);
}
let mut assigned: Vec<Option<u64>> = Vec::with_capacity(components.len());
for component in &components {
if seed_present && component.contains(seed) {
assigned.push(Some(seed_id));
continue;
}
let mut ids = component
.iter()
.map(|name| self.real_set_of(name).map(|set| set.id));
let first = ids.next().flatten();
let uniform =
first.filter(|id| !keeps.contains(id) && ids.all(|other| other == Some(*id)));
if let Some(id) = uniform {
keeps.insert(id);
}
assigned.push(uniform);
}
let fresh: Vec<Set> = assigned
.iter()
.filter(|id| id.is_none())
.map(|_| self.make_set())
.collect();
let mut fresh = fresh.into_iter();
for (component, id) in components.into_iter().zip(assigned) {
let set = match id {
Some(id) => self.active[&id].set,
None => fresh
.next()
.expect("one fresh set per unassigned component"),
};
self.place(component.iter().cloned(), set);
self.active.insert(
set.id,
SetState {
set,
members: component,
},
);
}
for id in touched.difference(&keeps) {
let dropped = self
.active
.remove(id)
.expect("every touched id named an active set");
self.free.push_back(dropped.set);
}
self.revision += 1;
}
fn every_set(&self) -> Vec<Set> {
self.active.values().map(|state| state.set).collect()
}
fn info(&self, id: u64, rows: &HashMap<u64, MutexInfo>) -> LockSetInfo {
let state = &self.active[&id];
LockSetInfo {
id,
members: state.members.iter().cloned().collect(),
refs: state.members.len() + state.set.many_holds.load(Ordering::Relaxed),
locked: state.set.is_locked(),
mutex: rows.get(&state.set.mutex_seq).cloned(),
}
}
}
fn lock_set_mutex_rows() -> HashMap<u64, MutexInfo> {
mutex_report(false)
.shown
.into_iter()
.filter(|info| info.file() == file!())
.enumerate()
.map(|(seq, info)| (seq as u64, info))
.collect()
}
pub struct LockSetInfo {
pub id: u64,
pub members: Vec<String>,
pub refs: usize,
pub locked: bool,
pub mutex: Option<MutexInfo>,
}
pub struct LockSetReport {
pub active: Vec<LockSetInfo>,
pub free: usize,
}
#[derive(Default)]
pub(crate) struct RecordLockRegistry {
inner: std::sync::Mutex<Registry>,
}
impl RecordLockRegistry {
fn lock(&self) -> std::sync::MutexGuard<'_, Registry> {
self.inner
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
fn set_id_of(&self, record: &str) -> Option<u64> {
self.lock().real_set_of(record).map(|set| set.id)
}
pub(crate) fn adopt(&self, name: &str, lr: &Arc<LockRecord>) {
self.lock().adopt(name, lr);
}
}
impl PvDatabase {
pub fn build_lock_sets(&self) {
let mut names: Vec<String> = self
.inner
.records
.read()
.keys()
.map(|n| n.to_string())
.collect();
names.sort();
let mut edges: Vec<(String, String)> = Vec::new();
for name in &names {
for target in self.db_link_targets(name) {
edges.push((name.clone(), target));
}
}
self.init_sets(&names);
for (from, to) in edges {
self.merge_sets(&from, &to);
}
}
fn init_sets(&self, names: &[String]) {
let _out_of = bootstrap_set().acquire(false);
let mut registry = self.inner.record_locks.lock();
for name in names {
let lr = registry.lock_record_of(name);
registry.mint_for(name, &lr);
}
let mut adopted: Vec<(String, Arc<LockRecord>)> = registry
.of_record
.iter()
.filter(|(_, lr)| lr.is_bootstrap())
.map(|(name, lr)| (name.clone(), lr.clone()))
.collect();
adopted.sort_by(|a, b| a.0.cmp(&b.0));
for (name, lr) in &adopted {
registry.mint_for(name, lr);
}
registry.built = true;
}
fn merge_sets(&self, first: &str, second: &str) {
loop {
let (a, b) = {
let registry = self.inner.record_locks.lock();
match (registry.real_set_of(first), registry.real_set_of(second)) {
(Some(a), Some(b)) => (a, b),
_ => return,
}
};
if std::ptr::eq(a, b) {
return;
}
let _held = hold_sorted(&[a, b]);
let mut registry = self.inner.record_locks.lock();
let unchanged = registry
.real_set_of(first)
.is_some_and(|set| std::ptr::eq(set, a))
&& registry
.real_set_of(second)
.is_some_and(|set| std::ptr::eq(set, b));
if unchanged {
registry.merge(first, second);
return;
}
}
}
fn ensure_set(&self, name: &str) -> Arc<LockRecord> {
let lr = self.inner.record_locks.lock().lock_record_of(name);
self.ensure_set_for(name, &lr);
lr
}
fn ensure_set_for(&self, name: &str, lr: &Arc<LockRecord>) {
if !lr.is_bootstrap() {
return;
}
let _out_of = bootstrap_set().acquire(false);
self.inner.record_locks.lock().mint_for(name, lr);
}
fn db_link_targets(&self, record: &str) -> Vec<String> {
use crate::server::record::ParsedLink;
self.record_link_fields(record)
.into_iter()
.filter_map(|(_, _, parsed)| match parsed {
ParsedLink::Db(link) => {
let name = self
.resolve_alias(&link.target().record)
.unwrap_or_else(|| link.target().record.clone());
self.get_record_no_resolve(&name).map(|_| name)
}
_ => None,
})
.collect()
}
pub fn lock_set_report(&self) -> LockSetReport {
let rows = lock_set_mutex_rows();
let registry = self.inner.record_locks.lock();
LockSetReport {
active: registry
.active
.keys()
.map(|id| registry.info(*id, &rows))
.collect(),
free: registry.free.len(),
}
}
pub fn lock_set_of(&self, record: &str) -> Option<LockSetInfo> {
let canonical = self
.resolve_alias(record)
.unwrap_or_else(|| record.to_string());
let rows = lock_set_mutex_rows();
let registry = self.inner.record_locks.lock();
let id = registry.real_set_of(&canonical)?.id;
Some(registry.info(id, &rows))
}
pub(crate) fn link_field_write<'a>(
&'a self,
record: &str,
field: &str,
) -> Option<LockSetEdit<'a>> {
let canonical = self
.resolve_alias(record)
.unwrap_or_else(|| record.to_string());
if !self.is_dbf_link_field(&canonical, field) {
return None;
}
if !self.inner.record_locks.lock().built {
return None;
}
Some(LockSetEdit {
db: self,
record: canonical,
scope: RelinkScope::LinkEdit,
})
}
pub(crate) fn lock_set_membership_change<'a>(
&'a self,
record: &str,
) -> Option<LockSetEdit<'a>> {
if !self.inner.record_locks.lock().built {
return None;
}
Some(LockSetEdit {
db: self,
record: self
.resolve_alias(record)
.unwrap_or_else(|| record.to_string()),
scope: RelinkScope::Membership,
})
}
fn relink_lock_sets(&self, record: &str, scope: RelinkScope) {
let mut region: Vec<Set> = Vec::new();
loop {
let revision = {
let registry = self.inner.record_locks.lock();
match scope {
RelinkScope::Membership => region = registry.every_set(),
RelinkScope::LinkEdit => {
if region.is_empty() {
region.extend(registry.real_set_of(record));
}
}
}
registry.revision
};
if scope == RelinkScope::LinkEdit {
let targets = self.db_link_targets(record);
let registry = self.inner.record_locks.lock();
for target in &targets {
let Some(set) = registry.set_of_registered(target) else {
continue;
};
if !region.iter().any(|held| std::ptr::eq(*held, set)) {
region.push(set);
}
}
}
let held = hold_sorted(®ion);
let adjacency = match scope {
RelinkScope::Membership => self.db_link_adjacency(),
RelinkScope::LinkEdit => {
let members: Vec<String> = {
let registry = self.inner.record_locks.lock();
region
.iter()
.filter_map(|set| registry.active.get(&set.id))
.flat_map(|state| state.members.iter().cloned())
.collect()
};
self.db_link_adjacency_of(&members)
}
};
let mut registry = self.inner.record_locks.lock();
if registry.revision != revision {
drop(registry);
drop(held);
continue;
}
let outside: Vec<Set> = adjacency
.values()
.flatten()
.filter_map(|name| registry.set_of_registered(name))
.filter(|set| !region.iter().any(|held| std::ptr::eq(*held, *set)))
.collect();
if !outside.is_empty() {
region.extend(outside);
region.sort_unstable_by_key(|set| set.id);
region.dedup_by_key(|set| set.id);
drop(registry);
drop(held);
continue;
}
registry.repartition(record, &adjacency, ®ion);
return;
}
}
fn db_link_adjacency(&self) -> HashMap<String, BTreeSet<String>> {
let names: Vec<String> = self
.inner
.records
.read()
.keys()
.map(|n| n.to_string())
.collect();
self.db_link_adjacency_of(&names)
}
fn db_link_adjacency_of(&self, names: &[String]) -> HashMap<String, BTreeSet<String>> {
let mut adjacency: HashMap<String, BTreeSet<String>> = HashMap::new();
for name in names {
adjacency.entry(name.clone()).or_default();
}
for name in names {
for target in self.db_link_targets(name) {
adjacency
.entry(name.clone())
.or_default()
.insert(target.clone());
adjacency.entry(target).or_default().insert(name.clone());
}
}
adjacency
}
}
#[must_use = "the lock-set graph is only re-derived when this is dropped; binding it to `_` drops it immediately and relinks too early"]
pub(crate) struct LockSetEdit<'a> {
db: &'a PvDatabase,
record: String,
scope: RelinkScope,
}
impl Drop for LockSetEdit<'_> {
fn drop(&mut self) {
self.db.relink_lock_sets(&self.record, self.scope);
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum RelinkScope {
LinkEdit,
Membership,
}
#[must_use = "the lock set is released as soon as the guard is dropped"]
pub struct RecordWriteGuard {
_guard: SetGuard,
}
#[must_use = "the locked epoch ends as soon as the guard is dropped"]
pub struct ManyRecordWriteGuard {
_guards: Vec<SetGuard>,
}
impl PvDatabase {
pub fn lock_record(&self, record: &str) -> RecordWriteGuard {
let resolved = self.resolve_alias(record);
let canonical: &str = resolved.as_deref().unwrap_or(record);
let lr = self.ensure_set(canonical);
RecordWriteGuard {
_guard: lr.acquire(),
}
}
pub fn lock_instance(&self, rec: &Arc<RecordCell>) -> RecordWriteGuard {
let lr = rec.lock_record();
if lr.is_bootstrap() {
let name = rec.read().name.clone();
self.ensure_set_for(&name, lr);
}
RecordWriteGuard {
_guard: lr.acquire(),
}
}
pub fn lock_records<I, S>(&self, records: I) -> ManyRecordWriteGuard
where
I: IntoIterator<Item = S>,
S: AsRef<str>,
{
let names: Vec<String> = records
.into_iter()
.map(|record| {
let record = record.as_ref();
self.resolve_alias(record)
.unwrap_or_else(|| record.to_string())
})
.collect();
let cells: Vec<Arc<LockRecord>> = names.iter().map(|name| self.ensure_set(name)).collect();
loop {
let mut sets: Vec<Set> = cells.iter().map(|lr| lr.set()).collect();
sets.sort_unstable_by_key(|set| set.id);
sets.dedup_by_key(|set| set.id);
let guards: Vec<SetGuard> = sets.iter().map(|set| set.acquire(true)).collect();
let held: BTreeSet<u64> = sets.iter().map(|set| set.id).collect();
if names
.iter()
.all(|name| matches!(self.inner.record_locks.set_id_of(name), Some(id) if held.contains(&id)))
{
return ManyRecordWriteGuard { _guards: guards };
}
drop(guards);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
const SETTLE: Duration = Duration::from_millis(50);
#[test]
fn lock_record_excludes_same_record() {
let db = PvDatabase::new();
let order = Arc::new(AtomicUsize::new(0));
let g = db.lock_record("ai:1");
let db2 = db.clone();
let order2 = order.clone();
let h = std::thread::spawn(move || {
let _g2 = db2.lock_record("ai:1");
order2.fetch_add(10, Ordering::SeqCst);
});
std::thread::sleep(SETTLE);
assert_eq!(order.load(Ordering::SeqCst), 0);
order.fetch_add(1, Ordering::SeqCst);
drop(g);
h.join().unwrap();
assert_eq!(order.load(Ordering::SeqCst), 11);
}
#[test]
fn lock_records_excludes_single_member_write() {
let db = PvDatabase::new();
let many = db.lock_records(["g:a", "g:b", "g:c"]);
let db2 = db.clone();
let acquired = Arc::new(AtomicUsize::new(0));
let acquired2 = acquired.clone();
let h = std::thread::spawn(move || {
let _g = db2.lock_record("g:b");
acquired2.store(1, Ordering::SeqCst);
});
std::thread::sleep(SETTLE);
assert_eq!(
acquired.load(Ordering::SeqCst),
0,
"single-member write must block while ManyRecordWriteGuard is held"
);
drop(many);
h.join().unwrap();
assert_eq!(acquired.load(Ordering::SeqCst), 1);
}
#[test]
fn lock_records_overlapping_sets_no_deadlock() {
let db = PvDatabase::new();
let done = Arc::new(AtomicUsize::new(0));
let handles: Vec<_> = [["x", "y", "z"], ["z", "y", "x"]]
.into_iter()
.map(|set| {
let db = db.clone();
let done = done.clone();
std::thread::spawn(move || {
for _ in 0..500 {
let _g = db.lock_records(set);
std::thread::yield_now();
}
done.fetch_add(1, Ordering::SeqCst);
})
})
.collect();
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while done.load(Ordering::SeqCst) < 2 {
assert!(
std::time::Instant::now() < deadline,
"overlapping lock_records sets must not deadlock"
);
std::thread::sleep(Duration::from_millis(10));
}
for h in handles {
h.join().unwrap();
}
}
#[test]
fn overlapping_epochs_are_mutually_exclusive_and_deadlock_free() {
let db = PvDatabase::new();
let a = vec!["RECA".to_string(), "RECB".to_string()];
let b = vec!["RECB".to_string(), "RECC".to_string()];
let guard_a = db.lock_records(&a);
let db2 = db.clone();
let entered = Arc::new(AtomicUsize::new(0));
let entered2 = entered.clone();
let handle = std::thread::spawn(move || {
let _guard_b = db2.lock_records(&b);
entered2.store(1, Ordering::SeqCst);
});
std::thread::sleep(SETTLE);
assert_eq!(
entered.load(Ordering::SeqCst),
0,
"epoch B must block on shared RECB"
);
drop(guard_a);
handle.join().expect("epoch B thread");
assert_eq!(entered.load(Ordering::SeqCst), 1);
}
#[test]
fn disjoint_epochs_do_not_block_each_other() {
let db = PvDatabase::new();
let _g1 = db.lock_records(&["X1".to_string()]);
let _g2 = db.lock_records(&["X2".to_string()]);
}
#[test]
fn a_name_reaches_one_set_and_two_unlinked_records_reach_two() {
let db = PvDatabase::new();
assert!(
std::ptr::eq(db.ensure_set("REC:A").set(), db.ensure_set("REC:A").set()),
"the same canonical name must map to the same lock set"
);
assert!(
!std::ptr::eq(db.ensure_set("REC:A").set(), db.ensure_set("REC:B").set()),
"records no link joins must not share a lock set"
);
}
#[test]
fn a_second_overlapping_epoch_on_one_thread_recurses() {
let db = PvDatabase::new();
let _epoch = db.lock_records(&["RE:A".to_string(), "RE:B".to_string()]);
let _overlapping = db.lock_records(&["RE:B".to_string(), "RE:C".to_string()]);
}
#[test]
fn re_taking_one_record_s_set_recurses_like_db_scan_lock() {
let db = PvDatabase::new();
let _held = db.lock_record("RE:SELF");
let _again = db.lock_record("RE:SELF");
}
#[test]
fn recursion_does_not_let_a_second_thread_in() {
let db = PvDatabase::new();
let outer = db.lock_record("RE:DEPTH");
let inner = db.lock_record("RE:DEPTH");
let db2 = db.clone();
let entered = Arc::new(AtomicUsize::new(0));
let entered2 = entered.clone();
let h = std::thread::spawn(move || {
let _g = db2.lock_record("RE:DEPTH");
entered2.store(1, Ordering::SeqCst);
});
std::thread::sleep(SETTLE);
assert_eq!(entered.load(Ordering::SeqCst), 0, "outer level still held");
drop(inner);
std::thread::sleep(SETTLE);
assert_eq!(
entered.load(Ordering::SeqCst),
0,
"one release of two must not hand the set over"
);
drop(outer);
h.join().unwrap();
assert_eq!(entered.load(Ordering::SeqCst), 1);
}
#[test]
fn this_file_creates_no_mutex_but_lock_sets() {
let db = PvDatabase::new();
for name in ["MS:1", "MS:2", "MS:3"] {
drop(db.lock_record(name));
}
let made = *SET_MUTEX_SEQ.lock().unwrap();
assert_eq!(
lock_set_mutex_rows().len() as u64,
made,
"a second mutex created in this file would shift every set's row"
);
for set in db.lock_set_report().active {
assert!(set.mutex.is_some(), "set {} has no row", set.id);
}
}
#[test]
fn read_in_rides_the_held_set_and_takes_another() {
use crate::server::record::{RecordCell, RecordInstance};
use crate::server::records::calc::CalcRecord;
let db = PvDatabase::new();
let cell = |name: &str| {
let cell = Arc::new(RecordCell::new(RecordInstance::new(
name.into(),
CalcRecord::default(),
)));
db.inner.record_locks.adopt(name, cell.lock_record());
db.ensure_set_for(name, cell.lock_record());
cell
};
let reader = cell("RI:A");
let linked = cell("RI:B");
let apart = cell("RI:C");
db.merge_sets("RI:A", "RI:B");
let mut held = reader.write();
let (_, set) = held.split();
assert!(
linked.read_in(set).rides_held_set(),
"a record of the held set must be read without a second hold"
);
assert!(
!apart.read_in(set).rides_held_set(),
"a record of another set must take that set"
);
}
#[test]
fn the_set_is_released_when_the_guard_drops() {
let db = PvDatabase::new();
drop(db.lock_record("RE:SEQ"));
drop(db.lock_record("RE:SEQ"));
drop(db.lock_records(&["RE:SEQ".to_string()]));
let _a = db.lock_record("RE:ONE");
let _b = db.lock_record("RE:TWO");
}
fn adopted_cell(db: &PvDatabase, name: &str) -> Arc<crate::server::record::RecordCell> {
use crate::server::record::{RecordCell, RecordInstance};
use crate::server::records::calc::CalcRecord;
let cell = Arc::new(RecordCell::new(RecordInstance::new(
name.into(),
CalcRecord::default(),
)));
db.inner.record_locks.adopt(name, cell.lock_record());
cell
}
#[test]
fn adopt_before_build_leaves_the_record_on_the_bootstrap_set() {
let db = PvDatabase::new();
let cell = adopted_cell(&db, "AD:PRE");
assert!(cell.lock_record().is_bootstrap());
assert_eq!(db.inner.record_locks.set_id_of("AD:PRE"), None);
}
#[test]
fn adopt_after_build_mints_the_record_s_set() {
let db = PvDatabase::new();
db.build_lock_sets();
let cell = adopted_cell(&db, "AD:POST");
assert!(!cell.lock_record().is_bootstrap());
assert_eq!(
db.inner.record_locks.set_id_of("AD:POST"),
Some(cell.lock_record().set().id)
);
}
#[test]
fn build_lock_sets_mints_every_adopted_cell() {
let db = PvDatabase::new();
let cell = adopted_cell(&db, "AD:STRAGGLER");
assert!(cell.lock_record().is_bootstrap());
db.build_lock_sets();
assert!(!cell.lock_record().is_bootstrap());
assert!(db.inner.record_locks.set_id_of("AD:STRAGGLER").is_some());
}
#[cfg(debug_assertions)]
#[test]
#[should_panic(expected = "read guard on the records map")]
fn a_fresh_set_under_the_map_read_guard_is_refused() {
let db = PvDatabase::new();
let _map = db.inner.records.read();
let _g = db.lock_record("MR:FRESH");
}
#[cfg(debug_assertions)]
#[test]
#[should_panic(expected = "read guard on the records map")]
fn a_fresh_set_under_the_alias_read_guard_is_refused() {
let db = PvDatabase::new();
let _map = db.inner.aliases.read();
let _g = db.lock_record("MR:ALIAS");
}
#[test]
fn re_entry_under_the_map_read_guard_and_a_fresh_set_after_it_are_fine() {
let db = PvDatabase::new();
let _held = db.lock_record("MR:HELD");
{
let _map = db.inner.records.read();
drop(db.lock_record("MR:HELD"));
}
drop(db.lock_record("MR:OTHER"));
}
#[test]
fn every_set_after_build_excludes_the_bootstrap_set() {
let db = PvDatabase::new();
adopted_cell(&db, "AD:ES1");
db.build_lock_sets();
adopted_cell(&db, "AD:ES2");
let registry = db.inner.record_locks.lock();
let sets = registry.every_set();
assert_eq!(sets.len(), 2);
assert!(
sets.iter().all(|set| !std::ptr::eq(*set, bootstrap_set())),
"every_set() must not hand a relink the bootstrap set"
);
}
}