diff --git a/Cargo.toml b/Cargo.toml index a8ff61e..b9fc07e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,7 +76,7 @@ tokio-websockets = { version = "0.13.2", features = ["client", "server", "rustls multibase = "0.9.2" # asm gates the armv8 sha2 instructions; without it aarch64 hashes in software sha2 = { version = "0.10.9", features = ["asm"] } -parking_lot = "0.12.5" +parking_lot = { version = "0.12.5", features = ["arc_lock"] } hyper = "1.8.1" http = "1" nohash-hasher = "0.2.0" diff --git a/src/control/repos/indexer.rs b/src/control/repos/indexer.rs index c24cb09..03c6496 100644 --- a/src/control/repos/indexer.rs +++ b/src/control/repos/indexer.rs @@ -89,10 +89,7 @@ impl ReposControl { .db .run(move |db| { let mut txn = crate::db::Txn::new(db); - txn.hold_record_lock_indexes_sorted( - dids.iter() - .map(|did| crate::db::record_lock_index_for_did(did)), - ); + txn.hold_repo_write_locks_sorted(&dids); let mut queued: Vec = Vec::new(); for did in dids { @@ -128,10 +125,7 @@ impl ReposControl { .db .run(move |db| { let mut txn = crate::db::Txn::new(db); - txn.hold_record_lock_indexes_sorted( - dids.iter() - .map(|did| crate::db::record_lock_index_for_did(did)), - ); + txn.hold_repo_write_locks_sorted(&dids); let mut queued: Vec = Vec::new(); for did in dids { @@ -199,9 +193,7 @@ impl ReposControl { .run(move |db| { let mut txn = crate::db::Txn::new(db); // a backfill checks its queue entry under these before it commits - txn.hold_record_lock_indexes_sorted( - dids.iter().map(crate::db::record_lock_index_for_did), - ); + txn.hold_repo_write_locks_sorted(&dids); let mut untracked: Vec = Vec::new(); for did in dids { diff --git a/src/crawler/worker.rs b/src/crawler/worker.rs index d9a60c4..d954858 100644 --- a/src/crawler/worker.rs +++ b/src/crawler/worker.rs @@ -182,29 +182,31 @@ impl CrawlerWorker { return Ok(()); } - // Group by the one record lock each DID needs. Acquiring a whole crawler - // page at once usually captures nearly all 256 locks and turns one long - // operator redaction into a global ingestion convoy. + // commit the page as a few hundred small txns instead of one. a txn keeps every + // repo it has locked while it waits on the next, and a backfill or an operator + // redaction can hold one repo for seconds, which would stall live commits for + // the whole page. let app_state = self.state.clone(); let surviving = app_state .db .run(move |db| { let mut rng: SmallRng = rand::make_rng(); + let group_of = |did: &Did| crate::util::hash(did) as u8; let mut listing_groups = std::collections::BTreeMap::>::new(); for listing in listings { listing_groups - .entry(crate::db::record_lock_index_for_did(&listing.did)) + .entry(group_of(&listing.did)) .or_default() .push(listing); } let mut guard_groups = std::collections::BTreeMap::>::new(); for guard in guards { guard_groups - .entry(crate::db::record_lock_index_for_did(&guard)) + .entry(group_of(&guard)) .or_default() .push(guard); } - let lock_indices = listing_groups + let groups = listing_groups .keys() .chain(guard_groups.keys()) .copied() @@ -212,15 +214,22 @@ impl CrawlerWorker { let mut surviving = Vec::new(); let mut changed_listings = Vec::new(); - for lock_index in lock_indices { + for group in groups { + let listings = listing_groups.remove(&group).unwrap_or_default(); + let guards = guard_groups.remove(&group).unwrap_or_default(); let mut txn = crate::db::Txn::new(db); - txn.hold_record_lock_index(lock_index); - for listing in listing_groups.remove(&lock_index).unwrap_or_default() { + txn.hold_repo_write_locks_sorted( + listings + .iter() + .map(|listing| &listing.did) + .chain(guards.iter().map(|guard| &guard.did)), + ); + for listing in listings { if reconcile_listing(&mut txn, &listing)?.emits_account() { changed_listings.push(listing); } } - for guard in guard_groups.remove(&lock_index).unwrap_or_default() { + for guard in guards { if db .filter .contains_key(crate::db::filter::exclude_key(guard.as_str())?) @@ -341,8 +350,9 @@ pub(crate) fn reconcile_listing( txn: &mut crate::db::Txn<'_>, listing: &RepoListing, ) -> Result { - // production callers pre-acquire this listing's one lock group; this local - // check keeps direct callers on the same record-lock -> lifecycle hierarchy. + // the crawler already holds this repo's lock, taken in order with the rest of its + // group. this local check keeps direct callers on the same record-lock -> lifecycle + // hierarchy. let db = txn.db; if txn.lock_repo_and_is_excluded(&listing.did)? { return Ok(ReconcileOutcome::Unchanged); diff --git a/src/db/indexer.rs b/src/db/indexer.rs index 27860d7..fbdb8f8 100644 --- a/src/db/indexer.rs +++ b/src/db/indexer.rs @@ -212,10 +212,7 @@ pub(crate) fn redact_record_bodies( rkey: &crate::db::types::DbRkey, target: DeleteBodyTarget, ) -> Result { - let lock_index = crate::db::record_lock_index_for_did(did); - let _guard = db.record_write_locks[lock_index as usize] - .lock() - .unwrap_or_else(|e| e.into_inner()); + let _guard = db.repo_write_locks.lock(&keys::repo_key(did)); let record_key = keys::record_key(did, collection, rkey); let mut report = DeleteBodiesReport::default(); diff --git a/src/db/mod.rs b/src/db/mod.rs index d62d9e6..8fcafb0 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -34,6 +34,8 @@ pub mod keyspaces; mod open; pub(crate) mod outbox; pub mod registry; +#[cfg(feature = "indexer")] +pub(crate) mod repo_locks; pub mod schema; #[cfg(any(feature = "indexer_stream", feature = "relay"))] pub(crate) mod sequencer; @@ -88,29 +90,15 @@ pub struct Db { pub(crate) poison: Poison, #[cfg(test)] pub(crate) persist_failures: Arc, - /// 256 lock-sharded mutexes keyed by one byte of the trimmed DID. this is - /// synchronization only, not storage partitioning: unrelated DIDs may - /// conservatively share a lock. record mutations hold the selected lock - /// from read through commit so operator redaction cannot race them. + /// one lock per repo, keyed by its trimmed did. record mutations hold it from read + /// through commit so operator redaction can't race them. #[cfg(feature = "indexer")] - pub(crate) record_write_locks: Arc<[std::sync::Mutex<()>; 256]>, + pub(crate) repo_write_locks: Arc, /// the only source of stream positions, see [`sequencer`] #[cfg(any(feature = "indexer_stream", feature = "relay"))] pub(crate) sequencer: Arc, } -/// lock index for a DID. skip the method tag so all PLC DIDs do not collapse -/// onto one mutex. -#[cfg(feature = "indexer")] -pub(crate) fn record_lock_index_for_did(did: &jacquard_common::types::did::Did) -> u8 { - let prefix = keys::record_prefix_did(did); - match prefix.first() { - Some(&types::TAG_PLC) | Some(&types::TAG_WEB) => prefix.get(1).copied().unwrap_or(0), - Some(&b) => b, - None => 0, - } -} - /// [`Db::run`] work that never ran, because its runtime shut down first #[derive(Debug, miette::Diagnostic, thiserror::Error)] #[error("the runtime shut down before this database work ran")] diff --git a/src/db/open.rs b/src/db/open.rs index a49a4ef..eaba356 100644 --- a/src/db/open.rs +++ b/src/db/open.rs @@ -202,7 +202,7 @@ impl Db { #[cfg(test)] persist_failures: Arc::new(AtomicUsize::new(0)), #[cfg(feature = "indexer")] - record_write_locks: Arc::new(std::array::from_fn(|_| std::sync::Mutex::new(()))), + repo_write_locks: Arc::default(), #[cfg(any(feature = "indexer_stream", feature = "relay"))] sequencer, }) diff --git a/src/db/repo_locks.rs b/src/db/repo_locks.rs new file mode 100644 index 0000000..c06fac2 --- /dev/null +++ b/src/db/repo_locks.rs @@ -0,0 +1,179 @@ +//! one write lock per repository. a lock only exists while something holds or waits on +//! it, so the table stays as small as the set of repos being written right now. + +use std::collections::HashMap; +use std::sync::Arc; + +use parking_lot::{ArcMutexGuard, Mutex, RawMutex}; + +/// the table is split so writers to unrelated repos don't queue on one map +const SHARDS: usize = 64; +/// a shard at or under this many slots is left alone, so small ones don't keep reallocating +const MIN_CAPACITY: usize = 64; + +type Shard = Mutex, Arc>>>; + +pub(crate) struct RepoLocks { + shards: Box<[Shard]>, +} + +/// holds one repo's lock until dropped +pub(crate) struct RepoGuard<'a> { + shard: &'a Shard, + key: Box<[u8]>, + guard: Option>, +} + +impl Default for RepoLocks { + fn default() -> Self { + Self { + shards: (0..SHARDS).map(|_| Shard::default()).collect(), + } + } +} + +impl RepoLocks { + /// waits until nothing else holds `key`'s lock + pub(crate) fn lock(&self, key: &[u8]) -> RepoGuard<'_> { + let shard = &self.shards[crate::util::hash(&key) as usize % SHARDS]; + let key: Box<[u8]> = key.into(); + // wait outside the shard lock, so a busy repo only stalls its own writers + let lock = shard.lock().entry(key.clone()).or_default().clone(); + RepoGuard { + shard, + key, + guard: Some(lock.lock_arc()), + } + } + + #[cfg(test)] + fn len(&self) -> usize { + self.shards.iter().map(|shard| shard.lock().len()).sum() + } + + #[cfg(test)] + fn capacity(&self) -> usize { + self.shards + .iter() + .map(|shard| shard.lock().capacity()) + .sum() + } +} + +impl Drop for RepoGuard<'_> { + fn drop(&mut self) { + let Some(guard) = self.guard.take() else { + return; + }; + let mut shard = self.shard.lock(); + let lock = ArcMutexGuard::into_arc(guard); + // clones are only made under the shard lock, so if the map has the only other one + // nobody is waiting and the entry can go + if Arc::strong_count(&lock) == 2 { + shard.remove(&self.key); + // one txn can lock a whole list of repos, and a map keeps its peak size, so give + // the room back once the shard is mostly empty. shrinking to twice the len leaves + // len / 2 removals before the next shrink, which keeps it amortized + let len = shard.len(); + if shard.capacity() > MIN_CAPACITY && len < shard.capacity() / 4 { + shard.shrink_to(len * 2); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::mpsc; + use std::time::Duration; + + // the holders and channels below live inside each scope, so a failed assert drops them + // while unwinding and the other thread gets out instead of hanging the test + + #[test] + fn one_repo_waits_for_its_holder() { + let locks = RepoLocks::default(); + let entered = AtomicBool::new(false); + std::thread::scope(|s| { + let held = locks.lock(b"a"); + let waiter = s.spawn(|| { + let _guard = locks.lock(b"a"); + entered.store(true, Ordering::SeqCst); + }); + std::thread::sleep(Duration::from_millis(50)); + assert!(!entered.load(Ordering::SeqCst)); + drop(held); + waiter.join().unwrap(); + }); + assert!(entered.load(Ordering::SeqCst)); + } + + /// runs `f` on a thread nobody joins, so a lock that's wrongly shared fails the test + /// instead of hanging it + fn within_10s(f: impl FnOnce() -> T + Send + 'static) -> T { + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || tx.send(f())); + rx.recv_timeout(Duration::from_secs(10)) + .expect("deadlocked or panicked") + } + + #[test] + fn other_repos_dont_wait() { + let (held, after) = within_10s(|| { + let locks = RepoLocks::default(); + // 1000 keys over 64 shards, so plenty share a shard but none share a lock + let guards: Vec<_> = (0..1000u32).map(|i| locks.lock(&i.to_be_bytes())).collect(); + let held = locks.len(); + drop(guards); + (held, locks.len()) + }); + assert_eq!((held, after), (1000, 0)); + } + + #[test] + fn a_waiter_keeps_the_entry_until_it_is_done() { + let locks = RepoLocks::default(); + std::thread::scope(|s| { + let held = locks.lock(b"a"); + let (entered_tx, entered) = mpsc::channel(); + let (release, release_rx) = mpsc::channel::<()>(); + let locks = &locks; + let waiter = s.spawn(move || { + let _guard = locks.lock(b"a"); + entered_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + // the waiter has its clone once the count passes the map's and `held`'s + while Arc::strong_count(held.guard.as_ref().map(ArcMutexGuard::mutex).unwrap()) < 3 { + assert!( + entered.try_recv().is_err(), + "got in while `held` held the lock" + ); + std::thread::yield_now(); + } + drop(held); + entered.recv().unwrap(); + assert_eq!(locks.len(), 1); + release.send(()).unwrap(); + waiter.join().unwrap(); + }); + assert_eq!(locks.len(), 0); + } + + #[test] + fn a_big_lock_set_gives_its_room_back() { + let (peak, kept) = within_10s(|| { + let locks = RepoLocks::default(); + let guards: Vec<_> = (0..100_000u32) + .map(|i| locks.lock(&i.to_be_bytes())) + .collect(); + let peak = locks.capacity(); + drop(guards); + (peak, locks.capacity()) + }); + assert!(peak >= 100_000); + assert!(kept <= SHARDS * MIN_CAPACITY, "{kept} of {peak} slots kept"); + } +} diff --git a/src/db/txn.rs b/src/db/txn.rs index 7dc5612..c8a4d03 100644 --- a/src/db/txn.rs +++ b/src/db/txn.rs @@ -64,13 +64,10 @@ pub(crate) struct Txn<'db> { pds_accounts: HashMap, Option>, #[cfg(feature = "indexer")] lifecycle: Option>, - /// record-write guards retained from the first record/purge touch until - /// this transaction drops after commit. + /// repo locks held from the first record/purge touch until this transaction drops + /// after commit, by repo key so taking one again can't self-deadlock #[cfg(feature = "indexer")] - record_lock_guards: Vec>, - /// lock indexes already held, so a repeat acquisition cannot self-deadlock. - #[cfg(feature = "indexer")] - held_record_locks: std::collections::BTreeSet, + repo_locks: HashMap, super::repo_locks::RepoGuard<'db>>, } impl<'db> Txn<'db> { @@ -93,34 +90,27 @@ impl<'db> Txn<'db> { #[cfg(feature = "indexer")] lifecycle: None, #[cfg(feature = "indexer")] - record_lock_guards: Vec::new(), - #[cfg(feature = "indexer")] - held_record_locks: std::collections::BTreeSet::new(), + repo_locks: HashMap::new(), } } - /// hold one record-write lock until this transaction drops. recover a - /// poisoned mutex so a panicking writer cannot wedge the shard forever. + /// hold one repository's write barrier until this transaction drops. #[cfg(feature = "indexer")] - pub(crate) fn hold_record_lock_index(&mut self, lock_index: u8) { - if self.held_record_locks.contains(&lock_index) { + pub(crate) fn hold_repo_write_lock(&mut self, did: &Did) { + self.hold_repo_key(keys::repo_key(did)); + } + + #[cfg(feature = "indexer")] + fn hold_repo_key(&mut self, key: Vec) { + if self.repo_locks.contains_key(&key) { return; } assert!( self.lifecycle.is_none(), "record locks must be acquired before lifecycle counts" ); - self.held_record_locks.insert(lock_index); - let guard = self.db.record_write_locks[lock_index as usize] - .lock() - .unwrap_or_else(|e| e.into_inner()); - self.record_lock_guards.push(guard); - } - - /// hold one repository's write barrier until this transaction drops. - #[cfg(feature = "indexer")] - pub(crate) fn hold_repo_write_lock(&mut self, did: &Did) { - self.hold_record_lock_index(crate::db::record_lock_index_for_did(did)); + let guard = self.db.repo_write_locks.lock(&key); + self.repo_locks.insert(key, guard); } /// lock this repository, then read its durable exclusion status inside the @@ -159,20 +149,23 @@ impl<'db> Txn<'db> { } /// acquire a transaction's complete record-lock set in one canonical order. - /// callers that can touch several DIDs must do this before initializing - /// lifecycle counts; dynamic encounter-order locking can deadlock against - /// another multi-repository operation. + /// callers that can touch several DIDs must do this before taking any other + /// repo lock or initializing lifecycle counts; dynamic encounter-order locking + /// can deadlock against another multi-repository operation. #[cfg(feature = "indexer")] - pub(crate) fn hold_record_lock_indexes_sorted(&mut self, locks: impl IntoIterator) { + pub(crate) fn hold_repo_write_locks_sorted<'a>( + &mut self, + dids: impl IntoIterator, + ) { assert!( - self.lifecycle.is_none(), - "record locks must be acquired before lifecycle counts" - ); - let mut locks: Vec<_> = locks.into_iter().collect(); - locks.sort_unstable(); - locks.dedup(); - for lock in locks { - self.hold_record_lock_index(lock); + self.repo_locks.is_empty() && self.lifecycle.is_none(), + "a multi-repo lock set must be the transaction's first lock" + ); + let mut repo_keys: Vec<_> = dids.into_iter().map(keys::repo_key).collect(); + repo_keys.sort_unstable(); + repo_keys.dedup(); + for key in repo_keys { + self.hold_repo_key(key); } } @@ -1714,4 +1707,29 @@ mod tests { "history entry death rev must sort strictly after future root rev so event replay resolves it" ); } + + #[test] + fn repos_that_shared_a_stripe_dont_wait_on_each_other() { + let (_tmp, state) = test_state(); + let a = did(); + let b = Did::new_static("did:plc:ewaaaaaaaaaaaaaaaaaaaaaa").unwrap(); + // the same tag and first byte, which put them on one of the old 256 stripes + assert_eq!(keys::repo_key(&a)[..2], keys::repo_key(&b)[..2]); + + let mut held = Txn::new(&state.db); + held.hold_repo_write_lock(&a); + let (done_tx, done) = std::sync::mpsc::channel(); + std::thread::scope(|s| { + let (db, b) = (&state.db, &b); + s.spawn(move || { + let mut txn = Txn::new(db); + txn.hold_repo_write_lock(b); + done_tx.send(()).unwrap(); + }); + let other_went = done.recv_timeout(Duration::from_secs(10)).is_ok(); + // let the other thread finish either way, so a failure doesn't hang the scope + drop(held); + assert!(other_went); + }); + } }