From 7487e5f4bb911938b0a9ea6cb90541ee18ec7521 Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Thu, 30 Jul 2026 09:54:08 +0800 Subject: [PATCH 1/3] perf(btree): decide page-reuse eligibility at free time, not per allocation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit allocate_page evaluated the in-session reuse rule by scanning the whole `freed` list and, for each entry, all of `free_page_consumed` — both plain Vecs, so `contains` is linear. Because `reuse_threshold` is the session-start `next_page_id`, every page freed during the txn sits below it and the cheap first disjunct is false for nearly every entry, so the full O(|freed| x |consumed|) scan ran on every single page allocation, with both vectors growing monotonically through the flush. The result is unbounded work rather than a deadlock: a server was found pinning one core at 99.4% for 84 minutes inside a single batch_write, 5048s of user CPU against 2.9s of system time and 104 read syscalls for the whole process lifetime, with 99.51% of perf samples in allocate_page. Both inputs to the eligibility decision are monotonic — reuse_threshold is fixed for the session and free_page_consumed only grows — so deciding once in free_page is equivalent to deciding per allocation. Eligible pages go to a second `freed_reusable` list that allocate_page pops in O(1). set_reuse_threshold re-partitions, for callers that set it after a free (normally a walk over two empty vectors), and drain_freed drains both lists so an eligible page no allocation claimed still reaches the deferred-free queue. Three tests cover the eligibility rule, post-hoc threshold setting, and a hang detector for the scan's return: with the old logic that test does not finish in 300s, with this change it completes in 0.00s. --- CHANGELOG.md | 6 ++ src/btree/tree/core.rs | 183 +++++++++++++++++++++++++++++++++++----- src/btree/tree/flush.rs | 7 +- 3 files changed, 171 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b82abe..8db387b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- **Page allocation no longer rescans the freed list.** `allocate_page` decided in-session reuse eligibility per call, scanning every page freed so far and, for each, every page drawn from the shared free-page cache — both `Vec`, so the membership test was linear. Since the reuse threshold is the session-start `next_page_id`, every page freed during a transaction fell below it and took the slow arm, making allocation quadratic in the pages a flush touches: a flush large enough would occupy a core indefinitely without completing. Eligibility is now decided once, when the page is freed, and allocation is a pop. Behaviour is unchanged — both inputs to the decision are monotonic within a session, so deciding early is equivalent. + ## [0.1.0] - 2026-07-28 The first release. Pre-releases were published as `0.1.0-beta.N`; the entries below describe `0.1.0` as a whole rather than deltas against a shipped version, since none exists yet. diff --git a/src/btree/tree/core.rs b/src/btree/tree/core.rs index d7485d3..e709312 100644 --- a/src/btree/tree/core.rs +++ b/src/btree/tree/core.rs @@ -108,6 +108,17 @@ 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 both inputs to that decision are + /// monotonic: `reuse_threshold` is fixed for the session and + /// `free_page_consumed` only grows. Deciding it here keeps allocation O(1); + /// evaluating it per allocation instead meant rescanning `freed` against + /// `free_page_consumed` 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 @@ -187,6 +198,7 @@ impl BTree { root_page_id, next_page_id: next, freed: Vec::new(), + freed_reusable: Vec::new(), page_size, reuse_threshold: 0, dirty_leaves: FxHashMap::default(), @@ -205,8 +217,21 @@ impl BTree { /// 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. + /// + /// Re-partitions anything already freed, since eligibility is decided at + /// free time: callers normally set the threshold before the first free, so + /// this is a no-op walk over two empty vectors. pub fn set_reuse_threshold(&mut self, threshold: u64) { self.reuse_threshold = threshold; + let mut pending = std::mem::take(&mut self.freed); + pending.append(&mut self.freed_reusable); + for id in pending { + if self.is_reusable_in_session(id) { + self.freed_reusable.push(id); + } else { + self.freed.push(id); + } + } } /// Wire in the `Db`'s shared free-page cache. After this call, @@ -263,28 +288,15 @@ impl BTree { // 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 @@ -320,7 +332,26 @@ 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. + fn is_reusable_in_session(&self, page_id: u64) -> bool { + if self.reuse_threshold == 0 || page_id >= self.reuse_threshold { + return true; + } + self.free_page_consumed + .as_ref() + .is_some_and(|c| c.lock().contains(&page_id)) } pub(super) fn validate_insert_record_fits(&self, key: &[u8], value: &[u8]) -> Result<()> { @@ -463,3 +494,109 @@ impl BTree { .await } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use super::BTree; + use crate::RealmId; + use crate::crypto::CipherId; + use crate::crypto::kdf::derive_mk; + use crate::pager::{Pager, PagerConfig}; + use crate::vfs::memory::MemVfs; + + const PAGE: usize = 4096; + + async fn fresh_tree() -> 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()); + BTree::open(pager, RealmId::new([1; 16]), 0, 4, PAGE) + } + + /// Below-threshold frees must not be recycled in-session, above-threshold + /// ones must be, and a cache-drawn page recorded in `free_page_consumed` + /// 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 mut tree = fresh_tree().await; + let consumed = Arc::new(parking_lot::Mutex::new(Vec::new())); + tree.set_free_page_consumed(Arc::clone(&consumed)); + tree.set_reuse_threshold(1000); + + // 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.lock().push(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 page freed before the threshold was set is classified when it is set. + #[tokio::test(flavor = "current_thread")] + async fn threshold_set_after_a_free_repartitions() { + let mut tree = fresh_tree().await; + tree.free_page(500); + tree.free_page(1500); + tree.set_reuse_threshold(1000); + assert_eq!(tree.allocate_page(), 1500, "above threshold stays reusable"); + assert_eq!(tree.allocate_page(), 4, "below threshold was held back"); + assert_eq!(tree.drain_freed(), vec![500]); + } + + /// Allocation must not rescan the freed list. A hang detector, not a + /// benchmark: the bound is ~1000x the fixed cost, and only a return to a + /// per-allocation scan over `freed` x `free_page_consumed` can exceed it. + /// That scan is what pinned a core for 84 minutes inside one flush. + #[tokio::test(flavor = "current_thread")] + async fn allocation_cost_does_not_grow_with_the_freed_list() { + const N: u64 = 20_000; + let mut tree = fresh_tree().await; + let consumed = Arc::new(parking_lot::Mutex::new(Vec::new())); + tree.set_free_page_consumed(Arc::clone(&consumed)); + tree.set_reuse_threshold(N * 2); + + // Every one of these is ineligible, so each allocation below had to + // scan all of them (and, per entry, all of `consumed`) before the fix. + for id in 4..N { + tree.free_page(id); + } + for id in N..(N + 500) { + consumed.lock().push(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 } From fdfe9b4b35abcb5c0fdc50112ae6b4d4b6c6c6ed Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Thu, 30 Jul 2026 12:15:03 +0800 Subject: [PATCH 2/3] perf(btree): make the reused-page sink a set so the free-time check is O(1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The free-time eligibility test introduced with `freed_reusable` still scanned `free_page_consumed` linearly, and a copy-on-write free of a pre-existing page is below the reuse threshold by construction — so the dominant case in a flush kept paying O(frees x |consumed|), with |consumed| bounded by the free-list window rather than by anything small. Making the shared sink an FxHashSet drops that to O(1) per free: the commit path already collapsed it into a set and only tests membership, begin_write only clears it, and nothing reads its order. The hang detector now seeds `consumed` at window scale (WINDOW_PAGES x chain_capacity(page_size)) instead of a token 500 entries; at the old size it could not observe this term at all. set_reuse_threshold now debug_asserts that nothing has been freed yet, as do the two sibling setters, since eligibility is decided at free time and a threshold arriving afterwards would judge already-classified pages against the wrong one. That makes the previous re-partition branch unreachable, so it and the test covering it are gone rather than left as dead code. Also drops the `reuse_threshold == 0` disjunct, which never decided anything: every page id is >= 0. --- CHANGELOG.md | 6 --- src/btree/tree/core.rs | 98 +++++++++++++++++++++---------------- src/txn/db/core.rs | 2 +- src/txn/db/open/create.rs | 2 +- src/txn/db/open/existing.rs | 2 +- src/txn/write/commit.rs | 4 +- 6 files changed, 60 insertions(+), 54 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8db387b..1b82abe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,12 +2,6 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] - -### Fixed - -- **Page allocation no longer rescans the freed list.** `allocate_page` decided in-session reuse eligibility per call, scanning every page freed so far and, for each, every page drawn from the shared free-page cache — both `Vec`, so the membership test was linear. Since the reuse threshold is the session-start `next_page_id`, every page freed during a transaction fell below it and took the slow arm, making allocation quadratic in the pages a flush touches: a flush large enough would occupy a core indefinitely without completing. Eligibility is now decided once, when the page is freed, and allocation is a pop. Behaviour is unchanged — both inputs to the decision are monotonic within a session, so deciding early is equivalent. - ## [0.1.0] - 2026-07-28 The first release. Pre-releases were published as `0.1.0-beta.N`; the entries below describe `0.1.0` as a whole rather than deltas against a shipped version, since none exists yet. diff --git a/src/btree/tree/core.rs b/src/btree/tree/core.rs index e709312..213b2e8 100644 --- a/src/btree/tree/core.rs +++ b/src/btree/tree/core.rs @@ -3,7 +3,7 @@ use std::collections::BTreeSet; use std::sync::Arc; -use rustc_hash::FxHashMap; +use rustc_hash::{FxHashMap, FxHashSet}; use crate::errors::PagedbError; use crate::pager::format::page_kind::PageKind; @@ -168,7 +168,13 @@ pub struct BTree { /// 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>>>, + /// + /// A set, not a list: [`Self::free_page`] tests membership 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 here would keep the per-free cost proportional to + /// the window. Nothing reads the insertion order or needs duplicates. + pub(super) free_page_consumed: 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 @@ -218,20 +224,16 @@ impl BTree { /// 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. /// - /// Re-partitions anything already freed, since eligibility is decided at - /// free time: callers normally set the threshold before the first free, so - /// this is a no-op walk over two empty vectors. + /// Must be called before the first free: [`Self::free_page`] decides reuse + /// eligibility at the moment of the free, so a threshold arriving later + /// would leave already-classified pages judged against the wrong one. Every + /// caller sets it on a fresh tree, which the `debug_assert` states. pub fn set_reuse_threshold(&mut self, threshold: u64) { + debug_assert!( + self.freed.is_empty() && self.freed_reusable.is_empty(), + "set_reuse_threshold after a free: eligibility is decided at free time" + ); self.reuse_threshold = threshold; - let mut pending = std::mem::take(&mut self.freed); - pending.append(&mut self.freed_reusable); - for id in pending { - if self.is_reusable_in_session(id) { - self.freed_reusable.push(id); - } else { - self.freed.push(id); - } - } } /// Wire in the `Db`'s shared free-page cache. After this call, @@ -245,13 +247,21 @@ impl BTree { /// 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>>) { + debug_assert!( + self.freed.is_empty() && self.freed_reusable.is_empty(), + "set_free_page_cache after a free: eligibility is decided at free time" + ); 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>>) { + pub fn set_free_page_consumed(&mut self, consumed: Arc>>) { + debug_assert!( + self.freed.is_empty() && self.freed_reusable.is_empty(), + "set_free_page_consumed after a free: eligibility is decided at free time" + ); self.free_page_consumed = Some(consumed); } @@ -310,7 +320,7 @@ impl BTree { "allocate_page recycled reserved page {id} from free-list cache" ); if let Some(consumed) = &self.free_page_consumed { - consumed.lock().push(id); + consumed.lock().insert(id); } return id; } @@ -346,7 +356,8 @@ impl BTree { /// cache this session, in which case the durable header references none of /// its content. fn is_reusable_in_session(&self, page_id: u64) -> bool { - if self.reuse_threshold == 0 || page_id >= self.reuse_threshold { + // A zero threshold needs no special case: every page id clears it. + if page_id >= self.reuse_threshold { return true; } self.free_page_consumed @@ -499,7 +510,7 @@ impl BTree { mod tests { use std::sync::Arc; - use super::BTree; + use super::{BTree, FxHashSet}; use crate::RealmId; use crate::crypto::CipherId; use crate::crypto::kdf::derive_mk; @@ -534,7 +545,7 @@ mod tests { #[tokio::test(flavor = "current_thread")] async fn reuse_eligibility_matches_the_threshold_rule() { let mut tree = fresh_tree().await; - let consumed = Arc::new(parking_lot::Mutex::new(Vec::new())); + let consumed = Arc::new(parking_lot::Mutex::new(FxHashSet::default())); tree.set_free_page_consumed(Arc::clone(&consumed)); tree.set_reuse_threshold(1000); @@ -547,7 +558,7 @@ mod tests { assert_eq!(tree.allocate_page(), 1500); // Below threshold but drawn from the shared cache this session. - consumed.lock().push(600); + consumed.lock().insert(600); tree.free_page(600); assert_eq!(tree.allocate_page(), 600); @@ -555,38 +566,41 @@ mod tests { assert_eq!(tree.drain_freed(), vec![500]); } - /// A page freed before the threshold was set is classified when it is set. - #[tokio::test(flavor = "current_thread")] - async fn threshold_set_after_a_free_repartitions() { - let mut tree = fresh_tree().await; - tree.free_page(500); - tree.free_page(1500); - tree.set_reuse_threshold(1000); - assert_eq!(tree.allocate_page(), 1500, "above threshold stays reusable"); - assert_eq!(tree.allocate_page(), 4, "below threshold was held back"); - assert_eq!(tree.drain_freed(), vec![500]); - } - - /// Allocation must not rescan the freed list. A hang detector, not a - /// benchmark: the bound is ~1000x the fixed cost, and only a return to a - /// per-allocation scan over `freed` x `free_page_consumed` can exceed it. - /// That scan is what pinned a core for 84 minutes inside one flush. + /// 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 mut tree = fresh_tree().await; - let consumed = Arc::new(parking_lot::Mutex::new(Vec::new())); + let consumed = Arc::new(parking_lot::Mutex::new(FxHashSet::default())); tree.set_free_page_consumed(Arc::clone(&consumed)); tree.set_reuse_threshold(N * 2); - // Every one of these is ineligible, so each allocation below had to - // scan all of them (and, per entry, all of `consumed`) before the fix. + // Seed the sink at window scale *before* the frees, so every free below + // pays the eligibility test against a full-size `consumed`. + { + let mut c = consumed.lock(); + for id in (N * 4)..(N * 4 + window as u64) { + c.insert(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); } - for id in N..(N + 500) { - consumed.lock().push(id); - } let start = std::time::Instant::now(); for _ in 0..N { diff --git a/src/txn/db/core.rs b/src/txn/db/core.rs index 02c7249..6785f51 100644 --- a/src/txn/db/core.rs +++ b/src/txn/db/core.rs @@ -234,7 +234,7 @@ pub struct Db { /// Per-txn sink (cleared 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>>, + 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..62566b5 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: Arc::new(parking_lot::Mutex::new(rustc_hash::FxHashSet::default())), #[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..7b18c20 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: Arc::new(parking_lot::Mutex::new(rustc_hash::FxHashSet::default())), #[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..210e6ee 100644 --- a/src/txn/write/commit.rs +++ b/src/txn/write/commit.rs @@ -144,9 +144,7 @@ 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(); + let consumed = std::mem::take(&mut *self.db.free_page_consumed.lock()); // The scanned window after this commit: its entries minus the ones // reused, kept with their original freeing-commit tag. Every id in From b76708fd2fcfaa28283272806baa986df2b5fb18 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Thu, 30 Jul 2026 17:40:29 +0800 Subject: [PATCH 3/3] refactor(btree): bind a session's page-reuse state into one PageSource allocate_page/free_page decide reuse eligibility from three inputs (reuse_threshold, the shared free-page cache, and the consumed-page sink) that previously arrived through separate setters called on a freshly opened tree, with a debug_assert guarding against calling them out of order or after the first free. Collapsing them into a single PageSource, handed to BTree::open_session once at construction, makes a partially wired tree unconstructable instead of merely asserted against. The consumed-page sink itself becomes a ConsumedPages type: trees hold a ConsumedHandle that can record and test membership but never empty it, since emptying mid-session would strip the record that made already-banked pages eligible. Only the sink's owner, at a transaction boundary, can call take() to drain it. BTree::open remains for trees outside a write session (readers, compaction's repack trees), which still bump-allocate and may recycle anything they free. --- src/btree/mod.rs | 2 +- src/btree/tree/core.rs | 250 ++++++++++++++++++---------------- src/btree/tree/mod.rs | 2 + src/btree/tree/page_source.rs | 133 ++++++++++++++++++ src/txn/db/catalog.rs | 22 +-- src/txn/db/core.rs | 9 +- src/txn/db/open/create.rs | 2 +- src/txn/db/open/existing.rs | 2 +- src/txn/write/commit.rs | 4 +- src/txn/write/txn.rs | 24 ++-- 10 files changed, 309 insertions(+), 141 deletions(-) create mode 100644 src/btree/tree/page_source.rs 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 213b2e8..daa85a8 100644 --- a/src/btree/tree/core.rs +++ b/src/btree/tree/core.rs @@ -3,7 +3,7 @@ use std::collections::BTreeSet; use std::sync::Arc; -use rustc_hash::{FxHashMap, FxHashSet}; +use rustc_hash::FxHashMap; use crate::errors::PagedbError; use crate::pager::format::page_kind::PageKind; @@ -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. @@ -111,25 +112,15 @@ pub struct BTree { /// 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 both inputs to that decision are - /// monotonic: `reuse_threshold` is fixed for the session and - /// `free_page_consumed` only grows. Deciding it here keeps allocation O(1); - /// evaluating it per allocation instead meant rescanning `freed` against - /// `free_page_consumed` 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`. + /// [`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 @@ -157,24 +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. + /// 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. /// - /// A set, not a list: [`Self::free_page`] tests membership 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 here would keep the per-free cost proportional to - /// the window. Nothing reads the insertion order or needs duplicates. - pub(super) free_page_consumed: Option>>>, + /// 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 @@ -190,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 { @@ -206,65 +232,16 @@ impl BTree { 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. - /// - /// Must be called before the first free: [`Self::free_page`] decides reuse - /// eligibility at the moment of the free, so a threshold arriving later - /// would leave already-classified pages judged against the wrong one. Every - /// caller sets it on a fresh tree, which the `debug_assert` states. - pub fn set_reuse_threshold(&mut self, threshold: u64) { - debug_assert!( - self.freed.is_empty() && self.freed_reusable.is_empty(), - "set_reuse_threshold after a free: eligibility is decided at free time" - ); - 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>>) { - debug_assert!( - self.freed.is_empty() && self.freed_reusable.is_empty(), - "set_free_page_cache after a free: eligibility is decided at free time" - ); - 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>>) { - debug_assert!( - self.freed.is_empty() && self.freed_reusable.is_empty(), - "set_free_page_consumed after a free: eligibility is decided at free time" - ); - self.free_page_consumed = Some(consumed); - } - #[must_use] pub fn root_page_id(&self) -> u64 { self.root_page_id @@ -291,13 +268,14 @@ 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. + // // `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. @@ -313,15 +291,13 @@ impl BTree { // 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().insert(id); - } + source.consumed.record(id); return id; } } @@ -355,14 +331,15 @@ impl BTree { /// 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 { - // A zero threshold needs no special case: every page id clears it. - if page_id >= self.reuse_threshold { + let Some(source) = &self.page_source else { return true; - } - self.free_page_consumed - .as_ref() - .is_some_and(|c| c.lock().contains(&page_id)) + }; + // 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<()> { @@ -510,8 +487,9 @@ impl BTree { mod tests { use std::sync::Arc; - use super::{BTree, FxHashSet}; + 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}; @@ -519,7 +497,19 @@ mod tests { const PAGE: usize = 4096; - async fn fresh_tree() -> BTree { + /// 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, @@ -535,19 +525,21 @@ mod tests { metrics_enabled: true, }; let pager = Arc::new(Pager::open(MemVfs::new(), mk, cfg).await.unwrap()); - BTree::open(pager, RealmId::new([1; 16]), 0, 4, PAGE) + 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 `free_page_consumed` - /// must be recyclable even below the threshold. Same rule the per-allocation + /// 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 mut tree = fresh_tree().await; - let consumed = Arc::new(parking_lot::Mutex::new(FxHashSet::default())); - tree.set_free_page_consumed(Arc::clone(&consumed)); - tree.set_reuse_threshold(1000); + 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); @@ -558,7 +550,7 @@ mod tests { assert_eq!(tree.allocate_page(), 1500); // Below threshold but drawn from the shared cache this session. - consumed.lock().insert(600); + consumed.handle().record(600); tree.free_page(600); assert_eq!(tree.allocate_page(), 600); @@ -566,6 +558,34 @@ mod tests { 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 @@ -581,17 +601,15 @@ mod tests { const N: u64 = 20_000; let window = crate::pager::freelist::WINDOW_PAGES * crate::pager::freelist::layout::chain_capacity(PAGE); - let mut tree = fresh_tree().await; - let consumed = Arc::new(parking_lot::Mutex::new(FxHashSet::default())); - tree.set_free_page_consumed(Arc::clone(&consumed)); - tree.set_reuse_threshold(N * 2); + 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 mut c = consumed.lock(); + let sink = consumed.handle(); for id in (N * 4)..(N * 4 + window as u64) { - c.insert(id); + sink.record(id); } } 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 6785f51..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 62566b5..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(rustc_hash::FxHashSet::default())), + 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 7b18c20..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(rustc_hash::FxHashSet::default())), + 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 210e6ee..ed2e831 100644 --- a/src/txn/write/commit.rs +++ b/src/txn/write/commit.rs @@ -144,7 +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 = std::mem::take(&mut *self.db.free_page_consumed.lock()); + // 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;