Skip to main content

reddb_engine/storage/
page_cache.rs

1//! SIEVE Page Cache Implementation
2//!
3//! A simple, efficient page cache using the SIEVE eviction algorithm.
4//! SIEVE outperforms LRU in many workloads while being simpler to implement.
5//!
6//! # SIEVE Algorithm (NSDI '24)
7//!
8//! SIEVE uses a single "visited" bit instead of LRU's complex list management:
9//! 1. On cache hit: set visited = true
10//! 2. On cache miss (insertion): add to FIFO queue
11//! 3. On eviction: sweep from "hand" position
12//!    - If visited=true: clear bit, skip
13//!    - If visited=false: evict this entry
14//!
15//! # References
16//!
17//! - Turso `core/storage/page_cache.rs:24-80` - PageCacheShardEntry with ref_bit
18//! - Turso `core/storage/page_cache.rs:129-150` - advance_clock_hand()
19//! - "SIEVE is Simpler than LRU" (NSDI '24)
20
21use std::collections::{HashMap, VecDeque};
22use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
23use std::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
24
25use super::page::Page;
26
27fn cache_read<'a, T>(lock: &'a RwLock<T>) -> RwLockReadGuard<'a, T> {
28    lock.read().unwrap_or_else(|poisoned| poisoned.into_inner())
29}
30
31fn cache_write<'a, T>(lock: &'a RwLock<T>) -> RwLockWriteGuard<'a, T> {
32    lock.write()
33        .unwrap_or_else(|poisoned| poisoned.into_inner())
34}
35
36fn cache_lock<'a, T>(lock: &'a Mutex<T>) -> MutexGuard<'a, T> {
37    lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
38}
39
40/// Default cache capacity (number of pages)
41/// Turso uses 100,000 pages (~400MB for 4KB pages)
42pub const DEFAULT_CACHE_CAPACITY: usize = 100_000;
43
44/// Minimum cache capacity
45pub const MIN_CACHE_CAPACITY: usize = 1;
46
47/// Lock-protected cache entry.
48///
49/// Per ADR 0033 (`page-cache-lock-free-hit`), the SIEVE `visited` bit
50/// no longer lives here — it moved to [`ShardMeta`], a separate
51/// cache-line-packed atomic array, so a cache hit can flip it without
52/// upgrading the per-shard write lock. `pin_count` and `dirty` stay in
53/// the entry behind the `entries` `RwLock`: they are written only by
54/// structural operations (`pin`/`unpin`/`mark_dirty`/`mark_clean`,
55/// which take the write lock) and are *never* touched by the hit path.
56struct CacheEntry {
57    /// Identity of the index key owning this slot, protected with its image.
58    page_id: u32,
59    /// The cached page
60    page: Page,
61    /// Pin count (page cannot be evicted while pinned)
62    pin_count: usize,
63    /// Whether the page is dirty (modified)
64    dirty: bool,
65}
66
67impl CacheEntry {
68    fn new(page_id: u32, page: Page, dirty: bool) -> Self {
69        Self {
70            page_id,
71            page,
72            pin_count: 0,
73            dirty,
74        }
75    }
76}
77
78/// Cache line size in bytes (x86-64 / aarch64). The SIEVE `visited`
79/// bits and per-slot `tag`s are laid out in their own line-aligned,
80/// densely packed arrays (see [`ShardMeta`]) so the lock-free hit path
81/// — which only ever touches `visited`/`tag` — never shares a cache
82/// line with the lock-protected [`CacheEntry`] payload (page bytes,
83/// `pin_count`, `dirty`). ADR 0033.
84const CACHE_LINE: usize = 64;
85
86/// Sentinel `tag` value for an unoccupied slot. `u32::MAX` is reserved
87/// (no real page id uses it) so the hit path can tell, with a single
88/// relaxed load, whether the slot it resolved still belongs to the
89/// page it is marking.
90const TAG_EMPTY: u32 = u32::MAX;
91
92/// One cache line of packed `visited` bits: one [`AtomicBool`] (1 byte)
93/// per slot, 64 slots per line. `#[repr(align(64))]` forces the boxed
94/// backing buffer onto a line boundary, so the array starts aligned and
95/// no two unrelated lines ever false-share.
96#[repr(align(64))]
97struct VisitedLine([AtomicBool; CACHE_LINE]);
98
99/// Number of `AtomicU32` tags that fit in one cache line.
100const TAGS_PER_LINE: usize = CACHE_LINE / 4;
101
102/// One cache line of packed slot tags: one [`AtomicU32`] (4 bytes) per
103/// slot, 16 slots per line. Line-aligned for the same reason as
104/// [`VisitedLine`].
105#[repr(align(64))]
106struct TagLine([AtomicU32; TAGS_PER_LINE]);
107
108/// Cache-line-packed SIEVE metadata, split out of [`CacheEntry`].
109///
110/// Both arrays are indexed by slot and are *separate, line-aligned*
111/// allocations (ADR 0033 acceptance: "visited/tag metadata is laid out
112/// in separate cache-line-aligned arrays"). Because the elements are
113/// atomics, the hit path mutates `visited` through a shared `&self`
114/// with no lock at all — satisfying the contract's "cache hits set
115/// visited without acquiring the per-shard write lock".
116///
117/// The arrays are sized to the shard `capacity`. In normal operation a
118/// slot index never reaches `capacity` (`insert` evicts before growing
119/// `entries`), so every access lands in range. The one pathological
120/// exception — a full cache whose every slot is pinned, where `insert`
121/// grows `entries` past `capacity` — is handled by the bounds guards
122/// below: an out-of-range slot reports "not visited" / `TAG_EMPTY`,
123/// which makes that emergency over-capacity entry the first eviction
124/// candidate once a slot frees. No panic, no unbounded metadata.
125struct ShardMeta {
126    visited: Box<[VisitedLine]>,
127    tag: Box<[TagLine]>,
128    capacity: usize,
129}
130
131impl ShardMeta {
132    fn new(capacity: usize) -> Self {
133        let visited = (0..capacity.div_ceil(CACHE_LINE))
134            .map(|_| VisitedLine(std::array::from_fn(|_| AtomicBool::new(false))))
135            .collect::<Vec<_>>()
136            .into_boxed_slice();
137        let tag = (0..capacity.div_ceil(TAGS_PER_LINE))
138            .map(|_| TagLine(std::array::from_fn(|_| AtomicU32::new(TAG_EMPTY))))
139            .collect::<Vec<_>>()
140            .into_boxed_slice();
141        Self {
142            visited,
143            tag,
144            capacity,
145        }
146    }
147
148    /// Lock-free SIEVE hit mark. Benign-race by contract (ADR 0033): a
149    /// concurrent eviction clearing this same bit only costs one extra
150    /// SIEVE cycle. `Relaxed` is sufficient — the bit carries no
151    /// happens-before obligation toward any other state.
152    #[inline]
153    fn mark_visited(&self, slot: usize) {
154        if let Some(b) = self.visited_bit(slot) {
155            b.store(true, Ordering::Relaxed);
156        }
157    }
158
159    #[inline]
160    fn is_visited(&self, slot: usize) -> bool {
161        self.visited_bit(slot)
162            .map(|b| b.load(Ordering::Relaxed))
163            .unwrap_or(false)
164    }
165
166    /// Clear the visited bit. Structural (called from the eviction hand
167    /// under the per-shard locks), but the store itself is a relaxed
168    /// atomic so it composes with the lock-free hit mark.
169    #[inline]
170    fn clear_visited(&self, slot: usize) {
171        if let Some(b) = self.visited_bit(slot) {
172            b.store(false, Ordering::Relaxed);
173        }
174    }
175
176    /// The page id currently tagged at `slot`, or [`TAG_EMPTY`].
177    #[inline]
178    fn tag(&self, slot: usize) -> u32 {
179        self.tag_slot(slot)
180            .map(|t| t.load(Ordering::Relaxed))
181            .unwrap_or(TAG_EMPTY)
182    }
183
184    /// Bind a slot to a page and reset its visited bit. Structural:
185    /// the caller holds the per-shard write lock that serialises slot
186    /// ownership.
187    #[inline]
188    fn occupy(&self, slot: usize, page_id: u32) {
189        if let Some(t) = self.tag_slot(slot) {
190            t.store(page_id, Ordering::Relaxed);
191        }
192        self.clear_visited(slot);
193    }
194
195    /// Release a slot. Structural; caller holds the write lock.
196    #[inline]
197    fn vacate(&self, slot: usize) {
198        if let Some(t) = self.tag_slot(slot) {
199            t.store(TAG_EMPTY, Ordering::Relaxed);
200        }
201        self.clear_visited(slot);
202    }
203
204    /// Reset all metadata to the empty state (used by `clear`).
205    fn reset(&self) {
206        for line in self.visited.iter() {
207            for b in line.0.iter() {
208                b.store(false, Ordering::Relaxed);
209            }
210        }
211        for line in self.tag.iter() {
212            for t in line.0.iter() {
213                t.store(TAG_EMPTY, Ordering::Relaxed);
214            }
215        }
216    }
217
218    #[inline]
219    fn visited_bit(&self, slot: usize) -> Option<&AtomicBool> {
220        if slot >= self.capacity {
221            return None;
222        }
223        self.visited
224            .get(slot / CACHE_LINE)
225            .map(|line| &line.0[slot % CACHE_LINE])
226    }
227
228    #[inline]
229    fn tag_slot(&self, slot: usize) -> Option<&AtomicU32> {
230        if slot >= self.capacity {
231            return None;
232        }
233        self.tag
234            .get(slot / TAGS_PER_LINE)
235            .map(|line| &line.0[slot % TAGS_PER_LINE])
236    }
237}
238
239/// Cache statistics
240#[derive(Debug, Default, Clone)]
241pub struct CacheStats {
242    /// Number of cache hits
243    pub hits: u64,
244    /// Number of cache misses
245    pub misses: u64,
246    /// Number of evictions
247    pub evictions: u64,
248    /// Number of dirty page writebacks
249    pub writebacks: u64,
250}
251
252impl CacheStats {
253    /// Calculate hit rate (0.0 to 1.0)
254    pub fn hit_rate(&self) -> f64 {
255        let total = self.hits + self.misses;
256        if total == 0 {
257            0.0
258        } else {
259            self.hits as f64 / total as f64
260        }
261    }
262}
263
264/// SIEVE-based page cache
265///
266/// Thread-safe page cache using the SIEVE eviction algorithm.
267pub struct PageCacheShard {
268    /// Maximum number of pages to cache
269    capacity: usize,
270    /// Publish index, slots and FIFO changes together. Hits do not take this lock.
271    structure: Mutex<()>,
272    /// Page ID -> Entry index mapping
273    index: RwLock<HashMap<u32, usize>>,
274    /// FIFO queue of page IDs for eviction order
275    fifo: Mutex<VecDeque<u32>>,
276    /// Cache entries (indexed by slot)
277    entries: RwLock<Vec<Option<CacheEntry>>>,
278    /// Cache-line-packed SIEVE metadata (`visited` bits + slot `tag`s),
279    /// kept out of `entries` so the hit path can mark `visited` with a
280    /// lock-free relaxed atomic store. ADR 0033.
281    meta: ShardMeta,
282    /// Free slots
283    free_slots: Mutex<Vec<usize>>,
284    /// SIEVE eviction hand position
285    hand: Mutex<usize>,
286    /// Cache statistics
287    stats: Mutex<CacheStats>,
288}
289
290impl PageCacheShard {
291    /// Create a new page cache with specified capacity
292    pub fn new(capacity: usize) -> Self {
293        let capacity = capacity.max(MIN_CACHE_CAPACITY);
294
295        Self {
296            capacity,
297            structure: Mutex::new(()),
298            index: RwLock::new(HashMap::with_capacity(capacity)),
299            fifo: Mutex::new(VecDeque::with_capacity(capacity)),
300            entries: RwLock::new(Vec::with_capacity(capacity)),
301            meta: ShardMeta::new(capacity),
302            free_slots: Mutex::new(Vec::new()),
303            hand: Mutex::new(0),
304            stats: Mutex::new(CacheStats::default()),
305        }
306    }
307
308    /// Create a page cache with default capacity
309    pub fn with_default_capacity() -> Self {
310        Self::new(DEFAULT_CACHE_CAPACITY)
311    }
312
313    /// Get the current number of cached pages
314    pub fn len(&self) -> usize {
315        cache_read(&self.index).len()
316    }
317
318    /// Check if cache is empty
319    pub fn is_empty(&self) -> bool {
320        self.len() == 0
321    }
322
323    /// Get cache capacity
324    pub fn capacity(&self) -> usize {
325        self.capacity
326    }
327
328    /// Get cache statistics
329    pub fn stats(&self) -> CacheStats {
330        cache_lock(&self.stats).clone()
331    }
332
333    /// Reset statistics
334    pub fn reset_stats(&self) {
335        *cache_lock(&self.stats) = CacheStats::default();
336    }
337
338    /// Get a page from cache
339    ///
340    /// Returns None if page is not in cache (cache miss).
341    /// On hit, marks the page as visited (SIEVE).
342    pub fn get(&self, page_id: u32) -> Option<Page> {
343        // Check if page is in cache
344        let index = cache_read(&self.index);
345        let slot = match index.get(&page_id) {
346            Some(&s) => s,
347            None => {
348                drop(index);
349                // Cache miss
350                cache_lock(&self.stats).misses += 1;
351                return None;
352            }
353        };
354        drop(index);
355
356        #[cfg(test)]
357        tests::after_slot_lookup();
358
359        // Lock-free hit path (ADR 0033). Clone the page under a *read*
360        // lock — never a write lock — then set the SIEVE `visited` bit
361        // with a single relaxed atomic store on the separate
362        // cache-line-packed metadata array. The hit path acquires no
363        // write lock and never touches `pin_count`/`dirty` (those live
364        // in `CacheEntry` behind the write lock and are mutated only by
365        // structural operations). Dropping the per-hit write-lock
366        // upgrade is the whole point: hot-loop reads of an
367        // already-cached page (e.g. BTree inserts hammering the
368        // rightmost leaf) no longer serialise on the entries writer.
369        let entries = cache_read(&self.entries);
370        if let Some(entry) = entries
371            .get(slot)
372            .and_then(|e| e.as_ref())
373            .filter(|entry| entry.page_id == page_id)
374        {
375            let page = entry.page.clone();
376            drop(entries);
377
378            // Mark visited lock-free. Confirm the slot still tags this
379            // page id first: if a racing eviction repurposed the slot
380            // we simply skip the mark (benign — at worst one extra
381            // SIEVE cycle, never a wrong-page write). The tag check
382            // never changes hit/miss accounting; the requested page was
383            // validated under the entries lock, so this is a hit.
384            if self.meta.tag(slot) == page_id {
385                self.meta.mark_visited(slot);
386            }
387
388            cache_lock(&self.stats).hits += 1;
389
390            Some(page)
391        } else {
392            cache_lock(&self.stats).misses += 1;
393            None
394        }
395    }
396
397    /// Insert a page into cache
398    ///
399    /// May trigger eviction if cache is full.
400    /// Returns the evicted page (if dirty) that needs to be written back.
401    pub fn insert(&self, page_id: u32, page: Page) -> Option<Page> {
402        self.insert_inner(page_id, page, false, true)
403    }
404
405    fn insert_inner(
406        &self,
407        page_id: u32,
408        page: Page,
409        dirty: bool,
410        replace_existing: bool,
411    ) -> Option<Page> {
412        let _structure = cache_lock(&self.structure);
413        // Check if already in cache
414        {
415            let index = cache_read(&self.index);
416            if let Some(&slot) = index.get(&page_id) {
417                drop(index);
418
419                // Update existing entry under the write lock; the
420                // SIEVE visited bit lives in the separate metadata
421                // array and is set lock-free (matches the prior
422                // behaviour of marking a re-inserted page visited).
423                let mut entries = cache_write(&self.entries);
424                if let Some(Some(entry)) = entries.get_mut(slot) {
425                    if replace_existing {
426                        entry.page = page;
427                        entry.dirty |= dirty;
428                    }
429                }
430                drop(entries);
431                self.meta.mark_visited(slot);
432                return None;
433            }
434        }
435
436        // Need to insert new entry
437        let mut evicted = None;
438
439        // Check if we need to evict before getting free_slots lock
440        let current_len = self.len();
441        if current_len >= self.capacity {
442            // Need to evict first (no locks held)
443            evicted = self.evict();
444        }
445
446        // Find or create a slot
447        let slot = {
448            let mut free_slots = cache_lock(&self.free_slots);
449            if let Some(slot) = free_slots.pop() {
450                slot
451            } else {
452                drop(free_slots);
453                // Add a new slot
454                let mut entries = cache_write(&self.entries);
455                let slot = entries.len();
456                entries.push(None);
457                slot
458            }
459        };
460
461        // Insert entry
462        {
463            let mut entries = cache_write(&self.entries);
464
465            // Ensure slot exists
466            while entries.len() <= slot {
467                entries.push(None);
468            }
469
470            entries[slot] = Some(CacheEntry::new(page_id, page, dirty));
471        }
472
473        // Update index
474        {
475            let mut index = cache_write(&self.index);
476            index.insert(page_id, slot);
477        }
478
479        // Bind the metadata slot to this page (tag = page_id, visited
480        // reset). Structural: ordered after the entry/index writes that
481        // the per-shard write lock serialises.
482        self.meta.occupy(slot, page_id);
483
484        // Add to FIFO
485        {
486            let mut fifo = cache_lock(&self.fifo);
487            fifo.push_back(page_id);
488        }
489
490        evicted
491    }
492
493    /// Publish a changed image and its dirty state for pager writes.
494    pub fn insert_dirty(&self, page_id: u32, page: Page) -> Option<Page> {
495        let evicted = self.insert_inner(page_id, page, true, true);
496        #[cfg(test)]
497        tests::after_insert_publish();
498        evicted
499    }
500
501    /// Complete a disk read without replacing an image installed meanwhile.
502    pub fn insert_if_absent(&self, page_id: u32, page: Page) -> Option<Page> {
503        self.insert_inner(page_id, page, false, false)
504    }
505
506    /// Evict a page using SIEVE algorithm
507    ///
508    /// Returns the evicted page if it was dirty.
509    fn evict(&self) -> Option<Page> {
510        let mut fifo = cache_lock(&self.fifo);
511        let mut hand = cache_lock(&self.hand);
512
513        if fifo.is_empty() {
514            return None;
515        }
516
517        let fifo_len = fifo.len();
518        let mut attempts = 0;
519
520        // Sweep from hand position
521        loop {
522            if attempts >= fifo_len * 2 {
523                // Couldn't find anything to evict (all pinned?)
524                return None;
525            }
526
527            // Wrap around
528            if *hand >= fifo_len {
529                *hand = 0;
530            }
531
532            let page_id = fifo[*hand];
533            attempts += 1;
534
535            // Get entry slot
536            let slot = {
537                let index = cache_read(&self.index);
538                match index.get(&page_id) {
539                    Some(&s) => s,
540                    None => {
541                        // Page was removed, skip
542                        *hand += 1;
543                        continue;
544                    }
545                }
546            };
547
548            // Check entry. `pin_count`/`dirty` are read under the
549            // entries lock; the SIEVE `visited` bit is read from the
550            // separate atomic metadata array.
551            let should_evict = {
552                let entries = cache_read(&self.entries);
553                match entries.get(slot).and_then(|e| e.as_ref()) {
554                    Some(entry) => {
555                        if entry.pin_count > 0 {
556                            // Pinned, can't evict
557                            false
558                        } else if self.meta.is_visited(slot) {
559                            // Clear visited bit, skip (SIEVE)
560                            false
561                        } else {
562                            // Can evict
563                            true
564                        }
565                    }
566                    None => {
567                        *hand += 1;
568                        continue;
569                    }
570                }
571            };
572
573            if !should_evict {
574                // Clear visited bit lock-free (no entries write lock
575                // needed — `visited` is no longer part of `CacheEntry`).
576                self.meta.clear_visited(slot);
577                *hand += 1;
578                continue;
579            }
580
581            // Evict this entry. Contract guard (ADR 0033): the eviction
582            // hand must never select a pinned page. We re-check under
583            // the write lock — the only place `pin_count` is read for
584            // the eviction decision and the place it must hold.
585            #[cfg(test)]
586            tests::before_evict_take();
587            let (evicted_page, dirty) = {
588                let mut entries = cache_write(&self.entries);
589                let Some(entry) = entries.get(slot).and_then(|e| e.as_ref()) else {
590                    *hand += 1;
591                    continue;
592                };
593                // A pin or write may have arrived after candidate selection.
594                // Revalidate while taking the exact image, in release builds too.
595                if entry.pin_count > 0 {
596                    *hand += 1;
597                    continue;
598                }
599                assert_eq!(
600                    entry.page_id, page_id,
601                    "invariant: eviction owns the indexed slot"
602                );
603                let entry = entries[slot].take().expect("validated cache candidate");
604                (entry.page, entry.dirty)
605            };
606
607            // Release the metadata slot (tag -> empty, visited -> false).
608            self.meta.vacate(slot);
609
610            // Remove from index
611            {
612                let mut index = cache_write(&self.index);
613                index.remove(&page_id);
614            }
615
616            // Remove from FIFO
617            fifo.remove(*hand);
618
619            // Add slot to free list
620            {
621                let mut free_slots = cache_lock(&self.free_slots);
622                free_slots.push(slot);
623            }
624
625            // Update stats
626            {
627                let mut stats = cache_lock(&self.stats);
628                stats.evictions += 1;
629                if dirty {
630                    stats.writebacks += 1;
631                }
632            }
633
634            // Return evicted page if dirty
635            if dirty {
636                return Some(evicted_page);
637            } else {
638                return None;
639            }
640        }
641    }
642
643    /// Mark a page as dirty
644    pub fn mark_dirty(&self, page_id: u32) {
645        let index = cache_read(&self.index);
646        if let Some(&slot) = index.get(&page_id) {
647            drop(index);
648
649            #[cfg(test)]
650            tests::after_slot_lookup();
651
652            let mut entries = cache_write(&self.entries);
653            if let Some(Some(entry)) = entries.get_mut(slot) {
654                if entry.page_id == page_id {
655                    entry.dirty = true;
656                }
657            }
658        }
659    }
660
661    /// Mark a page as clean
662    pub fn mark_clean(&self, page_id: u32) {
663        let index = cache_read(&self.index);
664        if let Some(&slot) = index.get(&page_id) {
665            drop(index);
666
667            #[cfg(test)]
668            tests::after_slot_lookup();
669
670            let mut entries = cache_write(&self.entries);
671            if let Some(Some(entry)) = entries.get_mut(slot) {
672                if entry.page_id == page_id {
673                    entry.dirty = false;
674                }
675            }
676        }
677    }
678
679    /// Pin a page (prevent eviction)
680    pub fn pin(&self, page_id: u32) -> bool {
681        let index = cache_read(&self.index);
682        if let Some(&slot) = index.get(&page_id) {
683            drop(index);
684
685            #[cfg(test)]
686            tests::after_slot_lookup();
687
688            let mut entries = cache_write(&self.entries);
689            if let Some(Some(entry)) = entries.get_mut(slot) {
690                if entry.page_id == page_id {
691                    entry.pin_count += 1;
692                    return true;
693                }
694            }
695        }
696        false
697    }
698
699    /// Unpin a page
700    pub fn unpin(&self, page_id: u32) -> bool {
701        let index = cache_read(&self.index);
702        if let Some(&slot) = index.get(&page_id) {
703            drop(index);
704
705            #[cfg(test)]
706            tests::after_slot_lookup();
707
708            let mut entries = cache_write(&self.entries);
709            if let Some(Some(entry)) = entries.get_mut(slot) {
710                if entry.page_id == page_id && entry.pin_count > 0 {
711                    entry.pin_count -= 1;
712                    return true;
713                }
714            }
715        }
716        false
717    }
718
719    /// Remove a page from cache
720    pub fn remove(&self, page_id: u32) -> Option<Page> {
721        let _structure = cache_lock(&self.structure);
722        // Get and remove from index
723        let slot = {
724            let mut index = cache_write(&self.index);
725            index.remove(&page_id)?
726        };
727
728        // Remove entry
729        let entry = {
730            let mut entries = cache_write(&self.entries);
731            entries.get_mut(slot).and_then(|e| e.take())
732        };
733
734        // Release the metadata slot.
735        self.meta.vacate(slot);
736
737        // Remove from FIFO
738        {
739            let mut fifo = cache_lock(&self.fifo);
740            fifo.retain(|&id| id != page_id);
741        }
742
743        // Add slot to free list
744        {
745            let mut free_slots = cache_lock(&self.free_slots);
746            free_slots.push(slot);
747        }
748
749        entry.map(|e| e.page)
750    }
751
752    /// Flush all dirty pages
753    ///
754    /// Returns an iterator of (page_id, page) for dirty pages.
755    pub fn flush_dirty(&self) -> Vec<(u32, Page)> {
756        let mut dirty_pages = Vec::new();
757
758        let index = cache_read(&self.index);
759        let entries = cache_read(&self.entries);
760
761        for (&page_id, &slot) in index.iter() {
762            if let Some(Some(entry)) = entries.get(slot) {
763                if entry.dirty {
764                    dirty_pages.push((page_id, entry.page.clone()));
765                }
766            }
767        }
768
769        drop(entries);
770        drop(index);
771
772        // Mark all as clean
773        for (page_id, _) in &dirty_pages {
774            self.mark_clean(*page_id);
775        }
776
777        let count = dirty_pages.len();
778        cache_lock(&self.stats).writebacks += count as u64;
779
780        dirty_pages
781    }
782
783    /// Snapshot and clear one dirty image under the same entry lock. A later
784    /// writer must leave its replacement dirty, even while this image flushes.
785    fn take_dirty_page(&self, page_id: u32) -> Option<Page> {
786        let index = cache_read(&self.index);
787        let slot = *index.get(&page_id)?;
788        let mut entries = cache_write(&self.entries);
789        let entry = entries.get_mut(slot)?.as_mut()?;
790        if !entry.dirty {
791            return None;
792        }
793        let page = entry.page.clone();
794        entry.dirty = false;
795        drop(entries);
796        drop(index);
797        cache_lock(&self.stats).writebacks += 1;
798        Some(page)
799    }
800
801    /// Bounded counterpart of `flush_dirty` used by the background
802    /// writer. Snapshots up to `max` dirty pages, marks them clean,
803    /// and returns the (page_id, page) pairs for the caller to
804    /// persist via the pager. Clamps to the current dirty set —
805    /// returning fewer than `max` simply means we're caught up.
806    pub fn flush_some_dirty(&self, max: usize) -> Vec<(u32, Page)> {
807        if max == 0 {
808            return Vec::new();
809        }
810        let mut dirty_pages = Vec::with_capacity(max);
811
812        let index = cache_read(&self.index);
813        let entries = cache_read(&self.entries);
814
815        for (&page_id, &slot) in index.iter() {
816            if dirty_pages.len() >= max {
817                break;
818            }
819            if let Some(Some(entry)) = entries.get(slot) {
820                if entry.dirty {
821                    dirty_pages.push((page_id, entry.page.clone()));
822                }
823            }
824        }
825
826        drop(entries);
827        drop(index);
828
829        for (page_id, _) in &dirty_pages {
830            self.mark_clean(*page_id);
831        }
832
833        let count = dirty_pages.len();
834        cache_lock(&self.stats).writebacks += count as u64;
835
836        dirty_pages
837    }
838
839    /// Count dirty pages currently in the cache. Used by the
840    /// background writer to compute an adaptive flush budget.
841    pub fn dirty_count(&self) -> usize {
842        let index = cache_read(&self.index);
843        let entries = cache_read(&self.entries);
844        let mut count = 0;
845        for (_, &slot) in index.iter() {
846            if let Some(Some(entry)) = entries.get(slot) {
847                if entry.dirty {
848                    count += 1;
849                }
850            }
851        }
852        count
853    }
854
855    /// Clear all entries from cache
856    pub fn clear(&self) {
857        let _structure = cache_lock(&self.structure);
858        let mut index = cache_write(&self.index);
859        let mut entries = cache_write(&self.entries);
860        let mut fifo = cache_lock(&self.fifo);
861        let mut free_slots = cache_lock(&self.free_slots);
862
863        index.clear();
864        entries.clear();
865        fifo.clear();
866        free_slots.clear();
867        self.meta.reset();
868        *cache_lock(&self.hand) = 0;
869    }
870
871    /// Check if a page is in cache
872    pub fn contains(&self, page_id: u32) -> bool {
873        cache_read(&self.index).contains_key(&page_id)
874    }
875
876    /// Get all cached page IDs
877    pub fn page_ids(&self) -> Vec<u32> {
878        cache_read(&self.index).keys().copied().collect()
879    }
880
881    /// Snapshot IDs without cloning images or clearing their dirty flags.
882    fn dirty_page_ids(&self, max: usize) -> Vec<u32> {
883        let index = cache_read(&self.index);
884        let entries = cache_read(&self.entries);
885        index
886            .iter()
887            .filter_map(|(&page_id, &slot)| {
888                let entry = entries.get(slot)?.as_ref()?;
889                entry.dirty.then_some(page_id)
890            })
891            .take(max)
892            .collect()
893    }
894}
895
896impl Default for PageCacheShard {
897    fn default() -> Self {
898        Self::with_default_capacity()
899    }
900}
901
902/// Number of independent `PageCacheShard`s inside a `PageCache`. Must
903/// be a power of two so `page_id & (NUM_SHARDS - 1)` is a valid shard
904/// index. 8 was picked to cover typical bench concurrency (16 workers)
905/// without making each shard's own SIEVE queue too small; bumping to
906/// 16 is safe if profiling ever shows shard-level contention.
907const NUM_SHARDS: usize = 8;
908
909/// Sharded page cache.
910///
911/// Routes each `page_id` to one of at most `NUM_SHARDS` independent
912/// [`PageCacheShard`]s by the low bits of the id. Readers and writers
913/// for disjoint pages hit different shards and do not contend on the
914/// shard-internal RwLocks/Mutexes. Each shard runs its own SIEVE
915/// eviction loop over its share of the requested slots; hot/cold
916/// asymmetry across shards is accepted in exchange for the
917/// contention win. Tiny budgets use fewer shards so one requested page
918/// remains one physical page slot.
919pub struct PageCache {
920    shards: Box<[PageCacheShard]>,
921    capacity: usize,
922}
923
924impl PageCache {
925    pub fn new(capacity: usize) -> Self {
926        let shard_count = (capacity / MIN_CACHE_CAPACITY)
927            .max(1)
928            .next_power_of_two()
929            .min(NUM_SHARDS);
930        let per_shard = capacity.div_ceil(shard_count).max(MIN_CACHE_CAPACITY);
931        let total = per_shard * shard_count;
932        let shards: Vec<PageCacheShard> = (0..shard_count)
933            .map(|_| PageCacheShard::new(per_shard))
934            .collect();
935        Self {
936            shards: shards.into_boxed_slice(),
937            capacity: total,
938        }
939    }
940
941    pub fn with_default_capacity() -> Self {
942        Self::new(DEFAULT_CACHE_CAPACITY)
943    }
944
945    #[inline]
946    fn shard_for(&self, page_id: u32) -> &PageCacheShard {
947        &self.shards[(page_id as usize) & (self.shards.len() - 1)]
948    }
949
950    pub fn len(&self) -> usize {
951        self.shards.iter().map(|s| s.len()).sum()
952    }
953
954    pub fn is_empty(&self) -> bool {
955        self.shards.iter().all(|s| s.is_empty())
956    }
957
958    pub fn capacity(&self) -> usize {
959        self.capacity
960    }
961
962    pub fn stats(&self) -> CacheStats {
963        let mut agg = CacheStats::default();
964        for s in self.shards.iter() {
965            let cs = s.stats();
966            agg.hits += cs.hits;
967            agg.misses += cs.misses;
968            agg.evictions += cs.evictions;
969            agg.writebacks += cs.writebacks;
970        }
971        agg
972    }
973
974    pub fn reset_stats(&self) {
975        for s in self.shards.iter() {
976            s.reset_stats();
977        }
978    }
979
980    pub fn get(&self, page_id: u32) -> Option<Page> {
981        self.shard_for(page_id).get(page_id)
982    }
983
984    pub fn insert(&self, page_id: u32, page: Page) -> Option<Page> {
985        self.shard_for(page_id).insert(page_id, page)
986    }
987
988    pub fn insert_dirty(&self, page_id: u32, page: Page) -> Option<Page> {
989        self.shard_for(page_id).insert_dirty(page_id, page)
990    }
991
992    pub fn insert_if_absent(&self, page_id: u32, page: Page) -> Option<Page> {
993        self.shard_for(page_id).insert_if_absent(page_id, page)
994    }
995
996    pub fn mark_dirty(&self, page_id: u32) {
997        self.shard_for(page_id).mark_dirty(page_id)
998    }
999
1000    pub fn mark_clean(&self, page_id: u32) {
1001        self.shard_for(page_id).mark_clean(page_id)
1002    }
1003
1004    pub fn pin(&self, page_id: u32) -> bool {
1005        self.shard_for(page_id).pin(page_id)
1006    }
1007
1008    pub fn unpin(&self, page_id: u32) -> bool {
1009        self.shard_for(page_id).unpin(page_id)
1010    }
1011
1012    pub fn remove(&self, page_id: u32) -> Option<Page> {
1013        self.shard_for(page_id).remove(page_id)
1014    }
1015
1016    pub fn contains(&self, page_id: u32) -> bool {
1017        self.shard_for(page_id).contains(page_id)
1018    }
1019
1020    pub fn flush_dirty(&self) -> Vec<(u32, Page)> {
1021        let mut out = Vec::new();
1022        for s in self.shards.iter() {
1023            out.extend(s.flush_dirty());
1024        }
1025        out
1026    }
1027
1028    /// Snapshot only the requested dirty images, bounding payload copies by
1029    /// the caller's batch size rather than the full cache capacity.
1030    pub fn flush_dirty_pages(&self, page_ids: &[u32]) -> Vec<(u32, Page)> {
1031        let mut pages = Vec::with_capacity(page_ids.len());
1032        for &page_id in page_ids {
1033            if let Some(page) = self.shard_for(page_id).take_dirty_page(page_id) {
1034                pages.push((page_id, page));
1035            }
1036        }
1037        pages
1038    }
1039
1040    pub(crate) fn flush_dirty_page(&self, page_id: u32) -> Option<Page> {
1041        self.shard_for(page_id).take_dirty_page(page_id)
1042    }
1043
1044    pub fn flush_some_dirty(&self, max: usize) -> Vec<(u32, Page)> {
1045        if max == 0 {
1046            return Vec::new();
1047        }
1048        let mut out = Vec::with_capacity(max);
1049        for s in self.shards.iter() {
1050            if out.len() >= max {
1051                break;
1052            }
1053            let budget = max - out.len();
1054            out.extend(s.flush_some_dirty(budget));
1055        }
1056        out
1057    }
1058
1059    pub fn dirty_count(&self) -> usize {
1060        self.shards.iter().map(|s| s.dirty_count()).sum()
1061    }
1062
1063    pub fn clear(&self) {
1064        for s in self.shards.iter() {
1065            s.clear();
1066        }
1067    }
1068
1069    pub fn page_ids(&self) -> Vec<u32> {
1070        let mut out = Vec::with_capacity(self.len());
1071        for s in self.shards.iter() {
1072            out.extend(s.page_ids());
1073        }
1074        out
1075    }
1076
1077    /// Clean entries must not split a small dirty set into many I/O batches.
1078    pub fn dirty_page_ids(&self, max: usize) -> Vec<u32> {
1079        let mut ids = Vec::new();
1080        for shard in self.shards.iter() {
1081            if ids.len() == max {
1082                break;
1083            }
1084            ids.extend(shard.dirty_page_ids(max - ids.len()));
1085        }
1086        ids
1087    }
1088}
1089
1090impl Default for PageCache {
1091    fn default() -> Self {
1092        Self::with_default_capacity()
1093    }
1094}
1095
1096#[cfg(test)]
1097mod tests {
1098    use super::*;
1099    use crate::storage::page::PageType;
1100
1101    thread_local! {
1102        static SLOT_LOOKUP_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
1103            const { std::cell::RefCell::new(None) };
1104        static EVICT_TAKE_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
1105            const { std::cell::RefCell::new(None) };
1106        static INSERT_PUBLISH_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
1107            const { std::cell::RefCell::new(None) };
1108    }
1109
1110    pub(super) fn after_slot_lookup() {
1111        let hook = SLOT_LOOKUP_HOOK.with(|slot| slot.borrow_mut().take());
1112        if let Some(hook) = hook {
1113            hook();
1114        }
1115    }
1116
1117    pub(super) fn before_evict_take() {
1118        let hook = EVICT_TAKE_HOOK.with(|slot| slot.borrow_mut().take());
1119        if let Some(hook) = hook {
1120            hook();
1121        }
1122    }
1123
1124    pub(super) fn after_insert_publish() {
1125        let hook = INSERT_PUBLISH_HOOK.with(|slot| slot.borrow_mut().take());
1126        if let Some(hook) = hook {
1127            hook();
1128        }
1129    }
1130
1131    #[test]
1132    fn cache_publish_read_miss_does_not_replace_a_newer_dirty_image() {
1133        let cache = PageCacheShard::new(2);
1134        let mut delayed_read = make_page(0);
1135        delayed_read.as_bytes_mut()[100] = 0x11;
1136        let mut current = make_page(0);
1137        current.as_bytes_mut()[100] = 0x22;
1138        cache.insert_dirty(0, current);
1139        cache.insert_if_absent(0, delayed_read);
1140        assert_eq!(cache.get(0).expect("current image").as_bytes()[100], 0x22);
1141        assert_eq!(cache.dirty_count(), 1, "read cleared the current writer");
1142    }
1143
1144    #[test]
1145    fn cache_publish_dirty_image_before_eviction_is_possible() {
1146        use std::sync::{mpsc, Arc};
1147        use std::time::Duration;
1148        let cache = Arc::new(PageCacheShard::new(2));
1149        cache.insert(0, make_page(0));
1150        cache.insert(1, make_page(1));
1151        let (published, publication_done) = mpsc::channel();
1152        let (resume, resumed) = mpsc::channel();
1153        let writer_cache = Arc::clone(&cache);
1154        let writer = std::thread::spawn(move || {
1155            INSERT_PUBLISH_HOOK.with(|slot| {
1156                *slot.borrow_mut() = Some(Box::new(move || {
1157                    published.send(()).expect("notify image publication");
1158                    resumed
1159                        .recv_timeout(Duration::from_secs(5))
1160                        .expect("resume writer");
1161                }));
1162            });
1163            let mut page = make_page(2);
1164            page.as_bytes_mut()[100] = 0x42;
1165            writer_cache.insert_dirty(2, page)
1166        });
1167        publication_done
1168            .recv_timeout(Duration::from_secs(5))
1169            .expect("writer published page 2");
1170        let evicted = [
1171            cache.insert_if_absent(3, make_page(3)),
1172            cache.insert_if_absent(4, make_page(4)),
1173        ];
1174        resume.send(()).expect("resume after read pressure");
1175        assert!(writer.join().expect("writer finishes").is_none());
1176        let image = evicted
1177            .into_iter()
1178            .flatten()
1179            .find(|page| page.page_id() == 2)
1180            .expect("concurrent disk reads must preserve the new dirty image for writeback");
1181        assert_eq!(image.as_bytes()[100], 0x42);
1182    }
1183
1184    fn make_page(id: u32) -> Page {
1185        Page::new(PageType::BTreeLeaf, id)
1186    }
1187
1188    #[test]
1189    fn requested_small_cache_does_not_allocate_all_shards() {
1190        for capacity in [1, 2, 4, 8, 10_000] {
1191            assert_eq!(PageCache::new(capacity).capacity(), capacity);
1192        }
1193    }
1194
1195    #[test]
1196    fn cache_hit_cannot_return_a_replacement_page_after_slot_reuse() {
1197        use std::sync::{mpsc, Arc};
1198        use std::time::Duration;
1199
1200        let cache = Arc::new(PageCacheShard::new(2));
1201        cache.insert(0, make_page(0));
1202        cache.insert(1, make_page(1));
1203        let (looked_up, lookup_done) = mpsc::channel();
1204        let (resume, resumed) = mpsc::channel();
1205        let reader_cache = Arc::clone(&cache);
1206        let reader = std::thread::spawn(move || {
1207            SLOT_LOOKUP_HOOK.with(|slot| {
1208                *slot.borrow_mut() = Some(Box::new(move || {
1209                    looked_up.send(()).expect("notify slot lookup");
1210                    resumed
1211                        .recv_timeout(Duration::from_secs(5))
1212                        .expect("resume reader after eviction");
1213                }));
1214            });
1215            reader_cache.get(0)
1216        });
1217        lookup_done
1218            .recv_timeout(Duration::from_secs(5))
1219            .expect("reader resolved original page slot");
1220        cache.insert(2, make_page(2));
1221        assert!(!cache.contains(0), "fixture must evict the original page");
1222        resume.send(()).expect("resume original page reader");
1223        let observed = reader.join().expect("reader must finish");
1224        assert!(
1225            observed.is_none(),
1226            "evicted page 0 became a hit for replacement page {:?}",
1227            observed.map(|page| page.page_id())
1228        );
1229        assert_eq!(cache.get(2).expect("replacement retained").page_id(), 2);
1230        assert_eq!(cache.stats().misses, 1);
1231    }
1232
1233    #[test]
1234    fn cache_slot_reuse_preserves_replacement_pins_and_dirty_state() {
1235        use std::sync::{mpsc, Arc};
1236        use std::time::Duration;
1237        #[derive(Clone, Copy, Debug)]
1238        enum Action {
1239            Pin,
1240            Unpin,
1241            MarkDirty,
1242            MarkClean,
1243        }
1244        for action in [
1245            Action::Pin,
1246            Action::Unpin,
1247            Action::MarkDirty,
1248            Action::MarkClean,
1249        ] {
1250            let cache = Arc::new(PageCacheShard::new(2));
1251            cache.insert(0, make_page(0));
1252            cache.insert(1, make_page(1));
1253            let (looked_up, lookup_done) = mpsc::channel();
1254            let (resume, resumed) = mpsc::channel();
1255            let worker_cache = Arc::clone(&cache);
1256            let worker = std::thread::spawn(move || {
1257                SLOT_LOOKUP_HOOK.with(|slot| {
1258                    *slot.borrow_mut() = Some(Box::new(move || {
1259                        looked_up.send(()).expect("notify lookup");
1260                        resumed
1261                            .recv_timeout(Duration::from_secs(5))
1262                            .expect("resume stale lookup");
1263                    }));
1264                });
1265                match action {
1266                    Action::Pin => worker_cache.pin(0),
1267                    Action::Unpin => worker_cache.unpin(0),
1268                    Action::MarkDirty => {
1269                        worker_cache.mark_dirty(0);
1270                        false
1271                    }
1272                    Action::MarkClean => {
1273                        worker_cache.mark_clean(0);
1274                        false
1275                    }
1276                }
1277            });
1278            lookup_done
1279                .recv_timeout(Duration::from_secs(5))
1280                .expect("lookup original slot");
1281            cache.insert(2, make_page(2));
1282            if matches!(action, Action::Unpin) {
1283                assert!(cache.pin(2));
1284            }
1285            if matches!(action, Action::MarkClean) {
1286                cache.mark_dirty(2);
1287            }
1288            resume.send(()).expect("resume worker");
1289            let changed = worker.join().expect("worker finishes");
1290            assert!(!changed, "{action:?} targeted a replacement page");
1291            let index = cache_read(&cache.index);
1292            let entries = cache_read(&cache.entries);
1293            let entry = entries[index[&2]].as_ref().expect("replacement");
1294            assert_eq!(
1295                entry.pin_count,
1296                usize::from(matches!(action, Action::Unpin)),
1297                "{action:?}"
1298            );
1299            assert_eq!(
1300                entry.dirty,
1301                matches!(action, Action::MarkClean),
1302                "{action:?}"
1303            );
1304        }
1305    }
1306
1307    #[test]
1308    fn cache_eviction_rechecks_pins_and_dirty_writes_before_removing_a_candidate() {
1309        use std::sync::{mpsc, Arc};
1310        use std::time::Duration;
1311        for pin in [false, true] {
1312            let cache = Arc::new(PageCacheShard::new(2));
1313            cache.insert(0, make_page(0));
1314            cache.insert(1, make_page(1));
1315            let (selected, selection_done) = mpsc::channel();
1316            let (resume, resumed) = mpsc::channel();
1317            let writer_cache = Arc::clone(&cache);
1318            let writer = std::thread::spawn(move || {
1319                EVICT_TAKE_HOOK.with(|slot| {
1320                    *slot.borrow_mut() = Some(Box::new(move || {
1321                        selected.send(()).expect("notify candidate");
1322                        resumed
1323                            .recv_timeout(Duration::from_secs(5))
1324                            .expect("resume eviction");
1325                    }));
1326                });
1327                writer_cache.insert(2, make_page(2))
1328            });
1329            selection_done
1330                .recv_timeout(Duration::from_secs(5))
1331                .expect("selected page 0");
1332            if pin {
1333                assert!(cache.pin(0));
1334            } else {
1335                cache.mark_dirty(0);
1336            }
1337            resume.send(()).expect("resume selected eviction");
1338            let writeback = writer
1339                .join()
1340                .expect("eviction must respect a concurrent pin");
1341            if pin {
1342                assert!(cache.contains(0), "pinned candidate was evicted");
1343                assert!(writeback.is_none());
1344            } else {
1345                assert_eq!(
1346                    writeback
1347                        .expect("new dirty image must be written back")
1348                        .page_id(),
1349                    0
1350                );
1351                assert_eq!(cache.stats().writebacks, 1);
1352            }
1353        }
1354    }
1355
1356    #[test]
1357    fn cache_concurrent_same_page_insertions_do_not_retain_unindexed_slots() {
1358        use std::sync::{Arc, Barrier};
1359        let cache = Arc::new(PageCacheShard::new(4));
1360        for cycle in 0..100 {
1361            let start = Arc::new(Barrier::new(4));
1362            let workers: Vec<_> = (0..4)
1363                .map(|_| {
1364                    let cache = Arc::clone(&cache);
1365                    let start = Arc::clone(&start);
1366                    std::thread::spawn(move || {
1367                        start.wait();
1368                        for round in 0..2000 {
1369                            let id = round % 8;
1370                            cache.insert(id, make_page(id));
1371                        }
1372                    })
1373                })
1374                .collect();
1375            for worker in workers {
1376                worker.join().expect("concurrent insertion");
1377            }
1378            let index = cache_read(&cache.index);
1379            let entries = cache_read(&cache.entries);
1380            assert!(index.len() <= cache.capacity());
1381            assert!(
1382                entries.len() <= cache.capacity(),
1383                "retained slot metadata grew past capacity"
1384            );
1385            assert_eq!(
1386                entries.iter().filter(|entry| entry.is_some()).count(),
1387                index.len(),
1388                "unindexed pages retained"
1389            );
1390            assert_eq!(
1391                cache_lock(&cache.fifo).len(),
1392                index.len(),
1393                "duplicate FIFO identities retained"
1394            );
1395            assert_eq!(
1396                cache_lock(&cache.free_slots).len(),
1397                entries.len() - index.len(),
1398                "unused slots lost at cycle {cycle}"
1399            );
1400        }
1401    }
1402
1403    #[test]
1404    fn cache_key_identity_remains_independent_of_the_payload_header() {
1405        // Cache callers, including the existing concurrency workload, may reuse
1406        // a page image under distinct keys. Slot ownership is the index identity.
1407        let cache = PageCacheShard::new(2);
1408        cache.insert(42, make_page(0));
1409        assert_eq!(cache.get(42).expect("requested cache key").page_id(), 0);
1410        assert!(cache.pin(42));
1411        assert!(cache.unpin(42));
1412        cache.mark_dirty(42);
1413        assert_eq!(cache.dirty_count(), 1);
1414        cache.mark_clean(42);
1415        assert_eq!(cache.dirty_count(), 0);
1416    }
1417
1418    #[test]
1419    fn test_cache_basic() {
1420        let cache = PageCacheShard::new(100);
1421
1422        assert!(cache.is_empty());
1423        assert_eq!(cache.capacity(), 100);
1424
1425        // Insert
1426        cache.insert(1, make_page(1));
1427        assert_eq!(cache.len(), 1);
1428        assert!(cache.contains(1));
1429
1430        // Get
1431        let page = cache.get(1);
1432        assert!(page.is_some());
1433
1434        // Miss
1435        let page = cache.get(999);
1436        assert!(page.is_none());
1437    }
1438
1439    #[test]
1440    fn test_cache_eviction() {
1441        let cache = PageCacheShard::new(4);
1442
1443        // Fill cache
1444        for i in 0..4 {
1445            cache.insert(i, make_page(i));
1446        }
1447
1448        assert_eq!(cache.len(), 4);
1449
1450        // Insert one more, should trigger eviction
1451        cache.insert(100, make_page(100));
1452
1453        // One of the original pages should be evicted
1454        assert_eq!(cache.len(), 4);
1455        assert!(cache.contains(100));
1456    }
1457
1458    #[test]
1459    fn test_cache_sieve_visited() {
1460        let cache = PageCacheShard::new(4);
1461
1462        // Fill cache
1463        for i in 0..4 {
1464            cache.insert(i, make_page(i));
1465        }
1466
1467        // Access page 0 (marks as visited)
1468        cache.get(0);
1469
1470        // Insert new page, should evict non-visited first
1471        cache.insert(100, make_page(100));
1472
1473        // Page 0 should still be there (was visited)
1474        assert!(cache.contains(0));
1475        assert!(cache.contains(100));
1476    }
1477
1478    #[test]
1479    fn test_cache_dirty() {
1480        let cache = PageCacheShard::new(100);
1481
1482        cache.insert(1, make_page(1));
1483        cache.mark_dirty(1);
1484
1485        let dirty = cache.flush_dirty();
1486        assert_eq!(dirty.len(), 1);
1487        assert_eq!(dirty[0].0, 1);
1488
1489        // Should be clean now
1490        let dirty = cache.flush_dirty();
1491        assert_eq!(dirty.len(), 0);
1492    }
1493
1494    #[test]
1495    fn test_cache_pin() {
1496        let cache = PageCacheShard::new(2);
1497
1498        cache.insert(1, make_page(1));
1499        cache.insert(2, make_page(2));
1500
1501        // Pin page 1
1502        assert!(cache.pin(1));
1503
1504        // Fill cache to trigger eviction
1505        cache.insert(3, make_page(3));
1506
1507        // Page 1 should still be there (pinned)
1508        assert!(cache.contains(1));
1509
1510        // Unpin
1511        assert!(cache.unpin(1));
1512    }
1513
1514    #[test]
1515    fn test_cache_remove() {
1516        let cache = PageCacheShard::new(100);
1517
1518        cache.insert(1, make_page(1));
1519        assert!(cache.contains(1));
1520
1521        let removed = cache.remove(1);
1522        assert!(removed.is_some());
1523        assert!(!cache.contains(1));
1524    }
1525
1526    #[test]
1527    fn test_cache_stats() {
1528        let cache = PageCacheShard::new(100);
1529
1530        cache.insert(1, make_page(1));
1531
1532        // Hit
1533        cache.get(1);
1534        cache.get(1);
1535
1536        // Miss
1537        cache.get(999);
1538
1539        let stats = cache.stats();
1540        assert_eq!(stats.hits, 2);
1541        assert_eq!(stats.misses, 1);
1542        assert!((stats.hit_rate() - 0.666).abs() < 0.01);
1543    }
1544
1545    #[test]
1546    fn test_cache_clear() {
1547        let cache = PageCacheShard::new(100);
1548
1549        for i in 0..50 {
1550            cache.insert(i, make_page(i));
1551        }
1552
1553        assert_eq!(cache.len(), 50);
1554
1555        cache.clear();
1556        assert!(cache.is_empty());
1557    }
1558
1559    #[test]
1560    fn test_cache_update_existing() {
1561        let cache = PageCacheShard::new(100);
1562
1563        let mut page1 = make_page(1);
1564        page1.as_bytes_mut()[100] = 0xAA;
1565        cache.insert(1, page1);
1566
1567        let mut page1_updated = make_page(1);
1568        page1_updated.as_bytes_mut()[100] = 0xBB;
1569        cache.insert(1, page1_updated);
1570
1571        // Should still only have one entry
1572        assert_eq!(cache.len(), 1);
1573
1574        // Should have updated value
1575        let retrieved = cache.get(1).unwrap();
1576        assert_eq!(retrieved.as_bytes()[100], 0xBB);
1577    }
1578
1579    #[test]
1580    fn test_cache_recovers_after_index_lock_poisoning() {
1581        let cache = std::sync::Arc::new(PageCacheShard::new(8));
1582        let poison_target = std::sync::Arc::clone(&cache);
1583        let _ = std::thread::spawn(move || {
1584            let _guard = poison_target
1585                .index
1586                .write()
1587                .expect("index lock should be acquired");
1588            panic!("poison index lock");
1589        })
1590        .join();
1591
1592        cache.insert(1, make_page(1));
1593        assert!(cache.contains(1));
1594        assert!(cache.get(1).is_some());
1595    }
1596
1597    #[test]
1598    fn test_cache_recovers_after_stats_lock_poisoning() {
1599        let cache = std::sync::Arc::new(PageCacheShard::new(8));
1600        let poison_target = std::sync::Arc::clone(&cache);
1601        let _ = std::thread::spawn(move || {
1602            let _guard = poison_target
1603                .stats
1604                .lock()
1605                .expect("stats lock should be acquired");
1606            panic!("poison stats lock");
1607        })
1608        .join();
1609
1610        assert!(cache.get(999).is_none());
1611        assert_eq!(cache.stats().misses, 1);
1612        cache.reset_stats();
1613        assert_eq!(cache.stats().misses, 0);
1614    }
1615
1616    /// ADR 0033: the SIEVE `visited`/`tag` metadata lives in its own
1617    /// cache-line-packed, line-aligned arrays — separate from the
1618    /// lock-protected `CacheEntry`. Assert the line types are exactly
1619    /// one 64-byte line and 64-byte aligned (so the boxed backing
1620    /// buffers start on a line boundary and never false-share), and
1621    /// that `CacheEntry` no longer carries the visited bit.
1622    #[test]
1623    fn test_metadata_layout_is_cache_line_packed() {
1624        assert_eq!(std::mem::align_of::<VisitedLine>(), CACHE_LINE);
1625        assert_eq!(std::mem::size_of::<VisitedLine>(), CACHE_LINE);
1626        assert_eq!(std::mem::align_of::<TagLine>(), CACHE_LINE);
1627        assert_eq!(std::mem::size_of::<TagLine>(), CACHE_LINE);
1628
1629        // The metadata arrays are distinct allocations from `entries`.
1630        let meta = ShardMeta::new(200);
1631        // Densely packed: 64 visited bits per line, 16 tags per line.
1632        assert_eq!(meta.visited.len(), 200usize.div_ceil(64));
1633        assert_eq!(meta.tag.len(), 200usize.div_ceil(16));
1634
1635        // occupy/vacate round-trip through the tag array.
1636        meta.occupy(5, 42);
1637        assert_eq!(meta.tag(5), 42);
1638        assert!(!meta.is_visited(5));
1639        meta.mark_visited(5);
1640        assert!(meta.is_visited(5));
1641        meta.vacate(5);
1642        assert_eq!(meta.tag(5), TAG_EMPTY);
1643        assert!(!meta.is_visited(5));
1644    }
1645
1646    /// ADR 0033: a cache hit sets the SIEVE `visited` bit on the
1647    /// separate metadata array (no write-lock upgrade), and that bit
1648    /// drives eviction exactly as before — a hit page survives one
1649    /// eviction sweep while an un-hit page is evicted first.
1650    #[test]
1651    fn test_hit_sets_visited_and_preserves_sieve_policy() {
1652        let cache = PageCacheShard::new(4);
1653        for i in 0..4 {
1654            cache.insert(i, make_page(i));
1655        }
1656
1657        // Hit pages 0 and 1 (lock-free visited mark via metadata array).
1658        assert!(cache.get(0).is_some());
1659        assert!(cache.get(1).is_some());
1660
1661        // Insert two more: SIEVE evicts the un-hit 2 and 3 first.
1662        cache.insert(10, make_page(10));
1663        cache.insert(11, make_page(11));
1664
1665        assert!(cache.contains(0), "hit page 0 must survive");
1666        assert!(cache.contains(1), "hit page 1 must survive");
1667        assert!(cache.contains(10));
1668        assert!(cache.contains(11));
1669        assert!(!cache.contains(2));
1670        assert!(!cache.contains(3));
1671    }
1672
1673    /// ADR 0033 acceptance: under concurrent hits + eviction, no pinned
1674    /// page is ever evicted, and dirty pages are never silently lost —
1675    /// every dirty page that leaves the cache is returned for
1676    /// writeback. Stress/loom-style: many threads hammer hits while
1677    /// other threads drive insertions (and thus the eviction hand).
1678    #[test]
1679    fn test_concurrent_hits_never_evict_pinned_or_lose_dirty() {
1680        use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
1681        use std::sync::Arc;
1682
1683        const CAP: u32 = 64;
1684        const PINNED: u32 = 8; // pages 0..8 stay pinned for the whole run
1685        const DIRTY: u32 = 8; // pages 8..16 are dirty
1686        const HITTERS: usize = 6;
1687        const INSERTERS: usize = 4;
1688        const ROUNDS: usize = 4_000;
1689
1690        let cache = Arc::new(PageCacheShard::new(CAP as usize));
1691        for i in 0..CAP {
1692            cache.insert(i, make_page(i));
1693        }
1694        for p in 0..PINNED {
1695            assert!(cache.pin(p), "pin {p}");
1696        }
1697        for p in PINNED..PINNED + DIRTY {
1698            cache.mark_dirty(p);
1699        }
1700
1701        // Count dirty pages handed back by insert() so we can prove no
1702        // dirty data is silently dropped.
1703        let dirty_written_back = Arc::new(AtomicUsize::new(0));
1704        let stop = Arc::new(AtomicBool::new(false));
1705
1706        // Hitters: loop on the lock-free visited path until signalled.
1707        let mut hitters = Vec::with_capacity(HITTERS);
1708        for h in 0..HITTERS {
1709            let cache = Arc::clone(&cache);
1710            let stop = Arc::clone(&stop);
1711            hitters.push(std::thread::spawn(move || {
1712                let mut id = (h as u32) % CAP;
1713                while !stop.load(Ordering::Relaxed) {
1714                    let _ = cache.get(id);
1715                    id = (id + 1) % CAP;
1716                }
1717            }));
1718        }
1719
1720        // Inserters: bounded churn that drives the eviction hand.
1721        let mut inserters = Vec::with_capacity(INSERTERS);
1722        for w in 0..INSERTERS {
1723            let cache = Arc::clone(&cache);
1724            let dirty_written_back = Arc::clone(&dirty_written_back);
1725            inserters.push(std::thread::spawn(move || {
1726                let base = 1_000 + (w as u32) * 1_000_000;
1727                for i in 0..ROUNDS {
1728                    let id = base + i as u32;
1729                    if cache.insert(id, make_page(id)).is_some() {
1730                        // insert only returns Some for a dirty victim.
1731                        dirty_written_back.fetch_add(1, Ordering::Relaxed);
1732                    }
1733                    // Invariant checked mid-storm: pinned pages are
1734                    // never evicted.
1735                    for p in 0..PINNED {
1736                        assert!(
1737                            cache.contains(p),
1738                            "pinned page {p} was evicted under concurrency"
1739                        );
1740                    }
1741                }
1742            }));
1743        }
1744
1745        // Inserters finish their bounded work; then stop the hitters.
1746        for h in inserters {
1747            h.join()
1748                .expect("inserter thread panicked (invariant broke)");
1749        }
1750        stop.store(true, Ordering::Relaxed);
1751        for h in hitters {
1752            h.join().expect("hitter thread panicked (invariant broke)");
1753        }
1754
1755        // Pinned pages survived the entire storm.
1756        for p in 0..PINNED {
1757            assert!(cache.contains(p), "pinned page {p} missing after storm");
1758        }
1759
1760        // Dirty accounting: a dirty page still in the cache, plus any
1761        // dirty victims handed back for writeback, must conserve the
1762        // original dirty set — no dirty page silently vanished.
1763        let dirty_resident = (PINNED..PINNED + DIRTY)
1764            .filter(|&p| cache.contains(p))
1765            .count();
1766        assert!(
1767            dirty_resident + dirty_written_back.load(Ordering::Relaxed) >= DIRTY as usize,
1768            "dirty pages lost: resident={dirty_resident} written_back={}",
1769            dirty_written_back.load(Ordering::Relaxed)
1770        );
1771
1772        // Note: we deliberately do NOT assert `len <= CAP`. The
1773        // pre-existing fine-grained-lock `insert` checks length and
1774        // evicts in separate steps, so concurrent inserters can
1775        // transiently push a shard past capacity. ShardMeta's bounds
1776        // guards make that overflow safe (over-capacity slots report
1777        // "not visited" / TAG_EMPTY and drain first); enforcing a hard
1778        // cap is out of scope for this lock-free-hit change.
1779    }
1780
1781    /// Legacy single-lock baseline used as a regression check for the
1782    /// sharded `PageCache`. Mirrors the pre-shard cache shape: a single
1783    /// `RwLock<HashMap<u32, Page>>` so every mutation serializes on one
1784    /// writer. We only need the surface used by the concurrency test
1785    /// (`insert` / `get`); keeping it minimal avoids accidentally
1786    /// drifting toward the sharded design and silently flattening the
1787    /// regression signal.
1788    mod legacy_baseline {
1789        use super::Page;
1790        use std::collections::HashMap;
1791        use std::sync::RwLock;
1792
1793        pub struct LegacyPageCache {
1794            entries: RwLock<HashMap<u32, Page>>,
1795        }
1796
1797        impl LegacyPageCache {
1798            pub fn new(_capacity: usize) -> Self {
1799                Self {
1800                    entries: RwLock::new(HashMap::new()),
1801                }
1802            }
1803
1804            pub fn insert(&self, page_id: u32, page: Page) {
1805                let mut entries = self.entries.write().unwrap();
1806                entries.insert(page_id, page);
1807            }
1808
1809            pub fn get(&self, page_id: u32) -> Option<Page> {
1810                let entries = self.entries.read().unwrap();
1811                entries.get(&page_id).cloned()
1812            }
1813        }
1814    }
1815
1816    /// Workload shared by both the sharded and legacy runs of the
1817    /// concurrency property test below. Each worker churns through its
1818    /// own disjoint `page_id` range so any contention seen between
1819    /// workers is purely lock-induced, not data-induced.
1820    fn run_workload<F>(workers: usize, ops_per_worker: usize, run: F) -> std::time::Duration
1821    where
1822        F: Fn(u32, &Page) + Send + Sync + 'static + Clone,
1823    {
1824        use std::sync::Arc;
1825        use std::time::Instant;
1826
1827        let run = Arc::new(run);
1828        let start = Instant::now();
1829        let mut handles = Vec::with_capacity(workers);
1830        for w in 0..workers {
1831            let run = Arc::clone(&run);
1832            handles.push(std::thread::spawn(move || {
1833                let base = (w as u32) * 1_000_000;
1834                let page = make_page(0);
1835                for i in 0..ops_per_worker {
1836                    let id = base + (i as u32);
1837                    run(id, &page);
1838                }
1839            }));
1840        }
1841        for h in handles {
1842            h.join().unwrap();
1843        }
1844        start.elapsed()
1845    }
1846
1847    #[test]
1848    fn test_sharded_cache_scales_concurrently() {
1849        // Property: with disjoint page_id ranges, 10 workers should
1850        // beat 1 worker by a meaningful factor on the sharded cache,
1851        // and the sharded cache should beat the legacy single-lock
1852        // baseline at 10 workers (since the baseline serializes every
1853        // mutation through one global RwLock).
1854        //
1855        // We assert "sub-linear scaling" in a deliberately loose form
1856        // (parallel < 7x serial) so CI flakiness on busy runners
1857        // doesn't false-positive. The legacy comparison is the
1858        // stronger regression signal.
1859        use std::sync::Arc;
1860
1861        const WORKERS: usize = 10;
1862        const OPS: usize = 5_000;
1863        const CAPACITY: usize = 200_000;
1864
1865        // Sharded: 1 worker baseline.
1866        let sharded = Arc::new(PageCache::new(CAPACITY));
1867        let s1 = Arc::clone(&sharded);
1868        let sharded_serial = run_workload(1, OPS * WORKERS, move |id, page| {
1869            s1.insert(id, page.clone());
1870            let _ = s1.get(id);
1871        });
1872
1873        // Sharded: WORKERS parallel.
1874        let sharded = Arc::new(PageCache::new(CAPACITY));
1875        let s2 = Arc::clone(&sharded);
1876        let sharded_parallel = run_workload(WORKERS, OPS, move |id, page| {
1877            s2.insert(id, page.clone());
1878            let _ = s2.get(id);
1879        });
1880
1881        // Legacy: WORKERS parallel.
1882        let legacy = Arc::new(legacy_baseline::LegacyPageCache::new(CAPACITY));
1883        let l2 = Arc::clone(&legacy);
1884        let legacy_parallel = run_workload(WORKERS, OPS, move |id, page| {
1885            l2.insert(id, page.clone());
1886            let _ = l2.get(id);
1887        });
1888
1889        eprintln!(
1890            "page_cache concurrency: sharded 1w={:?} sharded {}w={:?} legacy {}w={:?}",
1891            sharded_serial, WORKERS, sharded_parallel, WORKERS, legacy_parallel
1892        );
1893
1894        // Sub-linear scaling: parallel must not be worse than ~7x the
1895        // serial run. Linear (no scaling) would be ~10x.
1896        assert!(
1897            sharded_parallel.as_nanos() < sharded_serial.as_nanos() * 7,
1898            "sharded cache did not scale: 1w={:?} {}w={:?}",
1899            sharded_serial,
1900            WORKERS,
1901            sharded_parallel
1902        );
1903
1904        // Regression check: sharded must beat the legacy single-lock
1905        // baseline at WORKERS workers. We give the legacy a generous
1906        // 1.2x cushion so a noisy CI box doesn't flip the assertion
1907        // when both designs are within noise (would itself indicate a
1908        // sharding regression worth investigating).
1909        assert!(
1910            sharded_parallel.as_nanos() * 12 < legacy_parallel.as_nanos() * 10,
1911            "sharded cache did not beat legacy baseline: sharded {}w={:?} legacy {}w={:?}",
1912            WORKERS,
1913            sharded_parallel,
1914            WORKERS,
1915            legacy_parallel
1916        );
1917    }
1918}