diff --git a/src/btree/mod.rs b/src/btree/mod.rs index fbbcda5..4d9e1df 100644 --- a/src/btree/mod.rs +++ b/src/btree/mod.rs @@ -12,4 +12,4 @@ pub(crate) mod tree; // `pub` here is crate-scoped in effect: the `btree` module itself is // `pub(crate)`, so this does not escape the crate. -pub use tree::BTree; +pub use tree::{BTree, ConsumedPages, PageSource}; diff --git a/src/btree/tree/core.rs b/src/btree/tree/core.rs index d7485d3..daa85a8 100644 --- a/src/btree/tree/core.rs +++ b/src/btree/tree/core.rs @@ -15,6 +15,7 @@ use crate::btree::internal::{self, Internal}; use crate::btree::leaf::Leaf; use crate::btree::node::{NodeKind, OFF_DUAL_USE, body_capacity, read_header, write_u64_le}; use crate::btree::overflow; +use crate::btree::tree::page_source::PageSource; /// Encoded size a value of `value_len` bytes will occupy in a leaf record, /// computed before the value is built. @@ -108,17 +109,18 @@ pub struct BTree { pub(super) root_page_id: u64, pub(super) next_page_id: u64, pub(super) freed: Vec, + /// The subset of pages freed this session that `allocate_page` may hand + /// back immediately — i.e. those that satisfied the reuse rule below at the + /// moment they were freed. Eligibility is decided once, in + /// [`Self::free_page`], because neither input to that decision can move + /// afterwards: [`PageSource`] is fixed for the tree's whole life, and the + /// sink it carries cannot shrink while the session runs. Deciding it here + /// keeps allocation O(1); evaluating it per allocation instead meant + /// rescanning `freed` against the sink on every call, which is quadratic in + /// the number of pages a flush touches and made large flushes effectively + /// never finish. Drained together with `freed` by `drain_freed`. + pub(super) freed_reusable: Vec, pub(super) page_size: usize, - /// Minimum `page_id` that may be recycled from `freed` within this session. - /// Pages below this threshold existed before the session began: they are - /// still referenced by the last durable header (a copy-on-write free does - /// not unreference them on disk until the *next* header lands) and may - /// also be pinned by reader snapshots. They must not be overwritten - /// in-session; they re-enter circulation through the deferred-free queue - /// once the free-list naming them is durable. Callers set this to the - /// session-start `next_page_id`, so only pages bump-allocated within the - /// session are recyclable immediately. - pub(super) reuse_threshold: u64, /// Leaves modified during this write session but not yet promoted via /// `CoW` to a fresh page. Keyed by the leaf's current `page_id` as referenced /// by the tree spine. All mutations happen in place; encode + spine @@ -146,18 +148,18 @@ pub struct BTree { /// parallel-AEAD flush. In-place mutation by subsequent puts targeting /// the same leaf is allowed; no further allocation needed. pub(super) fresh_leaves: FxHashMap, - /// Cross-commit pool of reusable page IDs, shared (via `Arc`) across the - /// main, catalog, and history `BTree`s of one `Db`. Allocation pops from - /// here before bumping `next_page_id`, recycling pages freed by earlier - /// commits (once no reader or retained-history root can observe them) so - /// the file stays bounded under sustained writes. `None` for callers that - /// haven't wired a shared cache (e.g. compaction's repack trees), which - /// keep bump-only allocation. - pub(super) free_page_cache: Option>>>, - /// Sink recording page ids drawn from `free_page_cache` and reused this - /// session. The commit path removes them from the durable free-list (they - /// now hold live committed data). Shared (via `Arc`) across the txn's trees. - pub(super) free_page_consumed: Option>>>, + /// Where this tree may allocate from beyond bumping `next_page_id`: the + /// session's reuse threshold and its shared free-page cache, supplied as + /// one value at [`Self::open_session`] and never afterwards. `None` for + /// trees opened outside a write session (readers, compaction's repack + /// trees), which bump-allocate and may recycle anything they free — they + /// own every page they touch. + /// + /// One field rather than three because [`Self::free_page`] reads all of it + /// on the first free: a threshold arriving late would misclassify pages + /// already banked, and a cache without its sink would hand out pages whose + /// draw nothing recorded. Neither state is constructible. + pub(super) page_source: Option, /// Last key successfully appended via [`Self::put_append`]. Used to /// enforce the monotonic-key invariant on subsequent calls and to /// invalidate the cached path when any non-append mutation (regular @@ -173,12 +175,53 @@ pub struct BTree { } impl BTree { + /// Open a tree that allocates only by bumping its own cursor and may + /// recycle anything it frees. For a tree in a write session — one sharing + /// the `Db`'s free-page cache, and holding back pages a reader snapshot or + /// the last durable header may still name — use [`Self::open_session`]. pub fn open( pager: Arc>, realm_id: RealmId, root_page_id: u64, next_page_id: u64, page_size: usize, + ) -> Self { + Self::new(pager, realm_id, root_page_id, next_page_id, page_size, None) + } + + /// Open a tree belonging to `source`'s write session. + /// + /// The source is supplied here and nowhere else. [`Self::free_page`] + /// decides reuse eligibility at the moment of the free, so a threshold or a + /// consumed-page sink that could arrive later would leave pages classified + /// against state that has since changed; taking it at construction is what + /// rules that out rather than asking callers to sequence their calls + /// correctly. + pub fn open_session( + pager: Arc>, + realm_id: RealmId, + root_page_id: u64, + next_page_id: u64, + page_size: usize, + source: PageSource, + ) -> Self { + Self::new( + pager, + realm_id, + root_page_id, + next_page_id, + page_size, + Some(source), + ) + } + + fn new( + pager: Arc>, + realm_id: RealmId, + root_page_id: u64, + next_page_id: u64, + page_size: usize, + page_source: Option, ) -> Self { let next = next_page_id.max(4); Self { @@ -187,49 +230,18 @@ impl BTree { root_page_id, next_page_id: next, freed: Vec::new(), + freed_reusable: Vec::new(), page_size, - reuse_threshold: 0, + page_source, dirty_leaves: FxHashMap::default(), scheduled_frees: Vec::new(), dirty_parent_paths: FxHashMap::default(), fresh_leaves: FxHashMap::default(), - free_page_cache: None, - free_page_consumed: None, append_last_key: None, append_cached_path: None, } } - /// Set the reuse threshold. Any freed page with `page_id < threshold` will - /// not be recycled within this session; it goes to `self.freed` for later - /// deferred-queue promotion. Call with the session-start `next_page_id`: - /// pages below it are still referenced by the last durable header and must - /// keep their on-disk bytes until the header that frees them is durable. - pub fn set_reuse_threshold(&mut self, threshold: u64) { - self.reuse_threshold = threshold; - } - - /// Wire in the `Db`'s shared free-page cache. After this call, - /// `allocate_page` pops from the shared pool before bumping `next_page_id`. - /// The pool is loaded at `begin_write` with the floor-safe pages of the - /// bounded free-list window the following commit rewrites, so recycling - /// from it is always snapshot-safe. - /// - /// The pool is draw-only: nothing here may push into it. The commit deletes - /// a recycled id from the chain by finding it in that window, so an id - /// pushed in from elsewhere would be handed out while the unscanned tail - /// still named it — the same page id given to two live structures. - pub fn set_free_page_cache(&mut self, cache: Arc>>) { - self.free_page_cache = Some(cache); - } - - /// Wire the shared sink that records cache pages reused this session, so the - /// commit path can remove them from the durable free-list. Set alongside - /// [`Self::set_free_page_cache`]. - pub fn set_free_page_consumed(&mut self, consumed: Arc>>) { - self.free_page_consumed = Some(consumed); - } - #[must_use] pub fn root_page_id(&self) -> u64 { self.root_page_id @@ -256,50 +268,36 @@ impl BTree { // clears it (it leaves via `drain_freed` at commit instead). // // Exception: a below-threshold page originally drawn from the shared - // cache this session (recorded in `free_page_consumed`) is free per - // the last durable header regardless of what this session wrote to it - // since, so recycling it again in-session is crash-safe — a failed or - // torn commit leaves the durable header referencing none of its + // cache this session (recorded in the source's consumed sink) is free + // per the last durable header regardless of what this session wrote to + // it since, so recycling it again in-session is crash-safe — a failed + // or torn commit leaves the durable header referencing none of its // content. Without this, every cache-drawn page freed by a later // in-session split is burned for the rest of the txn and allocation // falls through to bump growth. - if self.reuse_threshold == 0 { - if let Some(id) = self.freed.pop() { - assert!( - id >= 4, - "allocate_page recycled reserved page {id} from freed" - ); - return id; - } - } else { - let consumed = self.free_page_consumed.as_ref().map(|c| c.lock()); - let pos = self.freed.iter().rposition(|&id| { - id >= self.reuse_threshold || consumed.as_ref().is_some_and(|c| c.contains(&id)) - }); - drop(consumed); - if let Some(pos) = pos { - let id = self.freed.remove(pos); - assert!( - id >= 4, - "allocate_page recycled reserved page {id} from freed" - ); - return id; - } + // + // `free_page` has already applied that rule to each freed page, so the + // eligible ones are exactly `freed_reusable` and this is a pop, not a + // search. + if let Some(id) = self.freed_reusable.pop() { + assert!( + id >= 4, + "allocate_page recycled reserved page {id} from freed" + ); + return id; } // Then draw from the shared cross-commit cache. It is loaded at txn // begin with *only* free-list pages below the reclamation floor — pages // no live reader and no retained-history root can observe — so reusing // them is safe regardless of `reuse_threshold`. Record each draw so the // commit path removes it from the durable free-list. - if let Some(cache) = &self.free_page_cache { - if let Some(id) = cache.lock().pop() { + if let Some(source) = &self.page_source { + if let Some(id) = source.cache.lock().pop() { assert!( id >= 4, "allocate_page recycled reserved page {id} from free-list cache" ); - if let Some(consumed) = &self.free_page_consumed { - consumed.lock().push(id); - } + source.consumed.record(id); return id; } } @@ -320,7 +318,28 @@ impl BTree { page_id >= 4, "free_page called on reserved page {page_id} (use-after-free / wild pointer)" ); - self.freed.push(page_id); + if self.is_reusable_in_session(page_id) { + self.freed_reusable.push(page_id); + } else { + self.freed.push(page_id); + } + } + + /// Whether a page freed now may be recycled within this session. See + /// [`Self::allocate_page`] for the reasoning behind each arm: below the + /// threshold a page may still be live in the last durable header or a + /// pinned reader's snapshot, unless it was drawn from the shared free-page + /// cache this session, in which case the durable header references none of + /// its content. + /// + /// A tree with no page source holds nothing back: it allocated every page + /// it can free, so no snapshot and no durable header names them. + fn is_reusable_in_session(&self, page_id: u64) -> bool { + let Some(source) = &self.page_source else { + return true; + }; + // A zero threshold needs no special case: every page id clears it. + page_id >= source.reuse_threshold || source.consumed.contains(page_id) } pub(super) fn validate_insert_record_fits(&self, key: &[u8], value: &[u8]) -> Result<()> { @@ -463,3 +482,153 @@ impl BTree { .await } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use super::BTree; + use crate::RealmId; + use crate::btree::tree::page_source::{ConsumedPages, PageSource}; + use crate::crypto::CipherId; + use crate::crypto::kdf::derive_mk; + use crate::pager::{Pager, PagerConfig}; + use crate::vfs::memory::MemVfs; + + const PAGE: usize = 4096; + + /// A session source over an empty cache, with its sink returned so a test + /// can seed pages the allocator is to treat as cache-drawn. + fn session(reuse_threshold: u64) -> (PageSource, Arc) { + let consumed = ConsumedPages::new(); + let source = PageSource::new( + reuse_threshold, + Arc::new(parking_lot::Mutex::new(Vec::new())), + &consumed, + ); + (source, consumed) + } + + async fn fresh_tree(source: Option) -> BTree { + let mk = derive_mk(&[1u8; 32], &[0u8; 16], 0).unwrap(); + let cfg = PagerConfig { + page_size: PAGE, + buffer_pool_pages: 256, + segment_cache_pages: 16, + cipher_id: CipherId::Aes256Gcm, + mk_epoch: 0, + main_db_file_id: [0xAB; 16], + main_db_path: "/main.db".into(), + anchor_budget: 100_000_000, + dek_lru_capacity: 16, + observer_retry_count: 0, + metrics_enabled: true, + }; + let pager = Arc::new(Pager::open(MemVfs::new(), mk, cfg).await.unwrap()); + let realm = RealmId::new([1; 16]); + match source { + Some(source) => BTree::open_session(pager, realm, 0, 4, PAGE, source), + None => BTree::open(pager, realm, 0, 4, PAGE), + } + } + + /// Below-threshold frees must not be recycled in-session, above-threshold + /// ones must be, and a cache-drawn page recorded in the consumed sink must + /// be recyclable even below the threshold. Same rule the per-allocation + /// scan used to apply; asserted here now that `free_page` decides it once. + #[tokio::test(flavor = "current_thread")] + async fn reuse_eligibility_matches_the_threshold_rule() { + let (source, consumed) = session(1000); + let mut tree = fresh_tree(Some(source)).await; + + // Below threshold and never drawn from the cache: not recyclable now. + tree.free_page(500); + assert_eq!(tree.allocate_page(), 4, "bumped, not recycled"); + + // At or above the threshold: recyclable immediately. + tree.free_page(1500); + assert_eq!(tree.allocate_page(), 1500); + + // Below threshold but drawn from the shared cache this session. + consumed.handle().record(600); + tree.free_page(600); + assert_eq!(tree.allocate_page(), 600); + + // The one page held back is still reported as freed this session. + assert_eq!(tree.drain_freed(), vec![500]); + } + + /// A tree opened without a page source owns every page it can free, so it + /// recycles them all and never holds one back. + #[tokio::test(flavor = "current_thread")] + async fn a_tree_without_a_page_source_recycles_everything() { + let mut tree = fresh_tree(None).await; + tree.free_page(9); + assert_eq!(tree.allocate_page(), 9); + assert!(tree.drain_freed().is_empty()); + } + + /// Eligibility is settled at the moment of the free and does not change + /// afterwards. A page held back stays held back even if the same id is + /// later recorded as cache-drawn — the deciding state only ever grows, so + /// deciding early can only be conservative, never unsafe. The old + /// per-allocation scan would have promoted this page; that difference is + /// intended, and this pins it. + #[tokio::test(flavor = "current_thread")] + async fn a_page_held_back_is_not_promoted_by_a_later_record() { + let (source, consumed) = session(1000); + let mut tree = fresh_tree(Some(source)).await; + + tree.free_page(500); + consumed.handle().record(500); + + assert_eq!(tree.allocate_page(), 4, "bumped, not promoted"); + assert_eq!(tree.drain_freed(), vec![500]); + } + + /// Neither freeing nor allocating may scan a collection that grows with the + /// flush. A hang detector, not a benchmark — the only wall-clock assertion + /// in `src/`, and the bound is ~1000x the fixed cost, so only a return to a + /// linear scan can exceed it. Two such scans are in scope: allocation over + /// `freed`, and the free-time eligibility test over `free_page_consumed`. + /// + /// `consumed` is therefore seeded at the size the free-list window permits + /// (`WINDOW_PAGES × chain_capacity(page_size)`), not at a token count: that + /// is what bounds it in a real flush, and a smaller seed cannot see the + /// free-time term at all. + #[tokio::test(flavor = "current_thread")] + async fn allocation_cost_does_not_grow_with_the_freed_list() { + const N: u64 = 20_000; + let window = crate::pager::freelist::WINDOW_PAGES + * crate::pager::freelist::layout::chain_capacity(PAGE); + let (source, consumed) = session(N * 2); + let mut tree = fresh_tree(Some(source)).await; + + // Seed the sink at window scale *before* the frees, so every free below + // pays the eligibility test against a full-size `consumed`. + { + let sink = consumed.handle(); + for id in (N * 4)..(N * 4 + window as u64) { + sink.record(id); + } + } + + // Every one of these is below the threshold and absent from `consumed`, + // so it is ineligible: before the fix each allocation scanned all of + // them and, per entry, all of `consumed`. + for id in 4..N { + tree.free_page(id); + } + + let start = std::time::Instant::now(); + for _ in 0..N { + tree.allocate_page(); + } + let elapsed = start.elapsed(); + assert!( + elapsed < std::time::Duration::from_secs(10), + "{N} allocations against {N} held-back frees took {elapsed:?} — \ + allocation is scanning the freed list again" + ); + } +} diff --git a/src/btree/tree/flush.rs b/src/btree/tree/flush.rs index d7fe859..9069fb0 100644 --- a/src/btree/tree/flush.rs +++ b/src/btree/tree/flush.rs @@ -100,10 +100,13 @@ impl BTree { /// Drain and return all `page_ids` that were freed during this tree's /// mutation session, plus any leaf pages scheduled for freeing at the - /// upcoming `flush`. After this call, both `self.freed` and - /// `self.scheduled_frees` are empty. + /// upcoming `flush`. After this call, `self.freed`, `self.freed_reusable` + /// and `self.scheduled_frees` are all empty — a page eligible for + /// in-session reuse that no allocation claimed is still a page this session + /// freed, and the deferred-free queue must hear about it. pub fn drain_freed(&mut self) -> Vec { let mut out = std::mem::take(&mut self.freed); + out.append(&mut self.freed_reusable); out.append(&mut self.scheduled_frees); out } diff --git a/src/btree/tree/mod.rs b/src/btree/tree/mod.rs index 7ee9580..5023454 100644 --- a/src/btree/tree/mod.rs +++ b/src/btree/tree/mod.rs @@ -5,8 +5,10 @@ pub(crate) mod core; pub(crate) mod flush; pub(crate) mod maintenance; pub(crate) mod navigate; +pub(crate) mod page_source; pub(crate) mod read; pub(crate) mod scan; pub(crate) mod write; pub use core::BTree; +pub use page_source::{ConsumedPages, PageSource}; diff --git a/src/btree/tree/page_source.rs b/src/btree/tree/page_source.rs new file mode 100644 index 0000000..9d12ca6 --- /dev/null +++ b/src/btree/tree/page_source.rs @@ -0,0 +1,133 @@ +//! Where a write session's pages come from, supplied to a tree as one value. +//! +//! Reuse eligibility is decided when a page is freed, not when one is +//! allocated (see [`BTree::free_page`](super::core::BTree::free_page)). That is +//! only sound while every input to the decision is already in place at the +//! first free and cannot move afterwards, so the inputs travel together in +//! [`PageSource`] and are handed over once, at +//! [`BTree::open_session`](super::core::BTree::open_session). There is no +//! setter for any of them: a tree either has a session's page source for its +//! whole life or has none and bump-allocates. +//! +//! [`ConsumedPages`] carries the other half of that guarantee. A page recorded +//! there is eligible for in-session reuse, and pages already banked on the +//! strength of that record cannot be un-banked — so the sink must not shrink +//! while a session is running. Trees therefore hold a [`ConsumedHandle`], which +//! can record and test but cannot empty; emptying is [`ConsumedPages::take`], +//! reachable only from whoever owns the sink, at a transaction boundary. + +use std::sync::Arc; + +use rustc_hash::FxHashSet; + +/// Page ids the allocator drew from the shared free-page cache and reused this +/// session. The commit path removes them from the durable free-list — they now +/// hold live committed data — and [`BTree::free_page`] consults them: a page +/// drawn from the cache is free as of the last durable header regardless of +/// what this session wrote to it, so it may be recycled again in-session even +/// from below the reuse threshold. +/// +/// A set, not a list: membership is tested once per freed page, and the size is +/// bounded by the free-list window (`WINDOW_PAGES × chain_capacity(page_size)`) +/// rather than by anything small — a linear test would keep the per-free cost +/// proportional to that window. Nothing reads insertion order or needs +/// duplicates. +/// +/// [`BTree::free_page`]: super::core::BTree::free_page +#[derive(Debug, Default)] +pub struct ConsumedPages(parking_lot::Mutex>); + +impl ConsumedPages { + /// An empty sink, ready to be shared with a session's trees via + /// [`Self::handle`]. + #[must_use] + pub fn new() -> Arc { + Arc::new(Self::default()) + } + + /// A tree's view of this sink: record and test, never empty. Handing trees + /// this instead of the `Arc` is what makes the sink's monotonicity within a + /// session a property of the types rather than of caller discipline. + #[must_use] + pub fn handle(self: &Arc) -> ConsumedHandle { + ConsumedHandle(Arc::clone(self)) + } + + /// Empty the sink and return what it held. + /// + /// A transaction-boundary operation, and the only way to empty it. Calling + /// this while a session's trees are still allocating would strip the record + /// that made already-banked pages eligible, leaving them banked on a reason + /// that no longer exists — which is why no [`ConsumedHandle`] can reach it. + pub fn take(&self) -> FxHashSet { + std::mem::take(&mut *self.0.lock()) + } +} + +/// A tree's handle on a session's [`ConsumedPages`]. Clone-shared across the +/// trees of one transaction. +#[derive(Clone, Debug)] +pub struct ConsumedHandle(Arc); + +impl ConsumedHandle { + pub(crate) fn record(&self, page_id: u64) { + self.0.0.lock().insert(page_id); + } + + pub(crate) fn contains(&self, page_id: u64) -> bool { + self.0.0.lock().contains(&page_id) + } +} + +/// The pages a write session may allocate from, beyond bumping its own cursor. +/// +/// Constructed once per session and cloned to each of its trees, so a tree can +/// never observe a partly-wired source — a cache without the sink that records +/// draws from it, or a threshold arriving after pages have already been +/// classified against the default. +#[derive(Clone, Debug)] +pub struct PageSource { + /// Minimum `page_id` that may be recycled from within this session. Pages + /// below it existed before the session began: they are still referenced by + /// the last durable header (a copy-on-write free does not unreference them + /// on disk until the *next* header lands) and may also be pinned by reader + /// snapshots, so their bytes must survive until the header that frees them + /// is durable. They re-enter circulation through the deferred-free queue. + /// + /// Callers pass the session-start `next_page_id`, so only pages + /// bump-allocated within the session are recyclable immediately. Zero means + /// every freed page is (the commit-history tree, whose pages no reader + /// snapshot names). + pub(crate) reuse_threshold: u64, + /// Cross-commit pool of reusable page ids, shared across the main, catalog + /// and history trees of one `Db`. Loaded at `begin_write` with *only* + /// free-list pages below the reclamation floor — pages no live reader and + /// no retained-history root can observe — so drawing from it is safe + /// regardless of `reuse_threshold`. + /// + /// Draw-only: nothing may push into it. The commit deletes a recycled id + /// from the chain by finding it in the scanned window, so an id pushed in + /// from elsewhere would be handed out while the unscanned tail still named + /// it — the same page id given to two live structures. + pub(crate) cache: Arc>>, + /// Where draws from `cache` are recorded. + pub(crate) consumed: ConsumedHandle, +} + +impl PageSource { + /// Bind a session's allocation state to one value. `reuse_threshold` is the + /// session-start `next_page_id` for trees a reader may hold a snapshot of, + /// and zero for those none can. + #[must_use] + pub fn new( + reuse_threshold: u64, + cache: Arc>>, + consumed: &Arc, + ) -> Self { + Self { + reuse_threshold, + cache, + consumed: consumed.handle(), + } + } +} diff --git a/src/txn/db/catalog.rs b/src/txn/db/catalog.rs index f22184a..39a13ea 100644 --- a/src/txn/db/catalog.rs +++ b/src/txn/db/catalog.rs @@ -232,13 +232,6 @@ impl Db { readers.iter().map(|r| r.commit_id.0).min() }; - let mut hist_tree = BTree::open( - self.pager.clone(), - self.realm_id, - state.commit_history_root_page_id, - state.next_page_id, - self.page_size, - ); // The commit-history tree is not part of any reader's pinned snapshot // (readers track the data and catalog roots, never the history root), so // every page its copy-on-write/prune frees is immediately reusable @@ -250,9 +243,18 @@ impl Db { // pushes into it. Its own frees leave through `drain_freed` into the // commit's entry set instead, so no page reaches an allocator without // the window entry that names it having been located first. - hist_tree.set_reuse_threshold(0); - hist_tree.set_free_page_cache(self.free_page_cache.clone()); - hist_tree.set_free_page_consumed(self.free_page_consumed.clone()); + let mut hist_tree = BTree::open_session( + self.pager.clone(), + self.realm_id, + state.commit_history_root_page_id, + state.next_page_id, + self.page_size, + crate::btree::PageSource::new( + 0, + self.free_page_cache.clone(), + &self.free_page_consumed, + ), + ); // Insert the new entry. let key = new_commit_id.to_be_bytes().to_vec(); diff --git a/src/txn/db/core.rs b/src/txn/db/core.rs index 02c7249..6aad0b3 100644 --- a/src/txn/db/core.rs +++ b/src/txn/db/core.rs @@ -231,10 +231,15 @@ pub struct Db { /// history) draw from the same pool. Cleared by `compact_now`'s full repack, /// which relocates pages and invalidates every cached id. pub(crate) free_page_cache: Arc>>, - /// Per-txn sink (cleared at `begin_write`) recording page ids the allocator + /// Per-txn sink (emptied at `begin_write`) recording page ids the allocator /// drew from `free_page_cache`. The commit path removes them from the /// durable free-list — they now hold live committed data. - pub(crate) free_page_consumed: Arc>>, + /// + /// The trees see it through a handle that cannot empty it: a page banked + /// for in-session reuse on the strength of a record here must not have that + /// record withdrawn under it, so emptying belongs to the transaction + /// boundaries alone. + pub(crate) free_page_consumed: Arc, #[cfg(test)] pub(crate) visibility_test_hook: parking_lot::Mutex>>, #[cfg(test)] diff --git a/src/txn/db/open/create.rs b/src/txn/db/open/create.rs index 491d8fe..50cc421 100644 --- a/src/txn/db/open/create.rs +++ b/src/txn/db/open/create.rs @@ -130,7 +130,7 @@ impl Db { })), poisoned_commit: parking_lot::Mutex::new(None), free_page_cache: Arc::new(parking_lot::Mutex::new(Vec::new())), - free_page_consumed: Arc::new(parking_lot::Mutex::new(Vec::new())), + free_page_consumed: crate::btree::ConsumedPages::new(), #[cfg(test)] visibility_test_hook: parking_lot::Mutex::new(None), #[cfg(test)] diff --git a/src/txn/db/open/existing.rs b/src/txn/db/open/existing.rs index 9e9e332..6ca0cb6 100644 --- a/src/txn/db/open/existing.rs +++ b/src/txn/db/open/existing.rs @@ -321,7 +321,7 @@ impl Db { })), poisoned_commit: parking_lot::Mutex::new(None), free_page_cache: Arc::new(parking_lot::Mutex::new(Vec::new())), - free_page_consumed: Arc::new(parking_lot::Mutex::new(Vec::new())), + free_page_consumed: crate::btree::ConsumedPages::new(), #[cfg(test)] visibility_test_hook: parking_lot::Mutex::new(None), #[cfg(test)] diff --git a/src/txn/write/commit.rs b/src/txn/write/commit.rs index 0dce921..ed2e831 100644 --- a/src/txn/write/commit.rs +++ b/src/txn/write/commit.rs @@ -144,9 +144,9 @@ impl WriteTxn<'_, V> { .collect(); // Pages the allocator reused from the cache this txn — they were free // before, now hold live committed data, so they leave the free-list. - let consumed: HashSet = std::mem::take(&mut *self.db.free_page_consumed.lock()) - .into_iter() - .collect(); + // Emptied here rather than mid-session: the trees have all drained, so + // nothing can still be banking pages against it. + let consumed = self.db.free_page_consumed.take(); // The scanned window after this commit: its entries minus the ones // reused, kept with their original freeing-commit tag. Every id in diff --git a/src/txn/write/txn.rs b/src/txn/write/txn.rs index 8f2db11..e9f01a9 100644 --- a/src/txn/write/txn.rs +++ b/src/txn/write/txn.rs @@ -159,28 +159,34 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> { } } } - db.free_page_consumed.lock().clear(); + // Emptied here, at the boundary, and not again until commit: the trees + // decide reuse eligibility against this sink as they free, so it must + // only grow for as long as the session runs. + db.free_page_consumed.take(); - let mut btree = BTree::open( + // One source for every tree in the session, so all three draw from the + // same cache and record into the same sink. + let source = crate::btree::PageSource::new( + reuse_threshold, + db.free_page_cache.clone(), + &db.free_page_consumed, + ); + let btree = BTree::open_session( db.pager.clone(), db.realm_id, guard.root_page_id, guard.next_page_id, db.page_size, + source.clone(), ); - btree.set_reuse_threshold(reuse_threshold); - btree.set_free_page_cache(db.free_page_cache.clone()); - btree.set_free_page_consumed(db.free_page_consumed.clone()); - let mut catalog_tree = BTree::open( + let catalog_tree = BTree::open_session( db.pager.clone(), db.realm_id, guard.catalog_root_page_id, guard.next_page_id, db.page_size, + source, ); - catalog_tree.set_reuse_threshold(reuse_threshold); - catalog_tree.set_free_page_cache(db.free_page_cache.clone()); - catalog_tree.set_free_page_consumed(db.free_page_consumed.clone()); // Assign a txn_seq starting from 1: fetch_add returns the old value (0 // for the first call), so we add 1 to produce 1-based ids. let txn_seq = db.txn_seq.fetch_add(1, Ordering::Relaxed) + 1;