From ccac1f2c125f195bf7f1f7b846c42f1533bcf2de Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 21:26:27 -0600 Subject: [PATCH 1/9] Register directory reader pins --- README.md | 17 ++-- benches/kv_directory.rs | 57 +++++++++++ docs/reclamation-plan.md | 42 ++++---- src/phase0.rs | 202 ++++++++++++++++++++++++++++++++++++++- 4 files changed, 291 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index 5e78487..fc1a47f 100644 --- a/README.md +++ b/README.md @@ -98,14 +98,15 @@ upserts include delete terms. CI runs only the correctness smoke profile; timing require controlled hardware. The `kv-directory` benchmark measures the caller-visible buffered write path, empty and dirty -flushes, 256 KiB chunk publication, closed- and active-writer deletion, and distinct-file -concurrency at one, two, four, and eight threads. It reports percentiles across per-sample mean -latencies and uses the deterministic in-memory Phase 0 store to isolate directory coordination from -RocksDB and Node transport costs. That store serializes access, so the concurrency cases detect -coordination regressions but do not predict RocksDB scaling. Compare two optimized builds on the -same quiet host; records include the Git revision and dirty state, while `--revision` can add a run -label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are -not comparable to a full run, and applies no timing threshold. +flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read +handle opens, and distinct-file concurrency at one, two, four, and eight threads. It reports +percentiles across per-sample mean latencies and uses the deterministic in-memory Phase 0 store to +isolate directory coordination from RocksDB and Node transport costs. That store serializes access, +so the concurrency cases detect coordination regressions but do not predict RocksDB scaling. +Compare two optimized builds on the same quiet host; records include the Git revision and dirty +state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI +executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing +threshold. Generated Node-API declarations in `ts/addon.d.ts` are private implementation types. Consumers use only the types exported from a package entry point. diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index 82faac5..7df4be3 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -19,6 +19,7 @@ const MIN_WRITE_OPERATIONS_PER_SAMPLE: usize = 256; const CHUNK_BYTES: usize = 256 * 1024; const STORAGE_OPERATIONS_PER_SAMPLE: usize = 64; const EMPTY_FLUSHES_PER_SAMPLE: usize = 10_000; +const OPEN_READS_PER_SAMPLE: usize = 10_000; const CONCURRENT_BYTES_PER_FILE: usize = 4 * 1024; const CONCURRENT_FILES_PER_THREAD: usize = 64; @@ -185,6 +186,20 @@ fn main() -> io::Result<()> { &arguments, delete_case(true, arguments.smoke), )?); + results.push(measure_case( + "open-read-retained".to_owned(), + "open-read", + 1, + &arguments, + open_read_case(true, arguments.smoke), + )?); + results.push(measure_case( + "open-read-churn".to_owned(), + "open-read", + 1, + &arguments, + open_read_case(false, arguments.smoke), + )?); for threads in [1, 2, 4, 8] { results.push(measure_case( @@ -456,6 +471,48 @@ fn delete_case(active_writer: bool, smoke: bool) -> impl FnMut(usize) -> io::Res } } +fn open_read_case(retain_handles: bool, smoke: bool) -> impl FnMut(usize) -> io::Result { + move |sample| { + let directory = FaultingDirectory::new(FaultingKv::default()); + let operations = if smoke { + 4 + } else if retain_handles { + STORAGE_OPERATIONS_PER_SAMPLE + } else { + OPEN_READS_PER_SAMPLE + }; + let file_count = if retain_handles { operations } else { 1 }; + let mut paths = Vec::with_capacity(file_count); + for file in 0..file_count { + let path = format!("open-read-{retain_handles}-{sample}-{file}"); + let mut writer = open_writer(&directory, Path::new(&path))?; + writer.write_all(b"contents")?; + writer.terminate()?; + paths.push(path); + } + let mut handles = Vec::with_capacity(if retain_handles { operations } else { 1 }); + let started = Instant::now(); + for operation in 0..operations { + let path = &paths[operation % file_count]; + let handle = directory + .open_read(Path::new(path)) + .map_err(|error| io::Error::other(error.to_string()))?; + if retain_handles { + handles.push(handle); + } else { + black_box(handle); + } + } + let elapsed_nanoseconds = started.elapsed().as_nanos(); + black_box(handles); + Ok(Sample { + elapsed_nanoseconds, + operations: operations as u64, + bytes: 0, + }) + } +} + fn concurrent_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Result { let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD }; move |sample| { diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 5397af9..f8a4cd0 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -9,13 +9,13 @@ readiness wait for a complete sweep. ## Grounding -This plan is written against fulltext main at `a982e8a`, the rebase-merged result of -[Track directory object high-water marks #25](https://github.com/HarperFast/fulltext/pull/25). -`KvDirectory` stores immutable 256 KiB chunks and revisioned tails. `delete()` currently removes -only logical bindings, each successful `flush()` can leave the previous tail revision unreachable, -and a crashed writer can leave staged chunks that no binding ever named. `KvStore` supplies point -read, atomic batch write, and sync, but no key enumeration. Harper PR #2535 supplies the same narrow -storage shape from Harper-owned RocksDB. +This plan is written against fulltext main at `36d96e3`, the rebase-merged result of +[Enqueue deleted directory objects for reclamation #26](https://github.com/HarperFast/fulltext/pull/26). +`KvDirectory` stores immutable 256 KiB chunks and revisioned tails. `delete()` atomically removes a +logical binding and enqueues the object's derivable physical extent, but no consumer deletes those +payloads yet. Each successful `flush()` can leave the previous tail revision unreachable. `KvStore` +supplies point read, atomic batch write, and sync, but no key enumeration. Harper PR #2535 supplies +the same narrow storage shape from Harper-owned RocksDB. Tantivy's `ManagedDirectory` decides when a logical file name is retired. `KvDirectory` owns the physical object behind that name and the lifetime of opened handles. Harper's derived-index runtime @@ -118,8 +118,11 @@ object. A handle owns an `Arc` pin, and drop only decrements the `Arc`. Pin inse entries above a small fixed threshold, and FIFO visits prune them again. This bounds control-block retention during repeated open/drop churn. Writer state is likewise retained through failed and unterminated writes. Gates recover poisoned state rather than panicking the writer actor. Pins land -with the FIFO consumer in slice 4, where a test can prove that they prevent payload deletion. Harper -may enable cleanup only after a two-worker test proves that every local search handle for one +in the first slice 4 unit before the FIFO consumer. A handle reads its binding and registers the pin +under the existing per-path lifecycle gate. Deletion uses the same gate, so it either observes a +registered pin or removes the binding before a later open can read it. This preserves one storage +read per handle open and adds no cross-path coordination. The consumer follows only after these +lifetime rules are independently covered. Harper may enable cleanup only after a two-worker test proves that every local search handle for one RocksDB-backed generation reaches the same native `DirectoryState`. A process that cannot establish that invariant cannot obtain the cleanup owner lease. @@ -248,11 +251,13 @@ Then deliver reclamation in reviewable slices: 3. per-path lifecycle state, a strided object-id allocator, an atomic writer fence, and sharded-FIFO enqueue at object deletion; key writer retirement by object id so path reuse cannot retire the replacement writer; -4. measure and, if justified, add tail-supersession entries; then add object-id/revision pins, - bounded FIFO draining and deferred queues, low-priority admission, failure observability, close - fencing, crash recovery, and Harper's exclusive-owner integration. +4. add object-id/revision reader pins and coordinate registration with deletion through the + per-path lifecycle gate; +5. measure and, if justified, add tail-supersession entries; then add bounded FIFO draining and + deferred queues, low-priority admission, failure observability, close fencing, crash recovery, and + Harper's exclusive-owner integration. -The production Harper package remains disabled until slice 4's lifecycle fence and Harper host- +The production Harper package remains disabled until the cleanup lifecycle fence and Harper host- storage integration both pass. No native RocksDB storage provider is added to the public library. ## Verification @@ -279,12 +284,13 @@ real Tantivy merge and garbage-collection cycles and after crash/reopen. `Faulti recovery at atomic batch boundaries; the Harper integration separately proves that one host write request commits as one RocksDB batch. -The dependency-free `kv_directory` release benchmark compares the merged slice-2 baseline with this -slice through Tantivy's buffered `WritePtr`, rather than calling the fence directly. It reports -per-sample-mean p50/p95/p99 and aggregate throughput for caller write sizes, empty and dirty flushes, chunk -publication, deletion with closed and active writers, and distinct-file concurrency. The +The dependency-free `kv_directory` release benchmark compares adjacent merged slices through +Tantivy's public directory interfaces. It reports per-sample-mean p50/p95/p99 and aggregate +throughput for caller write sizes, empty and dirty flushes, chunk publication, deletion with closed +and active writers, retained and churned read-handle opens, and distinct-file concurrency. The empty-flush case isolates the per-call retirement-fence cost; the buffered cases show how Tantivy's -writer amortizes it in practice. Results are versioned JSON labeled by revision. Shared CI runs a +writer amortizes it in practice. Retained handles measure registration growth, while churned handles +exercise weak-pin pruning. Results are versioned JSON labeled by revision. Shared CI runs a correctness smoke with no timing threshold; performance decisions use alternating runs on one fixed host. The deterministic Phase 0 store removes RocksDB and Node transport variance but serializes access, so its concurrency results detect directory-coordination regressions rather than predicting diff --git a/src/phase0.rs b/src/phase0.rs index c4a7619..b174e92 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -264,6 +264,7 @@ pub type FaultingDirectory = KvDirectory; struct DirectoryState { allocator: Mutex, paths: Arc, + reader_pins: ReaderPinRegistry, reclaim_shards: [Mutex<()>; RECLAIM_SHARD_COUNT], locks: Mutex, locks_changed: Condvar, @@ -304,6 +305,15 @@ struct WriterClaim<'a> { fence: &'a WriterFence, } +#[derive(Default)] +struct ReaderPinRegistry { + pins: Mutex>, +} + +type ObjectReaderPins = HashMap>>; + +struct ReaderPin; + #[derive(Default)] struct DirectoryLocks { held: HashSet, @@ -408,6 +418,52 @@ impl Drop for WriterClaim<'_> { } } +impl ReaderPinRegistry { + fn register(&self, binding: &Binding) -> Arc { + let pin = Arc::new(ReaderPin); + let mut pins = self.pins.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let revision_pins = pins + .entry(binding.object_id) + .or_default() + .entry(binding.tail_revision) + .or_default(); + if revision_pins.len() >= READER_PIN_PRUNE_THRESHOLD { + revision_pins.retain(|pin| pin.strong_count() != 0); + } + revision_pins.push(Arc::downgrade(&pin)); + pin + } + + #[cfg(test)] + fn is_pinned(&self, object_id: u64, tail_revision: Option) -> bool { + let mut pins = self.pins.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(revision_pins) = pins.get_mut(&object_id) else { + return false; + }; + let matched = if let Some(tail_revision) = tail_revision { + let Some(pins) = revision_pins.get_mut(&tail_revision) else { + return false; + }; + pins.retain(|pin| pin.strong_count() != 0); + let matched = !pins.is_empty(); + if !matched { + revision_pins.remove(&tail_revision); + } + matched + } else { + revision_pins.retain(|_, pins| { + pins.retain(|pin| pin.strong_count() != 0); + !pins.is_empty() + }); + !revision_pins.is_empty() + }; + if revision_pins.is_empty() { + pins.remove(&object_id); + } + matched + } +} + fn writer_retired_error() -> io::Error { io::Error::new(io::ErrorKind::NotFound, "file was deleted while its writer was open") } @@ -462,6 +518,7 @@ impl KvDirectory { let state = Arc::new(DirectoryState { allocator: Mutex::new(ObjectIdAllocator::default()), paths: Arc::new(PathRegistry::default()), + reader_pins: ReaderPinRegistry::default(), reclaim_shards: std::array::from_fn(|_| Mutex::new(())), locks: Mutex::new(DirectoryLocks::default()), locks_changed: Condvar::new(), @@ -569,6 +626,22 @@ impl KvDirectory { Mutation::Put(tail_key, next.to_be_bytes().to_vec()), ]) } + + fn pinned_binding( + &self, + path: &Path, + after_binding_read: impl FnOnce(), + ) -> Result<(Binding, Arc), OpenReadError> { + let path_state = self.state.paths.state(path); + let _lifecycle = path_state + .lifecycle + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let binding = self.read_binding(path)?; + after_binding_read(); + let pin = self.state.reader_pins.register(&binding); + Ok((binding, pin)) + } } struct KvFileHandle { @@ -576,6 +649,7 @@ struct KvFileHandle { namespace: Arc<[u8]>, path: PathBuf, binding: Binding, + _pin: Arc, } impl fmt::Debug for KvFileHandle { @@ -703,11 +777,13 @@ impl Drop for KvDirectoryLock { impl Directory for KvDirectory { fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { + let (binding, pin) = self.pinned_binding(path, || {})?; Ok(Arc::new(KvFileHandle { store: self.store.clone(), namespace: self.namespace.clone(), path: path.to_path_buf(), - binding: self.read_binding(path)?, + binding, + _pin: pin, })) } @@ -1289,6 +1365,7 @@ const MAX_OBJECT_EXTENT_BYTES: u128 = 1 << 40; const RECLAIM_ENTRY_FORMAT_VERSION: u8 = 1; const RECLAIM_ENTRY_WHOLE_OBJECT: u8 = 1; const RECLAIM_SHARD_COUNT: usize = 64; +const READER_PIN_PRUNE_THRESHOLD: usize = 64; const KEY_KIND_COUNTER: u8 = 1; const KEY_KIND_BINDING: u8 = 2; const KEY_KIND_ATOMIC: u8 = 3; @@ -2065,6 +2142,129 @@ mod tests { assert_eq!(second.read_bytes().unwrap().as_slice(), b"first second"); } + #[test] + fn open_handles_pin_their_object_and_tail_revision() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"first").unwrap(); + writer.flush().unwrap(); + let first_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let first = directory.open_read(path).unwrap(); + assert!(directory.state.reader_pins.is_pinned(first_binding.object_id, None)); + assert!(directory + .state + .reader_pins + .is_pinned(first_binding.object_id, Some(first_binding.tail_revision))); + + directory.delete(path).unwrap(); + let mut replacement = directory.open_write(path).unwrap(); + replacement.write_all(b"second").unwrap(); + replacement.flush().unwrap(); + let second_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let second = directory.open_read(path).unwrap(); + assert_ne!(first_binding.object_id, second_binding.object_id); + assert!(directory.state.reader_pins.is_pinned(first_binding.object_id, None)); + assert!(directory.state.reader_pins.is_pinned(second_binding.object_id, None)); + + drop(first); + assert!(!directory.state.reader_pins.is_pinned(first_binding.object_id, None)); + assert!(directory.state.reader_pins.is_pinned(second_binding.object_id, None)); + drop(second); + assert!(!directory.state.reader_pins.is_pinned(second_binding.object_id, None)); + } + + #[test] + fn binding_read_and_pin_registration_complete_before_delete() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.flush().unwrap(); + drop(writer); + let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let (binding_read, allow_delete) = std::sync::mpsc::sync_channel(0); + let (delete_started, deletion_entered) = std::sync::mpsc::sync_channel(0); + let deleting_directory = directory.clone(); + let deletion = std::thread::spawn(move || { + allow_delete.recv().unwrap(); + delete_started.send(()).unwrap(); + deleting_directory.delete(path) + }); + + let (opened, pin) = directory + .pinned_binding(path, || { + binding_read.send(()).unwrap(); + deletion_entered.recv().unwrap(); + }) + .unwrap(); + deletion.join().unwrap().unwrap(); + + assert_eq!(opened, binding); + assert!(directory.state.reader_pins.is_pinned(binding.object_id, None)); + assert!(!directory.exists(path).unwrap()); + drop(pin); + assert!(!directory.state.reader_pins.is_pinned(binding.object_id, None)); + } + + #[test] + fn independently_constructed_directories_share_reader_pins() { + let store = FaultingKv::default(); + let first = FaultingDirectory::new(store.clone()); + let second = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = first.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.flush().unwrap(); + let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let handle = first.open_read(path).unwrap(); + + assert!(second.state.reader_pins.is_pinned(binding.object_id, None)); + drop(handle); + assert!(!second.state.reader_pins.is_pinned(binding.object_id, None)); + } + + #[test] + fn open_read_adds_no_storage_operation_for_pin_registration() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.flush().unwrap(); + store.take_io_counts(); + + let handle = directory.open_read(path).unwrap(); + + assert_eq!(store.take_io_counts(), (1, 0, 0)); + drop(handle); + } + + #[test] + fn open_drop_churn_bounds_retained_weak_pins() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.flush().unwrap(); + let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + + for _ in 0..READER_PIN_PRUNE_THRESHOLD * 4 { + drop(directory.open_read(path).unwrap()); + } + + let pins = directory + .state + .reader_pins + .pins + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + assert!(pins[&binding.object_id][&binding.tail_revision].len() <= READER_PIN_PRUNE_THRESHOLD); + } + #[test] fn independently_constructed_directories_share_writer_locks() { let store = FaultingKv::default(); From bca77f47aa928cfe30436fa4b0d982e3168977c4 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 21:53:05 -0600 Subject: [PATCH 2/9] Bound reader pin lifecycle --- benches/kv_directory.rs | 3 +- docs/reclamation-plan.md | 27 ++-- src/phase0.rs | 272 +++++++++++++++++++++++++++++++-------- 3 files changed, 233 insertions(+), 69 deletions(-) diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index 7df4be3..ec172f9 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -19,6 +19,7 @@ const MIN_WRITE_OPERATIONS_PER_SAMPLE: usize = 256; const CHUNK_BYTES: usize = 256 * 1024; const STORAGE_OPERATIONS_PER_SAMPLE: usize = 64; const EMPTY_FLUSHES_PER_SAMPLE: usize = 10_000; +const RETAINED_OPEN_READS_PER_SAMPLE: usize = 1_024; const OPEN_READS_PER_SAMPLE: usize = 10_000; const CONCURRENT_BYTES_PER_FILE: usize = 4 * 1024; const CONCURRENT_FILES_PER_THREAD: usize = 64; @@ -477,7 +478,7 @@ fn open_read_case(retain_handles: bool, smoke: bool) -> impl FnMut(usize) -> io: let operations = if smoke { 4 } else if retain_handles { - STORAGE_OPERATIONS_PER_SAMPLE + RETAINED_OPEN_READS_PER_SAMPLE } else { OPEN_READS_PER_SAMPLE }; diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index f8a4cd0..7dec578 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -114,17 +114,20 @@ an unrelated timeout. Closing the host wakes pending operations and releases the gate is held by chunk staging or publication across a host round trip. Pins are indexed by object id, not logical path, so delete/recreate cannot hide handles to the prior -object. A handle owns an `Arc` pin, and drop only decrements the `Arc`. Pin insertion prunes dead weak -entries above a small fixed threshold, and FIFO visits prune them again. This bounds control-block -retention during repeated open/drop churn. Writer state is likewise retained through failed and -unterminated writes. Gates recover poisoned state rather than panicking the writer actor. Pins land -in the first slice 4 unit before the FIFO consumer. A handle reads its binding and registers the pin -under the existing per-path lifecycle gate. Deletion uses the same gate, so it either observes a -registered pin or removes the binding before a later open can read it. This preserves one storage -read per handle open and adds no cross-path coordination. The consumer follows only after these -lifetime rules are independently covered. Harper may enable cleanup only after a two-worker test proves that every local search handle for one -RocksDB-backed generation reaches the same native `DirectoryState`. A process that cannot establish -that invariant cannot obtain the cleanup owner lease. +object. Each object has counted pins by tail revision. A handle owns one pin and retains the shared +directory and path state; dropping it decrements the revision and object counts, and dropping the +last pin removes the object's weak registry entry. Registry shards use the same object-id partition +as reclamation, so unrelated objects do not share one pin mutex. This bounds retained control state +by live handles during repeated object, revision, and open/drop churn. Writer state is likewise +retained through failed and unterminated writes. Gates recover poisoned state rather than panicking +the writer actor. Pins land in the first slice 4 unit before the FIFO consumer. A handle reads its +binding and registers the pin under a shared per-path registration fence. Deletion takes the +exclusive side only for its final queue and binding transition, so concurrent opens remain parallel +and writer retirement does not block them. This preserves one storage read per handle open and adds +no cross-path coordination. The consumer follows only after these lifetime rules are independently +covered. Harper may enable cleanup only after a two-worker test proves that every local search +handle for one RocksDB-backed generation reaches the same native `DirectoryState`. A process that +cannot establish that invariant cannot obtain the cleanup owner lease. Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot gain a new pin because its binding is gone; a tail-only entry can gain no new pin for its old revision @@ -252,7 +255,7 @@ Then deliver reclamation in reviewable slices: enqueue at object deletion; key writer retirement by object id so path reuse cannot retire the replacement writer; 4. add object-id/revision reader pins and coordinate registration with deletion through the - per-path lifecycle gate; + per-path registration fence; 5. measure and, if justified, add tail-supersession entries; then add bounded FIFO draining and deferred queues, low-priority admission, failure observability, close fencing, crash recovery, and Harper's exclusive-owner integration. diff --git a/src/phase0.rs b/src/phase0.rs index b174e92..261bdfc 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -5,7 +5,7 @@ use std::io::Write; use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, AtomicUsize, Ordering}; -use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; +use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock, Weak}; use std::time::{Duration, Instant}; use async_trait::async_trait; @@ -264,7 +264,7 @@ pub type FaultingDirectory = KvDirectory; struct DirectoryState { allocator: Mutex, paths: Arc, - reader_pins: ReaderPinRegistry, + reader_pins: Arc, reclaim_shards: [Mutex<()>; RECLAIM_SHARD_COUNT], locks: Mutex, locks_changed: Condvar, @@ -286,6 +286,7 @@ struct PathState { path: PathBuf, registry: Weak, lifecycle: Mutex, + reader_registration: RwLock<()>, } #[derive(Default)] @@ -305,14 +306,34 @@ struct WriterClaim<'a> { fence: &'a WriterFence, } -#[derive(Default)] struct ReaderPinRegistry { - pins: Mutex>, + shards: [Mutex>>; RECLAIM_SHARD_COUNT], +} + +struct ObjectReaderPins { + object_id: u64, + registry: Weak, + state: Mutex, +} + +#[derive(Default)] +struct ObjectReaderPinState { + total: usize, + revisions: HashMap, } -type ObjectReaderPins = HashMap>>; +struct ReaderPin { + object: Arc, + tail_revision: u64, +} -struct ReaderPin; +impl Default for ReaderPinRegistry { + fn default() -> Self { + Self { + shards: std::array::from_fn(|_| Mutex::new(HashMap::new())), + } + } +} #[derive(Default)] struct DirectoryLocks { @@ -339,6 +360,7 @@ impl PathRegistry { path: path.to_path_buf(), registry: Arc::downgrade(self), lifecycle: Mutex::new(PathLifecycle::default()), + reader_registration: RwLock::new(()), }); states.insert(path.to_path_buf(), Arc::downgrade(&state)); state @@ -419,48 +441,102 @@ impl Drop for WriterClaim<'_> { } impl ReaderPinRegistry { - fn register(&self, binding: &Binding) -> Arc { - let pin = Arc::new(ReaderPin); - let mut pins = self.pins.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - let revision_pins = pins - .entry(binding.object_id) - .or_default() - .entry(binding.tail_revision) - .or_default(); - if revision_pins.len() >= READER_PIN_PRUNE_THRESHOLD { - revision_pins.retain(|pin| pin.strong_count() != 0); - } - revision_pins.push(Arc::downgrade(&pin)); - pin + fn register(self: &Arc, binding: &Binding) -> io::Result> { + let shard = usize::from(reclaim_shard(binding.object_id)); + let object = { + let mut objects = self.shards[shard] + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + objects + .get(&binding.object_id) + .and_then(Weak::upgrade) + .unwrap_or_else(|| { + let object = Arc::new(ObjectReaderPins { + object_id: binding.object_id, + registry: Arc::downgrade(self), + state: Mutex::new(ObjectReaderPinState::default()), + }); + objects.insert(binding.object_id, Arc::downgrade(&object)); + object + }) + }; + let mut state = object.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let total = state + .total + .checked_add(1) + .ok_or_else(|| io::Error::other("reader pin count exhausted"))?; + let revision = state + .revisions + .get(&binding.tail_revision) + .copied() + .unwrap_or(0) + .checked_add(1) + .ok_or_else(|| io::Error::other("reader revision pin count exhausted"))?; + state.total = total; + state.revisions.insert(binding.tail_revision, revision); + drop(state); + Ok(Arc::new(ReaderPin { + object, + tail_revision: binding.tail_revision, + })) } #[cfg(test)] fn is_pinned(&self, object_id: u64, tail_revision: Option) -> bool { - let mut pins = self.pins.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - let Some(revision_pins) = pins.get_mut(&object_id) else { + let shard = usize::from(reclaim_shard(object_id)); + let objects = self.shards[shard] + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(object) = objects.get(&object_id).and_then(Weak::upgrade) else { return false; }; - let matched = if let Some(tail_revision) = tail_revision { - let Some(pins) = revision_pins.get_mut(&tail_revision) else { - return false; - }; - pins.retain(|pin| pin.strong_count() != 0); - let matched = !pins.is_empty(); - if !matched { - revision_pins.remove(&tail_revision); + drop(objects); + let state = object.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + tail_revision.map_or(state.total != 0, |revision| { + state.revisions.get(&revision).is_some_and(|count| *count != 0) + }) + } +} + +impl Drop for ReaderPin { + fn drop(&mut self) { + let mut state = self + .object + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.total == 0 { + return; + } + let remove_revision = match state.revisions.get_mut(&self.tail_revision) { + Some(revision) if *revision != 0 => { + *revision -= 1; + *revision == 0 } - matched - } else { - revision_pins.retain(|_, pins| { - pins.retain(|pin| pin.strong_count() != 0); - !pins.is_empty() - }); - !revision_pins.is_empty() + _ => return, + }; + state.total -= 1; + if remove_revision { + state.revisions.remove(&self.tail_revision); + } + } +} + +impl Drop for ObjectReaderPins { + fn drop(&mut self) { + let Some(registry) = self.registry.upgrade() else { + return; }; - if revision_pins.is_empty() { - pins.remove(&object_id); + let shard = usize::from(reclaim_shard(self.object_id)); + let mut objects = registry.shards[shard] + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if objects + .get(&self.object_id) + .is_some_and(|object| std::ptr::eq(object.as_ptr(), self)) + { + objects.remove(&self.object_id); } - matched } } @@ -518,7 +594,7 @@ impl KvDirectory { let state = Arc::new(DirectoryState { allocator: Mutex::new(ObjectIdAllocator::default()), paths: Arc::new(PathRegistry::default()), - reader_pins: ReaderPinRegistry::default(), + reader_pins: Arc::new(ReaderPinRegistry::default()), reclaim_shards: std::array::from_fn(|_| Mutex::new(())), locks: Mutex::new(DirectoryLocks::default()), locks_changed: Condvar::new(), @@ -631,16 +707,21 @@ impl KvDirectory { &self, path: &Path, after_binding_read: impl FnOnce(), - ) -> Result<(Binding, Arc), OpenReadError> { + ) -> Result<(Binding, Arc, Arc), OpenReadError> { let path_state = self.state.paths.state(path); - let _lifecycle = path_state - .lifecycle - .lock() + let registration = path_state + .reader_registration + .read() .unwrap_or_else(|poisoned| poisoned.into_inner()); let binding = self.read_binding(path)?; after_binding_read(); - let pin = self.state.reader_pins.register(&binding); - Ok((binding, pin)) + let pin = self + .state + .reader_pins + .register(&binding) + .map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))?; + drop(registration); + Ok((binding, pin, path_state)) } } @@ -649,6 +730,8 @@ struct KvFileHandle { namespace: Arc<[u8]>, path: PathBuf, binding: Binding, + _state: Arc, + _path_state: Arc, _pin: Arc, } @@ -777,12 +860,14 @@ impl Drop for KvDirectoryLock { impl Directory for KvDirectory { fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { - let (binding, pin) = self.pinned_binding(path, || {})?; + let (binding, pin, path_state) = self.pinned_binding(path, || {})?; Ok(Arc::new(KvFileHandle { store: self.store.clone(), namespace: self.namespace.clone(), path: path.to_path_buf(), binding, + _state: self.state.clone(), + _path_state: path_state, _pin: pin, })) } @@ -849,6 +934,10 @@ impl Directory for KvDirectory { io::Error::new(io::ErrorKind::InvalidData, "file binding changed during deletion"), )); } + let _reader_registration = path_state + .reader_registration + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); let shard = reclaim_shard(final_binding.object_id); let _queue = self.state.reclaim_shards[usize::from(shard)] .lock() @@ -1365,7 +1454,6 @@ const MAX_OBJECT_EXTENT_BYTES: u128 = 1 << 40; const RECLAIM_ENTRY_FORMAT_VERSION: u8 = 1; const RECLAIM_ENTRY_WHOLE_OBJECT: u8 = 1; const RECLAIM_SHARD_COUNT: usize = 64; -const READER_PIN_PRUNE_THRESHOLD: usize = 64; const KEY_KIND_COUNTER: u8 = 1; const KEY_KIND_BINDING: u8 = 2; const KEY_KIND_ATOMIC: u8 = 3; @@ -2187,17 +2275,21 @@ mod tests { let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); let (binding_read, allow_delete) = std::sync::mpsc::sync_channel(0); let (delete_started, deletion_entered) = std::sync::mpsc::sync_channel(0); + let (deletion_completed, deleted) = std::sync::mpsc::channel(); let deleting_directory = directory.clone(); let deletion = std::thread::spawn(move || { allow_delete.recv().unwrap(); delete_started.send(()).unwrap(); - deleting_directory.delete(path) + let result = deleting_directory.delete(path); + deletion_completed.send(()).unwrap(); + result }); - let (opened, pin) = directory + let (opened, pin, _path_state) = directory .pinned_binding(path, || { binding_read.send(()).unwrap(); deletion_entered.recv().unwrap(); + assert!(deleted.recv_timeout(Duration::from_millis(50)).is_err()); }) .unwrap(); deletion.join().unwrap().unwrap(); @@ -2226,6 +2318,28 @@ mod tests { assert!(!second.state.reader_pins.is_pinned(binding.object_id, None)); } + #[test] + fn open_handle_keeps_the_canonical_directory_state_alive() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.flush().unwrap(); + drop(writer); + let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let state = Arc::downgrade(&directory.state); + let handle = directory.open_read(path).unwrap(); + drop(directory); + + let retained = state.upgrade().unwrap(); + let reopened = FaultingDirectory::new(store); + assert!(Arc::ptr_eq(&retained, &reopened.state)); + assert!(reopened.state.reader_pins.is_pinned(binding.object_id, None)); + drop(handle); + assert!(!reopened.state.reader_pins.is_pinned(binding.object_id, None)); + } + #[test] fn open_read_adds_no_storage_operation_for_pin_registration() { let store = CountingKv::new(); @@ -2252,17 +2366,15 @@ mod tests { writer.flush().unwrap(); let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); - for _ in 0..READER_PIN_PRUNE_THRESHOLD * 4 { + for _ in 0..256 { drop(directory.open_read(path).unwrap()); } - let pins = directory - .state - .reader_pins - .pins + let shard = usize::from(reclaim_shard(binding.object_id)); + let objects = directory.state.reader_pins.shards[shard] .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - assert!(pins[&binding.object_id][&binding.tail_revision].len() <= READER_PIN_PRUNE_THRESHOLD); + assert!(!objects.contains_key(&binding.object_id)); } #[test] @@ -2576,6 +2688,54 @@ mod tests { } } + #[test] + fn open_read_does_not_wait_for_writer_retirement() { + let store = BlockingKv::new(); + let directory = KvDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"first").unwrap(); + writer.flush().unwrap(); + writer.write_all(b" second").unwrap(); + let fence = directory + .state + .paths + .state(path) + .lifecycle + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .writer + .as_ref() + .and_then(Weak::upgrade) + .unwrap(); + store.arm_next_write(); + let helper_store = store.clone(); + let deleting_directory = directory.clone(); + let reading_directory = directory.clone(); + let helper = std::thread::spawn(move || { + assert!( + helper_store.wait_until_blocked(), + "write did not reach the test barrier" + ); + let deletion = std::thread::spawn(move || deleting_directory.delete(path)); + assert!(wait_for_retirement(&fence), "delete did not retire the writer"); + let (opened, received) = std::sync::mpsc::channel(); + let reading = std::thread::spawn(move || opened.send(reading_directory.open_read(path)).unwrap()); + let handle = received + .recv_timeout(Duration::from_secs(5)) + .expect("read waited for writer retirement") + .unwrap(); + helper_store.release_write(); + deletion.join().unwrap().unwrap(); + reading.join().unwrap(); + handle + }); + + writer.flush().unwrap(); + let handle = helper.join().unwrap(); + assert_eq!(handle.read_bytes().unwrap().as_slice(), b"first"); + } + #[test] fn delete_waits_for_in_flight_chunk_storage() { let store = BlockingKv::new(); From aa9afb1b1b5288d50eea1b53916526a8bc2dddfb Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 22:18:15 -0600 Subject: [PATCH 3/9] Shard reader registration fences --- README.md | 3 +- benches/kv_directory.rs | 93 ++++++++++++++++++++++++++++++++++++++++ docs/reclamation-plan.md | 18 ++++---- src/phase0.rs | 82 +++++++++++++++++++++++++++-------- 4 files changed, 168 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index fc1a47f..6704067 100644 --- a/README.md +++ b/README.md @@ -99,7 +99,8 @@ require controlled hardware. The `kv-directory` benchmark measures the caller-visible buffered write path, empty and dirty flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read -handle opens, and distinct-file concurrency at one, two, four, and eight threads. It reports +handle opens, and distinct-file read-open and write concurrency at one, two, four, and eight +threads. It reports percentiles across per-sample mean latencies and uses the deterministic in-memory Phase 0 store to isolate directory coordination from RocksDB and Node transport costs. That store serializes access, so the concurrency cases detect coordination regressions but do not predict RocksDB scaling. diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index ec172f9..97650a6 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -201,6 +201,15 @@ fn main() -> io::Result<()> { &arguments, open_read_case(false, arguments.smoke), )?); + for threads in [1, 2, 4, 8] { + results.push(measure_case( + format!("concurrent-open-read-{threads}t"), + "open-read", + threads, + &arguments, + concurrent_open_read_case(threads, arguments.smoke), + )?); + } for threads in [1, 2, 4, 8] { results.push(measure_case( @@ -514,6 +523,90 @@ fn open_read_case(retain_handles: bool, smoke: bool) -> impl FnMut(usize) -> io: } } +fn concurrent_open_read_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Result { + let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD }; + move |sample| { + let directory = FaultingDirectory::new(FaultingKv::default()); + let mut paths = Vec::with_capacity(threads); + for thread in 0..threads { + let mut thread_paths = Vec::with_capacity(files_per_thread); + for file in 0..files_per_thread { + let path = format!("concurrent-open-read-{sample}-{thread}-{file}"); + let mut writer = open_writer(&directory, Path::new(&path))?; + writer.write_all(b"contents")?; + writer.terminate()?; + thread_paths.push(path); + } + paths.push(thread_paths); + } + let start_gate = Arc::new(StartGate::new()); + let remaining = Arc::new(AtomicUsize::new(threads)); + let (completion, completed) = mpsc::sync_channel(1); + let elapsed_nanoseconds = std::thread::scope(|scope| -> io::Result { + let mut handles = Vec::with_capacity(threads); + for thread_paths in paths { + let directory = directory.clone(); + let worker_start_gate = start_gate.clone(); + let remaining = remaining.clone(); + let completion = completion.clone(); + let handle = std::thread::Builder::new() + .name("kv-directory-open-read-benchmark".to_owned()) + .spawn_scoped(scope, move || -> io::Result<()> { + let Some(started) = worker_start_gate.wait() else { + return Ok(()); + }; + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| -> io::Result<()> { + let mut opened = Vec::with_capacity(thread_paths.len()); + for path in thread_paths { + opened.push( + directory + .open_read(Path::new(&path)) + .map_err(|error| io::Error::other(error.to_string()))?, + ); + } + black_box(opened); + Ok(()) + })) + .unwrap_or_else(|_| Err(io::Error::other("benchmark worker panicked"))); + if remaining.fetch_sub(1, Ordering::AcqRel) == 1 { + let _ = completion.send(started.elapsed().as_nanos()); + } + result + }); + match handle { + Ok(handle) => handles.push(handle), + Err(error) => { + start_gate.cancel(); + for handle in handles { + handle + .join() + .map_err(|_| io::Error::other("benchmark worker panicked"))??; + } + return Err(error); + } + } + } + start_gate.start(threads); + drop(completion); + let elapsed_nanoseconds = completed + .recv() + .map_err(|_| io::Error::other("benchmark workers did not report completion"))?; + for handle in handles { + handle + .join() + .map_err(|_| io::Error::other("benchmark worker panicked"))??; + } + Ok(elapsed_nanoseconds) + })?; + let operations = threads * files_per_thread; + Ok(Sample { + elapsed_nanoseconds, + operations: operations as u64, + bytes: 0, + }) + } +} + fn concurrent_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Result { let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD }; move |sample| { diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 7dec578..cf6fdb7 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -121,11 +121,12 @@ as reclamation, so unrelated objects do not share one pin mutex. This bounds ret by live handles during repeated object, revision, and open/drop churn. Writer state is likewise retained through failed and unterminated writes. Gates recover poisoned state rather than panicking the writer actor. Pins land in the first slice 4 unit before the FIFO consumer. A handle reads its -binding and registers the pin under a shared per-path registration fence. Deletion takes the -exclusive side only for its final queue and binding transition, so concurrent opens remain parallel -and writer retirement does not block them. This preserves one storage read per handle open and adds -no cross-path coordination. The consumer follows only after these lifetime rules are independently -covered. Harper may enable cleanup only after a two-worker test proves that every local search +binding and registers the pin under the shared side of a fixed hash-sharded registration fence. +Deletion takes the exclusive side of the same shard only for its final binding-removal batch, so +concurrent opens remain parallel except for hash collisions and writer retirement does not block +them. This preserves one storage read per handle open and adds no directory-wide coordination. The +consumer follows only after these lifetime rules are independently covered. Harper may enable +cleanup only after a two-worker test proves that every local search handle for one RocksDB-backed generation reaches the same native `DirectoryState`. A process that cannot establish that invariant cannot obtain the cleanup owner lease. @@ -255,7 +256,7 @@ Then deliver reclamation in reviewable slices: enqueue at object deletion; key writer retirement by object id so path reuse cannot retire the replacement writer; 4. add object-id/revision reader pins and coordinate registration with deletion through the - per-path registration fence; + hash-sharded registration fence; 5. measure and, if justified, add tail-supersession entries; then add bounded FIFO draining and deferred queues, low-priority admission, failure observability, close fencing, crash recovery, and Harper's exclusive-owner integration. @@ -290,10 +291,11 @@ request commits as one RocksDB batch. The dependency-free `kv_directory` release benchmark compares adjacent merged slices through Tantivy's public directory interfaces. It reports per-sample-mean p50/p95/p99 and aggregate throughput for caller write sizes, empty and dirty flushes, chunk publication, deletion with closed -and active writers, retained and churned read-handle opens, and distinct-file concurrency. The +and active writers, retained and churned read-handle opens, and distinct-file read-open and write +concurrency. The empty-flush case isolates the per-call retirement-fence cost; the buffered cases show how Tantivy's writer amortizes it in practice. Retained handles measure registration growth, while churned handles -exercise weak-pin pruning. Results are versioned JSON labeled by revision. Shared CI runs a +exercise drop-time pin removal. Results are versioned JSON labeled by revision. Shared CI runs a correctness smoke with no timing threshold; performance decisions use alternating runs on one fixed host. The deterministic Phase 0 store removes RocksDB and Node transport variance but serializes access, so its concurrency results detect directory-coordination regressions rather than predicting diff --git a/src/phase0.rs b/src/phase0.rs index 261bdfc..955307b 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -1,5 +1,6 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::fmt; +use std::hash::{DefaultHasher, Hash, Hasher}; use std::io; use std::io::Write; use std::ops::Range; @@ -265,6 +266,7 @@ struct DirectoryState { allocator: Mutex, paths: Arc, reader_pins: Arc, + reader_registration_shards: [RwLock<()>; READER_REGISTRATION_SHARD_COUNT], reclaim_shards: [Mutex<()>; RECLAIM_SHARD_COUNT], locks: Mutex, locks_changed: Condvar, @@ -286,7 +288,6 @@ struct PathState { path: PathBuf, registry: Weak, lifecycle: Mutex, - reader_registration: RwLock<()>, } #[derive(Default)] @@ -360,7 +361,6 @@ impl PathRegistry { path: path.to_path_buf(), registry: Arc::downgrade(self), lifecycle: Mutex::new(PathLifecycle::default()), - reader_registration: RwLock::new(()), }); states.insert(path.to_path_buf(), Arc::downgrade(&state)); state @@ -505,9 +505,12 @@ impl Drop for ReaderPin { .state .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + debug_assert_ne!(state.total, 0, "reader object pin count is missing"); if state.total == 0 { return; } + let revision_count = state.revisions.get(&self.tail_revision).copied().unwrap_or(0); + debug_assert_ne!(revision_count, 0, "reader revision pin count is missing"); let remove_revision = match state.revisions.get_mut(&self.tail_revision) { Some(revision) if *revision != 0 => { *revision -= 1; @@ -595,6 +598,7 @@ impl KvDirectory { allocator: Mutex::new(ObjectIdAllocator::default()), paths: Arc::new(PathRegistry::default()), reader_pins: Arc::new(ReaderPinRegistry::default()), + reader_registration_shards: std::array::from_fn(|_| RwLock::new(())), reclaim_shards: std::array::from_fn(|_| Mutex::new(())), locks: Mutex::new(DirectoryLocks::default()), locks_changed: Condvar::new(), @@ -707,10 +711,8 @@ impl KvDirectory { &self, path: &Path, after_binding_read: impl FnOnce(), - ) -> Result<(Binding, Arc, Arc), OpenReadError> { - let path_state = self.state.paths.state(path); - let registration = path_state - .reader_registration + ) -> Result<(Binding, Arc), OpenReadError> { + let registration = self.state.reader_registration_shards[reader_registration_shard(path)] .read() .unwrap_or_else(|poisoned| poisoned.into_inner()); let binding = self.read_binding(path)?; @@ -721,7 +723,7 @@ impl KvDirectory { .register(&binding) .map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))?; drop(registration); - Ok((binding, pin, path_state)) + Ok((binding, pin)) } } @@ -731,7 +733,6 @@ struct KvFileHandle { path: PathBuf, binding: Binding, _state: Arc, - _path_state: Arc, _pin: Arc, } @@ -860,14 +861,13 @@ impl Drop for KvDirectoryLock { impl Directory for KvDirectory { fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { - let (binding, pin, path_state) = self.pinned_binding(path, || {})?; + let (binding, pin) = self.pinned_binding(path, || {})?; Ok(Arc::new(KvFileHandle { store: self.store.clone(), namespace: self.namespace.clone(), path: path.to_path_buf(), binding, _state: self.state.clone(), - _path_state: path_state, _pin: pin, })) } @@ -934,10 +934,6 @@ impl Directory for KvDirectory { io::Error::new(io::ErrorKind::InvalidData, "file binding changed during deletion"), )); } - let _reader_registration = path_state - .reader_registration - .write() - .unwrap_or_else(|poisoned| poisoned.into_inner()); let shard = reclaim_shard(final_binding.object_id); let _queue = self.state.reclaim_shards[usize::from(shard)] .lock() @@ -953,6 +949,9 @@ impl Directory for KvDirectory { }; mutations.push(Mutation::Delete(binding_key.clone())); mutations.push(Mutation::Delete(atomic_key)); + let _reader_registration = self.state.reader_registration_shards[reader_registration_shard(path)] + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); match self.store.write(&mutations, WritePolicy::WAL) { Ok(()) => { lifecycle.writer = None; @@ -1454,6 +1453,7 @@ const MAX_OBJECT_EXTENT_BYTES: u128 = 1 << 40; const RECLAIM_ENTRY_FORMAT_VERSION: u8 = 1; const RECLAIM_ENTRY_WHOLE_OBJECT: u8 = 1; const RECLAIM_SHARD_COUNT: usize = 64; +const READER_REGISTRATION_SHARD_COUNT: usize = 256; const KEY_KIND_COUNTER: u8 = 1; const KEY_KIND_BINDING: u8 = 2; const KEY_KIND_ATOMIC: u8 = 3; @@ -1583,6 +1583,12 @@ fn reclaim_shard(object_id: u64) -> u8 { (object_id % RECLAIM_SHARD_COUNT as u64) as u8 } +fn reader_registration_shard(path: &Path) -> usize { + let mut hasher = DefaultHasher::new(); + path.hash(&mut hasher); + (hasher.finish() % READER_REGISTRATION_SHARD_COUNT as u64) as usize +} + fn encode_binding(binding: &Binding) -> Vec { let mut bytes = Vec::with_capacity(45); bytes.push(BINDING_FORMAT_VERSION); @@ -2263,6 +2269,46 @@ mod tests { assert!(!directory.state.reader_pins.is_pinned(second_binding.object_id, None)); } + #[test] + fn revision_pins_are_independent_within_one_object() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"first").unwrap(); + writer.flush().unwrap(); + let first_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let first = directory.open_read(path).unwrap(); + + writer.write_all(b" second").unwrap(); + writer.flush().unwrap(); + let second_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let second = directory.open_read(path).unwrap(); + assert_eq!(first_binding.object_id, second_binding.object_id); + assert_ne!(first_binding.tail_revision, second_binding.tail_revision); + assert!(directory + .state + .reader_pins + .is_pinned(first_binding.object_id, Some(first_binding.tail_revision))); + assert!(directory + .state + .reader_pins + .is_pinned(second_binding.object_id, Some(second_binding.tail_revision))); + + drop(second); + assert!(directory.state.reader_pins.is_pinned(first_binding.object_id, None)); + assert!(directory + .state + .reader_pins + .is_pinned(first_binding.object_id, Some(first_binding.tail_revision))); + assert!(!directory + .state + .reader_pins + .is_pinned(second_binding.object_id, Some(second_binding.tail_revision))); + drop(first); + assert!(!directory.state.reader_pins.is_pinned(first_binding.object_id, None)); + } + #[test] fn binding_read_and_pin_registration_complete_before_delete() { let store = FaultingKv::default(); @@ -2285,7 +2331,7 @@ mod tests { result }); - let (opened, pin, _path_state) = directory + let (opened, pin) = directory .pinned_binding(path, || { binding_read.send(()).unwrap(); deletion_entered.recv().unwrap(); @@ -2721,11 +2767,9 @@ mod tests { assert!(wait_for_retirement(&fence), "delete did not retire the writer"); let (opened, received) = std::sync::mpsc::channel(); let reading = std::thread::spawn(move || opened.send(reading_directory.open_read(path)).unwrap()); - let handle = received - .recv_timeout(Duration::from_secs(5)) - .expect("read waited for writer retirement") - .unwrap(); + let handle = received.recv_timeout(Duration::from_secs(5)); helper_store.release_write(); + let handle = handle.expect("read waited for writer retirement").unwrap(); deletion.join().unwrap().unwrap(); reading.join().unwrap(); handle From 6c1c12a1ec4a2a684209617a2d285eec929e7c36 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 22:39:38 -0600 Subject: [PATCH 4/9] Address reader pin review findings --- README.md | 8 ++++---- benches/kv_directory.rs | 17 +++++++++-------- docs/reclamation-plan.md | 33 +++++++++++++++++---------------- src/phase0.rs | 19 +++++++++++-------- 4 files changed, 41 insertions(+), 36 deletions(-) diff --git a/README.md b/README.md index 6704067..cf94ad7 100644 --- a/README.md +++ b/README.md @@ -100,10 +100,10 @@ require controlled hardware. The `kv-directory` benchmark measures the caller-visible buffered write path, empty and dirty flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read handle opens, and distinct-file read-open and write concurrency at one, two, four, and eight -threads. It reports -percentiles across per-sample mean latencies and uses the deterministic in-memory Phase 0 store to -isolate directory coordination from RocksDB and Node transport costs. That store serializes access, -so the concurrency cases detect coordination regressions but do not predict RocksDB scaling. +threads. It reports percentiles across per-sample mean latencies and uses the deterministic +in-memory Phase 0 store to isolate directory coordination from RocksDB and Node transport costs. +That store serializes access, so the concurrency cases detect coordination regressions but do not +predict RocksDB scaling. Compare two optimized builds on the same quiet host; records include the Git revision and dirty state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index 97650a6..8d99c0b 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -544,18 +544,18 @@ fn concurrent_open_read_case(threads: usize, smoke: bool) -> impl FnMut(usize) - let (completion, completed) = mpsc::sync_channel(1); let elapsed_nanoseconds = std::thread::scope(|scope| -> io::Result { let mut handles = Vec::with_capacity(threads); - for thread_paths in paths { + for (thread, thread_paths) in paths.into_iter().enumerate() { let directory = directory.clone(); let worker_start_gate = start_gate.clone(); let remaining = remaining.clone(); let completion = completion.clone(); let handle = std::thread::Builder::new() - .name("kv-directory-open-read-benchmark".to_owned()) - .spawn_scoped(scope, move || -> io::Result<()> { + .name(format!("kv-directory-open-read-benchmark-{thread}")) + .spawn_scoped(scope, move || { let Some(started) = worker_start_gate.wait() else { - return Ok(()); + return Ok(Vec::new()); }; - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| -> io::Result<()> { + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { let mut opened = Vec::with_capacity(thread_paths.len()); for path in thread_paths { opened.push( @@ -564,8 +564,8 @@ fn concurrent_open_read_case(threads: usize, smoke: bool) -> impl FnMut(usize) - .map_err(|error| io::Error::other(error.to_string()))?, ); } - black_box(opened); - Ok(()) + black_box(&opened); + Ok::<_, io::Error>(opened) })) .unwrap_or_else(|_| Err(io::Error::other("benchmark worker panicked"))); if remaining.fetch_sub(1, Ordering::AcqRel) == 1 { @@ -592,9 +592,10 @@ fn concurrent_open_read_case(threads: usize, smoke: bool) -> impl FnMut(usize) - .recv() .map_err(|_| io::Error::other("benchmark workers did not report completion"))?; for handle in handles { - handle + let opened = handle .join() .map_err(|_| io::Error::other("benchmark worker panicked"))??; + black_box(opened); } Ok(elapsed_nanoseconds) })?; diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index cf6fdb7..30457c1 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -130,14 +130,16 @@ cleanup only after a two-worker test proves that every local search handle for one RocksDB-backed generation reaches the same native `DirectoryState`. A process that cannot establish that invariant cannot obtain the cleanup owner lease. -Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot gain -a new pin because its binding is gone; a tail-only entry can gain no new pin for its old revision -after the newer binding is published. Whole-object entries match any pin for that object, while -tail-only entries match only the exact object id and tail revision. If a matching pin remains, -cleanup moves the entry to a cleanup-only deferred queue with bounded backoff rather than blocking -later garbage. Foreground publishers only mutate the ingress tail; cleanup only mutates the ingress -head and deferred queue. A low-priority cleanup host operation therefore never owns the mutex or -counter needed by a foreground enqueue. +Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot +gain a new pin because its binding is gone. Before tail-only reclamation is enabled, publication of +the newer binding and its superseded-tail entry must take the exclusive side of the same +registration fence used by deletion. That prevents a reader from registering the old revision +after publication. Whole-object entries match any pin for that object, while tail-only entries +match only the exact object id and tail revision. If a matching pin remains, cleanup moves the entry +to a cleanup-only deferred queue with bounded backoff rather than blocking later garbage. +Foreground publishers only mutate the ingress tail; cleanup only mutates the ingress head and +deferred queue. A low-priority cleanup host operation therefore never owns the mutex or counter +needed by a foreground enqueue. This process-local pin model is valid only while Harper's derived-index lifecycle guarantees one active generation owner. Cleanup requires an explicit owner lease supplied by that lifecycle; the @@ -292,14 +294,13 @@ The dependency-free `kv_directory` release benchmark compares adjacent merged sl Tantivy's public directory interfaces. It reports per-sample-mean p50/p95/p99 and aggregate throughput for caller write sizes, empty and dirty flushes, chunk publication, deletion with closed and active writers, retained and churned read-handle opens, and distinct-file read-open and write -concurrency. The -empty-flush case isolates the per-call retirement-fence cost; the buffered cases show how Tantivy's -writer amortizes it in practice. Retained handles measure registration growth, while churned handles -exercise drop-time pin removal. Results are versioned JSON labeled by revision. Shared CI runs a -correctness smoke with no timing threshold; performance decisions use alternating runs on one fixed -host. The deterministic Phase 0 store removes RocksDB and Node transport variance but serializes -access, so its concurrency results detect directory-coordination regressions rather than predicting -RocksDB scaling. +concurrency. The empty-flush case isolates the per-call retirement-fence cost; the buffered cases +show how Tantivy's writer amortizes it in practice. Retained handles measure registration growth, +while churned handles exercise drop-time pin removal. Results are versioned JSON labeled by +revision. Shared CI runs a correctness smoke with no timing threshold; performance decisions use +alternating runs on one fixed host. The deterministic Phase 0 store removes RocksDB and Node +transport variance but serializes access, so its concurrency results detect directory-coordination +regressions rather than predicting RocksDB scaling. The completed mapping tests cover repeated flush, delete/recreate with a retained old reader, abandoned writers, partial object cleanup, restart during a range and between FIFO entries, namespace diff --git a/src/phase0.rs b/src/phase0.rs index 955307b..18ce160 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -505,23 +505,23 @@ impl Drop for ReaderPin { .state .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - debug_assert_ne!(state.total, 0, "reader object pin count is missing"); if state.total == 0 { return; } - let revision_count = state.revisions.get(&self.tail_revision).copied().unwrap_or(0); - debug_assert_ne!(revision_count, 0, "reader revision pin count is missing"); let remove_revision = match state.revisions.get_mut(&self.tail_revision) { Some(revision) if *revision != 0 => { *revision -= 1; *revision == 0 } - _ => return, + _ => false, }; state.total -= 1; if remove_revision { state.revisions.remove(&self.tail_revision); } + if state.total == 0 { + state.revisions.clear(); + } } } @@ -949,10 +949,13 @@ impl Directory for KvDirectory { }; mutations.push(Mutation::Delete(binding_key.clone())); mutations.push(Mutation::Delete(atomic_key)); - let _reader_registration = self.state.reader_registration_shards[reader_registration_shard(path)] - .write() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - match self.store.write(&mutations, WritePolicy::WAL) { + let write_result = { + let _reader_registration = self.state.reader_registration_shards[reader_registration_shard(path)] + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.store.write(&mutations, WritePolicy::WAL) + }; + match write_result { Ok(()) => { lifecycle.writer = None; Ok(()) From a443bf28ae25ed00cca303ac74e76deddf5be495 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 22:53:18 -0600 Subject: [PATCH 5/9] Remove reader pin allocation --- docs/reclamation-plan.md | 13 ++++++++++--- src/phase0.rs | 10 +++++----- 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 30457c1..6f7b228 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -126,9 +126,16 @@ Deletion takes the exclusive side of the same shard only for its final binding-r concurrent opens remain parallel except for hash collisions and writer retirement does not block them. This preserves one storage read per handle open and adds no directory-wide coordination. The consumer follows only after these lifetime rules are independently covered. Harper may enable -cleanup only after a two-worker test proves that every local search -handle for one RocksDB-backed generation reaches the same native `DirectoryState`. A process that -cannot establish that invariant cannot obtain the cleanup owner lease. +cleanup only after a two-worker test proves that every local search handle for one RocksDB-backed +generation reaches the same native `DirectoryState`. A process that cannot establish that invariant +cannot obtain the cleanup owner lease. + +The blocking fence deliberately spans a reader's binding read and pin registration and deletion's +single atomic host write. Ending either side earlier reopens the registration race. A stalled delete +can therefore delay an unrelated open whose path collides in the same one of 256 shards. Replacing +this bounded collision risk requires a quiescence or epoch protocol that prevents cleanup until all +readers that started before publication have either registered or exited; it is not a safe lock +substitution. Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot gain a new pin because its binding is gone. Before tail-only reclamation is enabled, publication of diff --git a/src/phase0.rs b/src/phase0.rs index 18ce160..12c1207 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -441,7 +441,7 @@ impl Drop for WriterClaim<'_> { } impl ReaderPinRegistry { - fn register(self: &Arc, binding: &Binding) -> io::Result> { + fn register(self: &Arc, binding: &Binding) -> io::Result { let shard = usize::from(reclaim_shard(binding.object_id)); let object = { let mut objects = self.shards[shard] @@ -475,10 +475,10 @@ impl ReaderPinRegistry { state.total = total; state.revisions.insert(binding.tail_revision, revision); drop(state); - Ok(Arc::new(ReaderPin { + Ok(ReaderPin { object, tail_revision: binding.tail_revision, - })) + }) } #[cfg(test)] @@ -711,7 +711,7 @@ impl KvDirectory { &self, path: &Path, after_binding_read: impl FnOnce(), - ) -> Result<(Binding, Arc), OpenReadError> { + ) -> Result<(Binding, ReaderPin), OpenReadError> { let registration = self.state.reader_registration_shards[reader_registration_shard(path)] .read() .unwrap_or_else(|poisoned| poisoned.into_inner()); @@ -733,7 +733,7 @@ struct KvFileHandle { path: PathBuf, binding: Binding, _state: Arc, - _pin: Arc, + _pin: ReaderPin, } impl fmt::Debug for KvFileHandle { From 355cc04bfcf4c990d32166917714cdf91155a15f Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 23:10:57 -0600 Subject: [PATCH 6/9] Optimize reader revision pins --- README.md | 10 ++-- benches/kv_directory.rs | 24 ++++++-- docs/reclamation-plan.md | 33 ++++++---- src/phase0.rs | 126 +++++++++++++++++++++++++++++++-------- 4 files changed, 148 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index cf94ad7..96706d7 100644 --- a/README.md +++ b/README.md @@ -99,11 +99,11 @@ require controlled hardware. The `kv-directory` benchmark measures the caller-visible buffered write path, empty and dirty flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read -handle opens, and distinct-file read-open and write concurrency at one, two, four, and eight -threads. It reports percentiles across per-sample mean latencies and uses the deterministic -in-memory Phase 0 store to isolate directory coordination from RocksDB and Node transport costs. -That store serializes access, so the concurrency cases detect coordination regressions but do not -predict RocksDB scaling. +handle opens, shared- and distinct-file read-open concurrency, and distinct-file write concurrency +at one, two, four, and eight threads. It reports percentiles across per-sample mean latencies and +uses the deterministic in-memory Phase 0 store to isolate directory coordination from RocksDB and +Node transport costs. That store serializes access, so the concurrency cases detect coordination +regressions but do not predict RocksDB scaling. Compare two optimized builds on the same quiet host; records include the Git revision and dirty state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index 8d99c0b..ab3527c 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -207,7 +207,14 @@ fn main() -> io::Result<()> { "open-read", threads, &arguments, - concurrent_open_read_case(threads, arguments.smoke), + concurrent_open_read_case(threads, arguments.smoke, false), + )?); + results.push(measure_case( + format!("concurrent-shared-open-read-{threads}t"), + "open-read", + threads, + &arguments, + concurrent_open_read_case(threads, arguments.smoke, true), )?); } @@ -523,15 +530,20 @@ fn open_read_case(retain_handles: bool, smoke: bool) -> impl FnMut(usize) -> io: } } -fn concurrent_open_read_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Result { +fn concurrent_open_read_case( + threads: usize, + smoke: bool, + shared_paths: bool, +) -> impl FnMut(usize) -> io::Result { let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD }; move |sample| { let directory = FaultingDirectory::new(FaultingKv::default()); let mut paths = Vec::with_capacity(threads); - for thread in 0..threads { + let path_groups = if shared_paths { 1 } else { threads }; + for thread in 0..path_groups { let mut thread_paths = Vec::with_capacity(files_per_thread); for file in 0..files_per_thread { - let path = format!("concurrent-open-read-{sample}-{thread}-{file}"); + let path = format!("concurrent-open-read-{shared_paths}-{sample}-{thread}-{file}"); let mut writer = open_writer(&directory, Path::new(&path))?; writer.write_all(b"contents")?; writer.terminate()?; @@ -539,6 +551,10 @@ fn concurrent_open_read_case(threads: usize, smoke: bool) -> impl FnMut(usize) - } paths.push(thread_paths); } + if shared_paths { + let shared = paths[0].clone(); + paths.resize_with(threads, || shared.clone()); + } let start_gate = Arc::new(StartGate::new()); let remaining = Arc::new(AtomicUsize::new(threads)); let (completion, completed) = mpsc::sync_channel(1); diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 6f7b228..5b29b14 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -115,12 +115,13 @@ gate is held by chunk staging or publication across a host round trip. Pins are indexed by object id, not logical path, so delete/recreate cannot hide handles to the prior object. Each object has counted pins by tail revision. A handle owns one pin and retains the shared -directory and path state; dropping it decrements the revision and object counts, and dropping the -last pin removes the object's weak registry entry. Registry shards use the same object-id partition -as reclamation, so unrelated objects do not share one pin mutex. This bounds retained control state -by live handles during repeated object, revision, and open/drop churn. Writer state is likewise -retained through failed and unterminated writes. Gates recover poisoned state rather than panicking -the writer actor. Pins land in the first slice 4 unit before the FIFO consumer. A handle reads its +directory state; dropping it decrements the revision and object counts, and dropping the last pin +removes the object's weak registry entry. Registry shards use the same object-id partition as +reclamation, while revision counts use a per-object mutex and keep the common single-revision case +inline. This bounds retained control state by live handles during repeated object, revision, and +open/drop churn. Writer state is likewise retained through failed and unterminated writes. Gates +recover poisoned state rather than panicking the writer actor. Pins land in the first slice 4 unit +before the FIFO consumer. A handle reads its binding and registers the pin under the shared side of a fixed hash-sharded registration fence. Deletion takes the exclusive side of the same shard only for its final binding-removal batch, so concurrent opens remain parallel except for hash collisions and writer retirement does not block @@ -135,7 +136,10 @@ single atomic host write. Ending either side earlier reopens the registration ra can therefore delay an unrelated open whose path collides in the same one of 256 shards. Replacing this bounded collision risk requires a quiescence or epoch protocol that prevents cleanup until all readers that started before publication have either registered or exited; it is not a safe lock -substitution. +substitution. Conversely, a stalled colliding reader can hold deletion behind its binding read; +because deletion already owns that path's lifecycle gate, this can also delay an `open_write()` for +the path being deleted. The host transport's definitive-completion contract bounds this only when +the host operation completes or the transport closes. Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot gain a new pin because its binding is gone. Before tail-only reclamation is enabled, publication of @@ -297,13 +301,20 @@ real Tantivy merge and garbage-collection cycles and after crash/reopen. `Faulti recovery at atomic batch boundaries; the Harper integration separately proves that one host write request commits as one RocksDB batch. +Slice 4 tests object and exact-revision pin counts, delete/recreate identity, the binding-read and +pin-registration race, shared state across independently constructed directories, handle-owned +state lifetime, one storage read per open, bounded open/drop churn, and reader independence from +writer retirement. The release benchmark adds retained and churned opens plus shared- and +distinct-file concurrency at one, two, four, and eight threads. + The dependency-free `kv_directory` release benchmark compares adjacent merged slices through Tantivy's public directory interfaces. It reports per-sample-mean p50/p95/p99 and aggregate throughput for caller write sizes, empty and dirty flushes, chunk publication, deletion with closed -and active writers, retained and churned read-handle opens, and distinct-file read-open and write -concurrency. The empty-flush case isolates the per-call retirement-fence cost; the buffered cases -show how Tantivy's writer amortizes it in practice. Retained handles measure registration growth, -while churned handles exercise drop-time pin removal. Results are versioned JSON labeled by +and active writers, retained and churned read-handle opens, shared- and distinct-file read-open +concurrency, and distinct-file write concurrency. The empty-flush case isolates the per-call +retirement-fence cost; the buffered cases show how Tantivy's writer amortizes it in practice. +Retained handles measure registration growth, churned handles exercise drop-time pin removal, and +shared-file concurrency exercises per-object pin accounting. Results are versioned JSON labeled by revision. Shared CI runs a correctness smoke with no timing threshold; performance decisions use alternating runs on one fixed host. The deterministic Phase 0 store removes RocksDB and Node transport variance but serializes access, so its concurrency results detect directory-coordination diff --git a/src/phase0.rs b/src/phase0.rs index 12c1207..ff6656d 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -320,7 +320,15 @@ struct ObjectReaderPins { #[derive(Default)] struct ObjectReaderPinState { total: usize, - revisions: HashMap, + revisions: ReaderRevisions, +} + +#[derive(Default)] +enum ReaderRevisions { + #[default] + None, + One(u64, usize), + Many(HashMap), } struct ReaderPin { @@ -465,15 +473,8 @@ impl ReaderPinRegistry { .total .checked_add(1) .ok_or_else(|| io::Error::other("reader pin count exhausted"))?; - let revision = state - .revisions - .get(&binding.tail_revision) - .copied() - .unwrap_or(0) - .checked_add(1) - .ok_or_else(|| io::Error::other("reader revision pin count exhausted"))?; + state.revisions.increment(binding.tail_revision)?; state.total = total; - state.revisions.insert(binding.tail_revision, revision); drop(state); Ok(ReaderPin { object, @@ -492,9 +493,82 @@ impl ReaderPinRegistry { }; drop(objects); let state = object.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - tail_revision.map_or(state.total != 0, |revision| { - state.revisions.get(&revision).is_some_and(|count| *count != 0) - }) + tail_revision.map_or(state.total != 0, |revision| state.revisions.contains(revision)) + } +} + +impl ReaderRevisions { + fn increment(&mut self, tail_revision: u64) -> io::Result<()> { + match self { + Self::None => *self = Self::One(tail_revision, 1), + Self::One(revision, count) if *revision == tail_revision => { + *count = count + .checked_add(1) + .ok_or_else(|| io::Error::other("reader revision pin count exhausted"))?; + } + Self::One(revision, count) => { + let mut revisions = HashMap::with_capacity(2); + revisions.insert(*revision, *count); + revisions.insert(tail_revision, 1); + *self = Self::Many(revisions); + } + Self::Many(revisions) => { + let count = revisions.get(&tail_revision).copied().unwrap_or(0); + revisions.insert( + tail_revision, + count + .checked_add(1) + .ok_or_else(|| io::Error::other("reader revision pin count exhausted"))?, + ); + } + } + Ok(()) + } + + #[cfg(test)] + fn contains(&self, tail_revision: u64) -> bool { + match self { + Self::None => false, + Self::One(revision, count) => *revision == tail_revision && *count != 0, + Self::Many(revisions) => revisions.get(&tail_revision).is_some_and(|count| *count != 0), + } + } + + fn decrement(&mut self, tail_revision: u64) -> bool { + match self { + Self::None => false, + Self::One(revision, count) if *revision == tail_revision && *count > 1 => { + *count -= 1; + true + } + Self::One(revision, count) if *revision == tail_revision && *count == 1 => { + *self = Self::None; + true + } + Self::One(_, _) => false, + Self::Many(revisions) => { + let remove = match revisions.get_mut(&tail_revision) { + Some(count) if *count > 1 => { + *count -= 1; + false + } + Some(count) if *count == 1 => true, + _ => return false, + }; + if remove { + revisions.remove(&tail_revision); + } + let remaining = if revisions.len() == 1 { + revisions.iter().next().map(|(revision, count)| (*revision, *count)) + } else { + None + }; + if let Some((revision, count)) = remaining { + *self = Self::One(revision, count); + } + true + } + } } } @@ -505,22 +579,20 @@ impl Drop for ReaderPin { .state .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + debug_assert!( + state.total != 0 || std::thread::panicking(), + "reader object pin count is missing" + ); if state.total == 0 { return; } - let remove_revision = match state.revisions.get_mut(&self.tail_revision) { - Some(revision) if *revision != 0 => { - *revision -= 1; - *revision == 0 - } - _ => false, - }; - state.total -= 1; - if remove_revision { - state.revisions.remove(&self.tail_revision); - } - if state.total == 0 { - state.revisions.clear(); + let revision_found = state.revisions.decrement(self.tail_revision); + debug_assert!( + revision_found || std::thread::panicking(), + "reader revision pin count is missing" + ); + if revision_found { + state.total -= 1; } } } @@ -2842,6 +2914,10 @@ mod tests { reclaim_shard(first_binding.object_id), reclaim_shard(second_binding.object_id) ); + assert_ne!( + reader_registration_shard(Path::new("first")), + reader_registration_shard(Path::new("second")) + ); store.arm_next_write(); let first_directory = directory.clone(); From 6fbf940b855b0cc044f1f4dd7fcfd6c0236567f8 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 23:19:28 -0600 Subject: [PATCH 7/9] Measure concurrent reader pins --- README.md | 7 +++-- benches/kv_directory.rs | 66 ++++++++++++++++++++++++++++++++++++---- docs/reclamation-plan.md | 7 +++-- 3 files changed, 68 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 96706d7..8738c9a 100644 --- a/README.md +++ b/README.md @@ -101,9 +101,10 @@ The `kv-directory` benchmark measures the caller-visible buffered write path, em flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read handle opens, shared- and distinct-file read-open concurrency, and distinct-file write concurrency at one, two, four, and eight threads. It reports percentiles across per-sample mean latencies and -uses the deterministic in-memory Phase 0 store to isolate directory coordination from RocksDB and -Node transport costs. That store serializes access, so the concurrency cases detect coordination -regressions but do not predict RocksDB scaling. +uses a deterministic in-memory store to isolate directory coordination from RocksDB and Node +transport costs. The store permits concurrent point reads but serializes writes, so read-open cases +isolate registration contention while write cases detect coordination regressions; neither predicts +RocksDB scaling. Compare two optimized builds on the same quiet host; records include the Git revision and dirty state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index ab3527c..c19dc6b 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeMap; use std::env; use std::fmt::Write as FmtWrite; use std::hint::black_box; @@ -5,14 +6,16 @@ use std::io; use std::io::Write as IoWrite; use std::path::Path; use std::process::Command; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc; -use std::sync::{Arc, Condvar, Mutex, OnceLock}; +use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock}; use std::time::{Instant, SystemTime, UNIX_EPOCH}; -use harper_fulltext::phase0::{FaultingDirectory, FaultingKv}; +use harper_fulltext::phase0::{ + FaultingDirectory, FaultingKv, KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy, +}; use harper_fulltext::TANTIVY_VERSION; -use tantivy::directory::{Directory, TerminatingWrite, WritePtr}; +use tantivy::directory::{Directory, OwnedBytes, TerminatingWrite, WritePtr}; const WRITE_BYTES_PER_SAMPLE: usize = 128 * 1024; const MIN_WRITE_OPERATIONS_PER_SAMPLE: usize = 256; @@ -24,6 +27,57 @@ const OPEN_READS_PER_SAMPLE: usize = 10_000; const CONCURRENT_BYTES_PER_FILE: usize = 4 * 1024; const CONCURRENT_FILES_PER_THREAD: usize = 64; +static NEXT_BENCHMARK_KV_IDENTITY: AtomicU64 = AtomicU64::new(1); + +#[derive(Clone)] +struct BenchmarkKv { + identity: u64, + values: Arc, Vec>>>, +} + +type BenchmarkDirectory = KvDirectory; + +impl Default for BenchmarkKv { + fn default() -> Self { + let identity = NEXT_BENCHMARK_KV_IDENTITY.fetch_add(1, Ordering::Relaxed); + assert_ne!(identity, 0, "benchmark store identity space exhausted"); + Self { + identity, + values: Arc::new(RwLock::new(BTreeMap::new())), + } + } +} + +impl KvStore for BenchmarkKv { + fn identity(&self) -> KvStoreIdentity { + KvStoreIdentity(0, self.identity, 0) + } + + fn read(&self, key: &[u8]) -> io::Result> { + let values = self.values.read().unwrap_or_else(|poisoned| poisoned.into_inner()); + Ok(values.get(key).cloned().map(OwnedBytes::new)) + } + + fn write(&self, mutations: &[Mutation], _policy: WritePolicy) -> io::Result<()> { + let mut values = self.values.write().unwrap_or_else(|poisoned| poisoned.into_inner()); + for mutation in mutations { + match mutation { + Mutation::Put(key, value) => { + values.insert(key.clone(), value.clone()); + } + Mutation::Delete(key) => { + values.remove(key); + } + } + } + Ok(()) + } + + fn sync(&self) -> io::Result<()> { + Ok(()) + } +} + struct Arguments { samples: usize, warmup_samples: usize, @@ -537,7 +591,7 @@ fn concurrent_open_read_case( ) -> impl FnMut(usize) -> io::Result { let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD }; move |sample| { - let directory = FaultingDirectory::new(FaultingKv::default()); + let directory = BenchmarkDirectory::new(BenchmarkKv::default()); let mut paths = Vec::with_capacity(threads); let path_groups = if shared_paths { 1 } else { threads }; for thread in 0..path_groups { @@ -696,7 +750,7 @@ fn concurrent_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Resu } } -fn open_writer(directory: &FaultingDirectory, path: &Path) -> io::Result { +fn open_writer(directory: &impl Directory, path: &Path) -> io::Result { directory .open_write(path) .map_err(|error| io::Error::other(error.to_string())) diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 5b29b14..7d77274 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -316,9 +316,10 @@ retirement-fence cost; the buffered cases show how Tantivy's writer amortizes it Retained handles measure registration growth, churned handles exercise drop-time pin removal, and shared-file concurrency exercises per-object pin accounting. Results are versioned JSON labeled by revision. Shared CI runs a correctness smoke with no timing threshold; performance decisions use -alternating runs on one fixed host. The deterministic Phase 0 store removes RocksDB and Node -transport variance but serializes access, so its concurrency results detect directory-coordination -regressions rather than predicting RocksDB scaling. +alternating runs on one fixed host. The deterministic benchmark store removes RocksDB and Node +transport variance, permits concurrent point reads, and serializes writes. Read-open concurrency +therefore isolates registration contention, while write concurrency detects directory-coordination +regressions; neither predicts RocksDB scaling. The completed mapping tests cover repeated flush, delete/recreate with a retained old reader, abandoned writers, partial object cleanup, restart during a range and between FIFO entries, namespace From 9af336a6ca15e7df65c4ba3948a4e2868ec81eed Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 23:25:49 -0600 Subject: [PATCH 8/9] Make reader benchmark comparisons explicit --- README.md | 3 ++- benches/kv_directory.rs | 13 ++++++++----- docs/reclamation-plan.md | 7 +++++-- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 8738c9a..9848a80 100644 --- a/README.md +++ b/README.md @@ -104,7 +104,8 @@ at one, two, four, and eight threads. It reports percentiles across per-sample m uses a deterministic in-memory store to isolate directory coordination from RocksDB and Node transport costs. The store permits concurrent point reads but serializes writes, so read-open cases isolate registration contention while write cases detect coordination regressions; neither predicts -RocksDB scaling. +RocksDB scaling. The `-rw-` read-open case names establish a new comparison series and must not be +compared with results from the earlier serialized-store cases. Compare two optimized builds on the same quiet host; records include the Git revision and dirty state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index c19dc6b..f94cddc 100644 --- a/benches/kv_directory.rs +++ b/benches/kv_directory.rs @@ -50,7 +50,7 @@ impl Default for BenchmarkKv { impl KvStore for BenchmarkKv { fn identity(&self) -> KvStoreIdentity { - KvStoreIdentity(0, self.identity, 0) + KvStoreIdentity(1, self.identity, 0) } fn read(&self, key: &[u8]) -> io::Result> { @@ -257,14 +257,16 @@ fn main() -> io::Result<()> { )?); for threads in [1, 2, 4, 8] { results.push(measure_case( - format!("concurrent-open-read-{threads}t"), + format!("concurrent-open-read-rw-{threads}t"), "open-read", threads, &arguments, concurrent_open_read_case(threads, arguments.smoke, false), )?); + } + for threads in [2, 4, 8] { results.push(measure_case( - format!("concurrent-shared-open-read-{threads}t"), + format!("concurrent-shared-open-read-rw-{threads}t"), "open-read", threads, &arguments, @@ -615,6 +617,7 @@ fn concurrent_open_read_case( let elapsed_nanoseconds = std::thread::scope(|scope| -> io::Result { let mut handles = Vec::with_capacity(threads); for (thread, thread_paths) in paths.into_iter().enumerate() { + let opened = Vec::with_capacity(thread_paths.len()); let directory = directory.clone(); let worker_start_gate = start_gate.clone(); let remaining = remaining.clone(); @@ -625,8 +628,8 @@ fn concurrent_open_read_case( let Some(started) = worker_start_gate.wait() else { return Ok(Vec::new()); }; - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - let mut opened = Vec::with_capacity(thread_paths.len()); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || { + let mut opened = opened; for path in thread_paths { opened.push( directory diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 7d77274..921c251 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -139,7 +139,9 @@ readers that started before publication have either registered or exited; it is substitution. Conversely, a stalled colliding reader can hold deletion behind its binding read; because deletion already owns that path's lifecycle gate, this can also delay an `open_write()` for the path being deleted. The host transport's definitive-completion contract bounds this only when -the host operation completes or the transport closes. +the host operation completes or the transport closes. The initial 256-shard count bounds memory and +keeps collision probability low; it is not yet validated against Harper host-read p99 and must be +measured before cleanup is enabled. Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot gain a new pin because its binding is gone. Before tail-only reclamation is enabled, publication of @@ -319,7 +321,8 @@ revision. Shared CI runs a correctness smoke with no timing threshold; performan alternating runs on one fixed host. The deterministic benchmark store removes RocksDB and Node transport variance, permits concurrent point reads, and serializes writes. Read-open concurrency therefore isolates registration contention, while write concurrency detects directory-coordination -regressions; neither predicts RocksDB scaling. +regressions; neither predicts RocksDB scaling. The `-rw-` read-open case names begin a new comparison +series because the earlier cases used the serialized fault-injection store. The completed mapping tests cover repeated flush, delete/recreate with a retained old reader, abandoned writers, partial object cleanup, restart during a range and between FIFO entries, namespace From 008e2d28d02480659899535b6c247a9693d84059 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Wed, 9 Sep 2026 23:29:26 -0600 Subject: [PATCH 9/9] Correct reader benchmark thread ranges --- README.md | 14 +++++++------- docs/reclamation-plan.md | 3 ++- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 9848a80..e23e3cd 100644 --- a/README.md +++ b/README.md @@ -99,13 +99,13 @@ require controlled hardware. The `kv-directory` benchmark measures the caller-visible buffered write path, empty and dirty flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read -handle opens, shared- and distinct-file read-open concurrency, and distinct-file write concurrency -at one, two, four, and eight threads. It reports percentiles across per-sample mean latencies and -uses a deterministic in-memory store to isolate directory coordination from RocksDB and Node -transport costs. The store permits concurrent point reads but serializes writes, so read-open cases -isolate registration contention while write cases detect coordination regressions; neither predicts -RocksDB scaling. The `-rw-` read-open case names establish a new comparison series and must not be -compared with results from the earlier serialized-store cases. +handle opens, distinct-file read-open and write concurrency at one, two, four, and eight threads, +and shared-file read-open concurrency at two, four, and eight threads. It reports percentiles across +per-sample mean latencies and uses a deterministic in-memory store to isolate directory coordination +from RocksDB and Node transport costs. The store permits concurrent point reads but serializes +writes, so read-open cases isolate registration contention while write cases detect coordination +regressions; neither predicts RocksDB scaling. The `-rw-` read-open case names establish a new +comparison series and must not be compared with results from the earlier serialized-store cases. Compare two optimized builds on the same quiet host; records include the Git revision and dirty state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 921c251..7c36a8e 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -307,7 +307,8 @@ Slice 4 tests object and exact-revision pin counts, delete/recreate identity, th pin-registration race, shared state across independently constructed directories, handle-owned state lifetime, one storage read per open, bounded open/drop churn, and reader independence from writer retirement. The release benchmark adds retained and churned opens plus shared- and -distinct-file concurrency at one, two, four, and eight threads. +distinct-file concurrency. Distinct-file cases run at one, two, four, and eight threads; shared-file +cases start at two threads. The dependency-free `kv_directory` release benchmark compares adjacent merged slices through Tantivy's public directory interfaces. It reports per-sample-mean p50/p95/p99 and aggregate