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 = 2;
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 `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 `capacity / NUM_SHARDS` slots; hot/cold
916/// asymmetry across shards is accepted in exchange for the
917/// contention win.
918pub struct PageCache {
919    shards: Box<[PageCacheShard]>,
920    capacity: usize,
921}
922
923impl PageCache {
924    pub fn new(capacity: usize) -> Self {
925        // Keep every shard above MIN_CACHE_CAPACITY so the inner
926        // SIEVE invariants stay valid even when callers pass tiny
927        // totals (tests frequently do).
928        let per_shard = capacity.div_ceil(NUM_SHARDS).max(MIN_CACHE_CAPACITY);
929        let total = per_shard * NUM_SHARDS;
930        let shards: Vec<PageCacheShard> = (0..NUM_SHARDS)
931            .map(|_| PageCacheShard::new(per_shard))
932            .collect();
933        Self {
934            shards: shards.into_boxed_slice(),
935            capacity: total,
936        }
937    }
938
939    pub fn with_default_capacity() -> Self {
940        Self::new(DEFAULT_CACHE_CAPACITY)
941    }
942
943    #[inline]
944    fn shard_for(&self, page_id: u32) -> &PageCacheShard {
945        &self.shards[(page_id as usize) & (NUM_SHARDS - 1)]
946    }
947
948    pub fn len(&self) -> usize {
949        self.shards.iter().map(|s| s.len()).sum()
950    }
951
952    pub fn is_empty(&self) -> bool {
953        self.shards.iter().all(|s| s.is_empty())
954    }
955
956    pub fn capacity(&self) -> usize {
957        self.capacity
958    }
959
960    pub fn stats(&self) -> CacheStats {
961        let mut agg = CacheStats::default();
962        for s in self.shards.iter() {
963            let cs = s.stats();
964            agg.hits += cs.hits;
965            agg.misses += cs.misses;
966            agg.evictions += cs.evictions;
967            agg.writebacks += cs.writebacks;
968        }
969        agg
970    }
971
972    pub fn reset_stats(&self) {
973        for s in self.shards.iter() {
974            s.reset_stats();
975        }
976    }
977
978    pub fn get(&self, page_id: u32) -> Option<Page> {
979        self.shard_for(page_id).get(page_id)
980    }
981
982    pub fn insert(&self, page_id: u32, page: Page) -> Option<Page> {
983        self.shard_for(page_id).insert(page_id, page)
984    }
985
986    pub fn insert_dirty(&self, page_id: u32, page: Page) -> Option<Page> {
987        self.shard_for(page_id).insert_dirty(page_id, page)
988    }
989
990    pub fn insert_if_absent(&self, page_id: u32, page: Page) -> Option<Page> {
991        self.shard_for(page_id).insert_if_absent(page_id, page)
992    }
993
994    pub fn mark_dirty(&self, page_id: u32) {
995        self.shard_for(page_id).mark_dirty(page_id)
996    }
997
998    pub fn mark_clean(&self, page_id: u32) {
999        self.shard_for(page_id).mark_clean(page_id)
1000    }
1001
1002    pub fn pin(&self, page_id: u32) -> bool {
1003        self.shard_for(page_id).pin(page_id)
1004    }
1005
1006    pub fn unpin(&self, page_id: u32) -> bool {
1007        self.shard_for(page_id).unpin(page_id)
1008    }
1009
1010    pub fn remove(&self, page_id: u32) -> Option<Page> {
1011        self.shard_for(page_id).remove(page_id)
1012    }
1013
1014    pub fn contains(&self, page_id: u32) -> bool {
1015        self.shard_for(page_id).contains(page_id)
1016    }
1017
1018    pub fn flush_dirty(&self) -> Vec<(u32, Page)> {
1019        let mut out = Vec::new();
1020        for s in self.shards.iter() {
1021            out.extend(s.flush_dirty());
1022        }
1023        out
1024    }
1025
1026    /// Snapshot only the requested dirty images, bounding payload copies by
1027    /// the caller's batch size rather than the full cache capacity.
1028    pub fn flush_dirty_pages(&self, page_ids: &[u32]) -> Vec<(u32, Page)> {
1029        let mut pages = Vec::with_capacity(page_ids.len());
1030        for &page_id in page_ids {
1031            if let Some(page) = self.shard_for(page_id).take_dirty_page(page_id) {
1032                pages.push((page_id, page));
1033            }
1034        }
1035        pages
1036    }
1037
1038    pub fn flush_some_dirty(&self, max: usize) -> Vec<(u32, Page)> {
1039        if max == 0 {
1040            return Vec::new();
1041        }
1042        let mut out = Vec::with_capacity(max);
1043        for s in self.shards.iter() {
1044            if out.len() >= max {
1045                break;
1046            }
1047            let budget = max - out.len();
1048            out.extend(s.flush_some_dirty(budget));
1049        }
1050        out
1051    }
1052
1053    pub fn dirty_count(&self) -> usize {
1054        self.shards.iter().map(|s| s.dirty_count()).sum()
1055    }
1056
1057    pub fn clear(&self) {
1058        for s in self.shards.iter() {
1059            s.clear();
1060        }
1061    }
1062
1063    pub fn page_ids(&self) -> Vec<u32> {
1064        let mut out = Vec::with_capacity(self.len());
1065        for s in self.shards.iter() {
1066            out.extend(s.page_ids());
1067        }
1068        out
1069    }
1070
1071    /// Clean entries must not split a small dirty set into many I/O batches.
1072    pub fn dirty_page_ids(&self, max: usize) -> Vec<u32> {
1073        let mut ids = Vec::new();
1074        for shard in self.shards.iter() {
1075            if ids.len() == max {
1076                break;
1077            }
1078            ids.extend(shard.dirty_page_ids(max - ids.len()));
1079        }
1080        ids
1081    }
1082}
1083
1084impl Default for PageCache {
1085    fn default() -> Self {
1086        Self::with_default_capacity()
1087    }
1088}
1089
1090#[cfg(test)]
1091mod tests {
1092    use super::*;
1093    use crate::storage::page::PageType;
1094
1095    thread_local! {
1096        static SLOT_LOOKUP_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
1097            const { std::cell::RefCell::new(None) };
1098        static EVICT_TAKE_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
1099            const { std::cell::RefCell::new(None) };
1100        static INSERT_PUBLISH_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
1101            const { std::cell::RefCell::new(None) };
1102    }
1103
1104    pub(super) fn after_slot_lookup() {
1105        let hook = SLOT_LOOKUP_HOOK.with(|slot| slot.borrow_mut().take());
1106        if let Some(hook) = hook {
1107            hook();
1108        }
1109    }
1110
1111    pub(super) fn before_evict_take() {
1112        let hook = EVICT_TAKE_HOOK.with(|slot| slot.borrow_mut().take());
1113        if let Some(hook) = hook {
1114            hook();
1115        }
1116    }
1117
1118    pub(super) fn after_insert_publish() {
1119        let hook = INSERT_PUBLISH_HOOK.with(|slot| slot.borrow_mut().take());
1120        if let Some(hook) = hook {
1121            hook();
1122        }
1123    }
1124
1125    #[test]
1126    fn cache_publish_read_miss_does_not_replace_a_newer_dirty_image() {
1127        let cache = PageCacheShard::new(2);
1128        let mut delayed_read = make_page(0);
1129        delayed_read.as_bytes_mut()[100] = 0x11;
1130        let mut current = make_page(0);
1131        current.as_bytes_mut()[100] = 0x22;
1132        cache.insert_dirty(0, current);
1133        cache.insert_if_absent(0, delayed_read);
1134        assert_eq!(cache.get(0).expect("current image").as_bytes()[100], 0x22);
1135        assert_eq!(cache.dirty_count(), 1, "read cleared the current writer");
1136    }
1137
1138    #[test]
1139    fn cache_publish_dirty_image_before_eviction_is_possible() {
1140        use std::sync::{mpsc, Arc};
1141        use std::time::Duration;
1142        let cache = Arc::new(PageCacheShard::new(2));
1143        cache.insert(0, make_page(0));
1144        cache.insert(1, make_page(1));
1145        let (published, publication_done) = mpsc::channel();
1146        let (resume, resumed) = mpsc::channel();
1147        let writer_cache = Arc::clone(&cache);
1148        let writer = std::thread::spawn(move || {
1149            INSERT_PUBLISH_HOOK.with(|slot| {
1150                *slot.borrow_mut() = Some(Box::new(move || {
1151                    published.send(()).expect("notify image publication");
1152                    resumed
1153                        .recv_timeout(Duration::from_secs(5))
1154                        .expect("resume writer");
1155                }));
1156            });
1157            let mut page = make_page(2);
1158            page.as_bytes_mut()[100] = 0x42;
1159            writer_cache.insert_dirty(2, page)
1160        });
1161        publication_done
1162            .recv_timeout(Duration::from_secs(5))
1163            .expect("writer published page 2");
1164        let evicted = [
1165            cache.insert_if_absent(3, make_page(3)),
1166            cache.insert_if_absent(4, make_page(4)),
1167        ];
1168        resume.send(()).expect("resume after read pressure");
1169        assert!(writer.join().expect("writer finishes").is_none());
1170        let image = evicted
1171            .into_iter()
1172            .flatten()
1173            .find(|page| page.page_id() == 2)
1174            .expect("concurrent disk reads must preserve the new dirty image for writeback");
1175        assert_eq!(image.as_bytes()[100], 0x42);
1176    }
1177
1178    fn make_page(id: u32) -> Page {
1179        Page::new(PageType::BTreeLeaf, id)
1180    }
1181
1182    #[test]
1183    fn cache_hit_cannot_return_a_replacement_page_after_slot_reuse() {
1184        use std::sync::{mpsc, Arc};
1185        use std::time::Duration;
1186
1187        let cache = Arc::new(PageCacheShard::new(2));
1188        cache.insert(0, make_page(0));
1189        cache.insert(1, make_page(1));
1190        let (looked_up, lookup_done) = mpsc::channel();
1191        let (resume, resumed) = mpsc::channel();
1192        let reader_cache = Arc::clone(&cache);
1193        let reader = std::thread::spawn(move || {
1194            SLOT_LOOKUP_HOOK.with(|slot| {
1195                *slot.borrow_mut() = Some(Box::new(move || {
1196                    looked_up.send(()).expect("notify slot lookup");
1197                    resumed
1198                        .recv_timeout(Duration::from_secs(5))
1199                        .expect("resume reader after eviction");
1200                }));
1201            });
1202            reader_cache.get(0)
1203        });
1204        lookup_done
1205            .recv_timeout(Duration::from_secs(5))
1206            .expect("reader resolved original page slot");
1207        cache.insert(2, make_page(2));
1208        assert!(!cache.contains(0), "fixture must evict the original page");
1209        resume.send(()).expect("resume original page reader");
1210        let observed = reader.join().expect("reader must finish");
1211        assert!(
1212            observed.is_none(),
1213            "evicted page 0 became a hit for replacement page {:?}",
1214            observed.map(|page| page.page_id())
1215        );
1216        assert_eq!(cache.get(2).expect("replacement retained").page_id(), 2);
1217        assert_eq!(cache.stats().misses, 1);
1218    }
1219
1220    #[test]
1221    fn cache_slot_reuse_preserves_replacement_pins_and_dirty_state() {
1222        use std::sync::{mpsc, Arc};
1223        use std::time::Duration;
1224        #[derive(Clone, Copy, Debug)]
1225        enum Action {
1226            Pin,
1227            Unpin,
1228            MarkDirty,
1229            MarkClean,
1230        }
1231        for action in [
1232            Action::Pin,
1233            Action::Unpin,
1234            Action::MarkDirty,
1235            Action::MarkClean,
1236        ] {
1237            let cache = Arc::new(PageCacheShard::new(2));
1238            cache.insert(0, make_page(0));
1239            cache.insert(1, make_page(1));
1240            let (looked_up, lookup_done) = mpsc::channel();
1241            let (resume, resumed) = mpsc::channel();
1242            let worker_cache = Arc::clone(&cache);
1243            let worker = std::thread::spawn(move || {
1244                SLOT_LOOKUP_HOOK.with(|slot| {
1245                    *slot.borrow_mut() = Some(Box::new(move || {
1246                        looked_up.send(()).expect("notify lookup");
1247                        resumed
1248                            .recv_timeout(Duration::from_secs(5))
1249                            .expect("resume stale lookup");
1250                    }));
1251                });
1252                match action {
1253                    Action::Pin => worker_cache.pin(0),
1254                    Action::Unpin => worker_cache.unpin(0),
1255                    Action::MarkDirty => {
1256                        worker_cache.mark_dirty(0);
1257                        false
1258                    }
1259                    Action::MarkClean => {
1260                        worker_cache.mark_clean(0);
1261                        false
1262                    }
1263                }
1264            });
1265            lookup_done
1266                .recv_timeout(Duration::from_secs(5))
1267                .expect("lookup original slot");
1268            cache.insert(2, make_page(2));
1269            if matches!(action, Action::Unpin) {
1270                assert!(cache.pin(2));
1271            }
1272            if matches!(action, Action::MarkClean) {
1273                cache.mark_dirty(2);
1274            }
1275            resume.send(()).expect("resume worker");
1276            let changed = worker.join().expect("worker finishes");
1277            assert!(!changed, "{action:?} targeted a replacement page");
1278            let index = cache_read(&cache.index);
1279            let entries = cache_read(&cache.entries);
1280            let entry = entries[index[&2]].as_ref().expect("replacement");
1281            assert_eq!(
1282                entry.pin_count,
1283                usize::from(matches!(action, Action::Unpin)),
1284                "{action:?}"
1285            );
1286            assert_eq!(
1287                entry.dirty,
1288                matches!(action, Action::MarkClean),
1289                "{action:?}"
1290            );
1291        }
1292    }
1293
1294    #[test]
1295    fn cache_eviction_rechecks_pins_and_dirty_writes_before_removing_a_candidate() {
1296        use std::sync::{mpsc, Arc};
1297        use std::time::Duration;
1298        for pin in [false, true] {
1299            let cache = Arc::new(PageCacheShard::new(2));
1300            cache.insert(0, make_page(0));
1301            cache.insert(1, make_page(1));
1302            let (selected, selection_done) = mpsc::channel();
1303            let (resume, resumed) = mpsc::channel();
1304            let writer_cache = Arc::clone(&cache);
1305            let writer = std::thread::spawn(move || {
1306                EVICT_TAKE_HOOK.with(|slot| {
1307                    *slot.borrow_mut() = Some(Box::new(move || {
1308                        selected.send(()).expect("notify candidate");
1309                        resumed
1310                            .recv_timeout(Duration::from_secs(5))
1311                            .expect("resume eviction");
1312                    }));
1313                });
1314                writer_cache.insert(2, make_page(2))
1315            });
1316            selection_done
1317                .recv_timeout(Duration::from_secs(5))
1318                .expect("selected page 0");
1319            if pin {
1320                assert!(cache.pin(0));
1321            } else {
1322                cache.mark_dirty(0);
1323            }
1324            resume.send(()).expect("resume selected eviction");
1325            let writeback = writer
1326                .join()
1327                .expect("eviction must respect a concurrent pin");
1328            if pin {
1329                assert!(cache.contains(0), "pinned candidate was evicted");
1330                assert!(writeback.is_none());
1331            } else {
1332                assert_eq!(
1333                    writeback
1334                        .expect("new dirty image must be written back")
1335                        .page_id(),
1336                    0
1337                );
1338                assert_eq!(cache.stats().writebacks, 1);
1339            }
1340        }
1341    }
1342
1343    #[test]
1344    fn cache_concurrent_same_page_insertions_do_not_retain_unindexed_slots() {
1345        use std::sync::{Arc, Barrier};
1346        let cache = Arc::new(PageCacheShard::new(4));
1347        for cycle in 0..100 {
1348            let start = Arc::new(Barrier::new(4));
1349            let workers: Vec<_> = (0..4)
1350                .map(|_| {
1351                    let cache = Arc::clone(&cache);
1352                    let start = Arc::clone(&start);
1353                    std::thread::spawn(move || {
1354                        start.wait();
1355                        for round in 0..2000 {
1356                            let id = round % 8;
1357                            cache.insert(id, make_page(id));
1358                        }
1359                    })
1360                })
1361                .collect();
1362            for worker in workers {
1363                worker.join().expect("concurrent insertion");
1364            }
1365            let index = cache_read(&cache.index);
1366            let entries = cache_read(&cache.entries);
1367            assert!(index.len() <= cache.capacity());
1368            assert!(
1369                entries.len() <= cache.capacity(),
1370                "retained slot metadata grew past capacity"
1371            );
1372            assert_eq!(
1373                entries.iter().filter(|entry| entry.is_some()).count(),
1374                index.len(),
1375                "unindexed pages retained"
1376            );
1377            assert_eq!(
1378                cache_lock(&cache.fifo).len(),
1379                index.len(),
1380                "duplicate FIFO identities retained"
1381            );
1382            assert_eq!(
1383                cache_lock(&cache.free_slots).len(),
1384                entries.len() - index.len(),
1385                "unused slots lost at cycle {cycle}"
1386            );
1387        }
1388    }
1389
1390    #[test]
1391    fn cache_key_identity_remains_independent_of_the_payload_header() {
1392        // Cache callers, including the existing concurrency workload, may reuse
1393        // a page image under distinct keys. Slot ownership is the index identity.
1394        let cache = PageCacheShard::new(2);
1395        cache.insert(42, make_page(0));
1396        assert_eq!(cache.get(42).expect("requested cache key").page_id(), 0);
1397        assert!(cache.pin(42));
1398        assert!(cache.unpin(42));
1399        cache.mark_dirty(42);
1400        assert_eq!(cache.dirty_count(), 1);
1401        cache.mark_clean(42);
1402        assert_eq!(cache.dirty_count(), 0);
1403    }
1404
1405    #[test]
1406    fn test_cache_basic() {
1407        let cache = PageCacheShard::new(100);
1408
1409        assert!(cache.is_empty());
1410        assert_eq!(cache.capacity(), 100);
1411
1412        // Insert
1413        cache.insert(1, make_page(1));
1414        assert_eq!(cache.len(), 1);
1415        assert!(cache.contains(1));
1416
1417        // Get
1418        let page = cache.get(1);
1419        assert!(page.is_some());
1420
1421        // Miss
1422        let page = cache.get(999);
1423        assert!(page.is_none());
1424    }
1425
1426    #[test]
1427    fn test_cache_eviction() {
1428        let cache = PageCacheShard::new(4);
1429
1430        // Fill cache
1431        for i in 0..4 {
1432            cache.insert(i, make_page(i));
1433        }
1434
1435        assert_eq!(cache.len(), 4);
1436
1437        // Insert one more, should trigger eviction
1438        cache.insert(100, make_page(100));
1439
1440        // One of the original pages should be evicted
1441        assert_eq!(cache.len(), 4);
1442        assert!(cache.contains(100));
1443    }
1444
1445    #[test]
1446    fn test_cache_sieve_visited() {
1447        let cache = PageCacheShard::new(4);
1448
1449        // Fill cache
1450        for i in 0..4 {
1451            cache.insert(i, make_page(i));
1452        }
1453
1454        // Access page 0 (marks as visited)
1455        cache.get(0);
1456
1457        // Insert new page, should evict non-visited first
1458        cache.insert(100, make_page(100));
1459
1460        // Page 0 should still be there (was visited)
1461        assert!(cache.contains(0));
1462        assert!(cache.contains(100));
1463    }
1464
1465    #[test]
1466    fn test_cache_dirty() {
1467        let cache = PageCacheShard::new(100);
1468
1469        cache.insert(1, make_page(1));
1470        cache.mark_dirty(1);
1471
1472        let dirty = cache.flush_dirty();
1473        assert_eq!(dirty.len(), 1);
1474        assert_eq!(dirty[0].0, 1);
1475
1476        // Should be clean now
1477        let dirty = cache.flush_dirty();
1478        assert_eq!(dirty.len(), 0);
1479    }
1480
1481    #[test]
1482    fn test_cache_pin() {
1483        let cache = PageCacheShard::new(2);
1484
1485        cache.insert(1, make_page(1));
1486        cache.insert(2, make_page(2));
1487
1488        // Pin page 1
1489        assert!(cache.pin(1));
1490
1491        // Fill cache to trigger eviction
1492        cache.insert(3, make_page(3));
1493
1494        // Page 1 should still be there (pinned)
1495        assert!(cache.contains(1));
1496
1497        // Unpin
1498        assert!(cache.unpin(1));
1499    }
1500
1501    #[test]
1502    fn test_cache_remove() {
1503        let cache = PageCacheShard::new(100);
1504
1505        cache.insert(1, make_page(1));
1506        assert!(cache.contains(1));
1507
1508        let removed = cache.remove(1);
1509        assert!(removed.is_some());
1510        assert!(!cache.contains(1));
1511    }
1512
1513    #[test]
1514    fn test_cache_stats() {
1515        let cache = PageCacheShard::new(100);
1516
1517        cache.insert(1, make_page(1));
1518
1519        // Hit
1520        cache.get(1);
1521        cache.get(1);
1522
1523        // Miss
1524        cache.get(999);
1525
1526        let stats = cache.stats();
1527        assert_eq!(stats.hits, 2);
1528        assert_eq!(stats.misses, 1);
1529        assert!((stats.hit_rate() - 0.666).abs() < 0.01);
1530    }
1531
1532    #[test]
1533    fn test_cache_clear() {
1534        let cache = PageCacheShard::new(100);
1535
1536        for i in 0..50 {
1537            cache.insert(i, make_page(i));
1538        }
1539
1540        assert_eq!(cache.len(), 50);
1541
1542        cache.clear();
1543        assert!(cache.is_empty());
1544    }
1545
1546    #[test]
1547    fn test_cache_update_existing() {
1548        let cache = PageCacheShard::new(100);
1549
1550        let mut page1 = make_page(1);
1551        page1.as_bytes_mut()[100] = 0xAA;
1552        cache.insert(1, page1);
1553
1554        let mut page1_updated = make_page(1);
1555        page1_updated.as_bytes_mut()[100] = 0xBB;
1556        cache.insert(1, page1_updated);
1557
1558        // Should still only have one entry
1559        assert_eq!(cache.len(), 1);
1560
1561        // Should have updated value
1562        let retrieved = cache.get(1).unwrap();
1563        assert_eq!(retrieved.as_bytes()[100], 0xBB);
1564    }
1565
1566    #[test]
1567    fn test_cache_recovers_after_index_lock_poisoning() {
1568        let cache = std::sync::Arc::new(PageCacheShard::new(8));
1569        let poison_target = std::sync::Arc::clone(&cache);
1570        let _ = std::thread::spawn(move || {
1571            let _guard = poison_target
1572                .index
1573                .write()
1574                .expect("index lock should be acquired");
1575            panic!("poison index lock");
1576        })
1577        .join();
1578
1579        cache.insert(1, make_page(1));
1580        assert!(cache.contains(1));
1581        assert!(cache.get(1).is_some());
1582    }
1583
1584    #[test]
1585    fn test_cache_recovers_after_stats_lock_poisoning() {
1586        let cache = std::sync::Arc::new(PageCacheShard::new(8));
1587        let poison_target = std::sync::Arc::clone(&cache);
1588        let _ = std::thread::spawn(move || {
1589            let _guard = poison_target
1590                .stats
1591                .lock()
1592                .expect("stats lock should be acquired");
1593            panic!("poison stats lock");
1594        })
1595        .join();
1596
1597        assert!(cache.get(999).is_none());
1598        assert_eq!(cache.stats().misses, 1);
1599        cache.reset_stats();
1600        assert_eq!(cache.stats().misses, 0);
1601    }
1602
1603    /// ADR 0033: the SIEVE `visited`/`tag` metadata lives in its own
1604    /// cache-line-packed, line-aligned arrays — separate from the
1605    /// lock-protected `CacheEntry`. Assert the line types are exactly
1606    /// one 64-byte line and 64-byte aligned (so the boxed backing
1607    /// buffers start on a line boundary and never false-share), and
1608    /// that `CacheEntry` no longer carries the visited bit.
1609    #[test]
1610    fn test_metadata_layout_is_cache_line_packed() {
1611        assert_eq!(std::mem::align_of::<VisitedLine>(), CACHE_LINE);
1612        assert_eq!(std::mem::size_of::<VisitedLine>(), CACHE_LINE);
1613        assert_eq!(std::mem::align_of::<TagLine>(), CACHE_LINE);
1614        assert_eq!(std::mem::size_of::<TagLine>(), CACHE_LINE);
1615
1616        // The metadata arrays are distinct allocations from `entries`.
1617        let meta = ShardMeta::new(200);
1618        // Densely packed: 64 visited bits per line, 16 tags per line.
1619        assert_eq!(meta.visited.len(), 200usize.div_ceil(64));
1620        assert_eq!(meta.tag.len(), 200usize.div_ceil(16));
1621
1622        // occupy/vacate round-trip through the tag array.
1623        meta.occupy(5, 42);
1624        assert_eq!(meta.tag(5), 42);
1625        assert!(!meta.is_visited(5));
1626        meta.mark_visited(5);
1627        assert!(meta.is_visited(5));
1628        meta.vacate(5);
1629        assert_eq!(meta.tag(5), TAG_EMPTY);
1630        assert!(!meta.is_visited(5));
1631    }
1632
1633    /// ADR 0033: a cache hit sets the SIEVE `visited` bit on the
1634    /// separate metadata array (no write-lock upgrade), and that bit
1635    /// drives eviction exactly as before — a hit page survives one
1636    /// eviction sweep while an un-hit page is evicted first.
1637    #[test]
1638    fn test_hit_sets_visited_and_preserves_sieve_policy() {
1639        let cache = PageCacheShard::new(4);
1640        for i in 0..4 {
1641            cache.insert(i, make_page(i));
1642        }
1643
1644        // Hit pages 0 and 1 (lock-free visited mark via metadata array).
1645        assert!(cache.get(0).is_some());
1646        assert!(cache.get(1).is_some());
1647
1648        // Insert two more: SIEVE evicts the un-hit 2 and 3 first.
1649        cache.insert(10, make_page(10));
1650        cache.insert(11, make_page(11));
1651
1652        assert!(cache.contains(0), "hit page 0 must survive");
1653        assert!(cache.contains(1), "hit page 1 must survive");
1654        assert!(cache.contains(10));
1655        assert!(cache.contains(11));
1656        assert!(!cache.contains(2));
1657        assert!(!cache.contains(3));
1658    }
1659
1660    /// ADR 0033 acceptance: under concurrent hits + eviction, no pinned
1661    /// page is ever evicted, and dirty pages are never silently lost —
1662    /// every dirty page that leaves the cache is returned for
1663    /// writeback. Stress/loom-style: many threads hammer hits while
1664    /// other threads drive insertions (and thus the eviction hand).
1665    #[test]
1666    fn test_concurrent_hits_never_evict_pinned_or_lose_dirty() {
1667        use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
1668        use std::sync::Arc;
1669
1670        const CAP: u32 = 64;
1671        const PINNED: u32 = 8; // pages 0..8 stay pinned for the whole run
1672        const DIRTY: u32 = 8; // pages 8..16 are dirty
1673        const HITTERS: usize = 6;
1674        const INSERTERS: usize = 4;
1675        const ROUNDS: usize = 4_000;
1676
1677        let cache = Arc::new(PageCacheShard::new(CAP as usize));
1678        for i in 0..CAP {
1679            cache.insert(i, make_page(i));
1680        }
1681        for p in 0..PINNED {
1682            assert!(cache.pin(p), "pin {p}");
1683        }
1684        for p in PINNED..PINNED + DIRTY {
1685            cache.mark_dirty(p);
1686        }
1687
1688        // Count dirty pages handed back by insert() so we can prove no
1689        // dirty data is silently dropped.
1690        let dirty_written_back = Arc::new(AtomicUsize::new(0));
1691        let stop = Arc::new(AtomicBool::new(false));
1692
1693        // Hitters: loop on the lock-free visited path until signalled.
1694        let mut hitters = Vec::with_capacity(HITTERS);
1695        for h in 0..HITTERS {
1696            let cache = Arc::clone(&cache);
1697            let stop = Arc::clone(&stop);
1698            hitters.push(std::thread::spawn(move || {
1699                let mut id = (h as u32) % CAP;
1700                while !stop.load(Ordering::Relaxed) {
1701                    let _ = cache.get(id);
1702                    id = (id + 1) % CAP;
1703                }
1704            }));
1705        }
1706
1707        // Inserters: bounded churn that drives the eviction hand.
1708        let mut inserters = Vec::with_capacity(INSERTERS);
1709        for w in 0..INSERTERS {
1710            let cache = Arc::clone(&cache);
1711            let dirty_written_back = Arc::clone(&dirty_written_back);
1712            inserters.push(std::thread::spawn(move || {
1713                let base = 1_000 + (w as u32) * 1_000_000;
1714                for i in 0..ROUNDS {
1715                    let id = base + i as u32;
1716                    if cache.insert(id, make_page(id)).is_some() {
1717                        // insert only returns Some for a dirty victim.
1718                        dirty_written_back.fetch_add(1, Ordering::Relaxed);
1719                    }
1720                    // Invariant checked mid-storm: pinned pages are
1721                    // never evicted.
1722                    for p in 0..PINNED {
1723                        assert!(
1724                            cache.contains(p),
1725                            "pinned page {p} was evicted under concurrency"
1726                        );
1727                    }
1728                }
1729            }));
1730        }
1731
1732        // Inserters finish their bounded work; then stop the hitters.
1733        for h in inserters {
1734            h.join()
1735                .expect("inserter thread panicked (invariant broke)");
1736        }
1737        stop.store(true, Ordering::Relaxed);
1738        for h in hitters {
1739            h.join().expect("hitter thread panicked (invariant broke)");
1740        }
1741
1742        // Pinned pages survived the entire storm.
1743        for p in 0..PINNED {
1744            assert!(cache.contains(p), "pinned page {p} missing after storm");
1745        }
1746
1747        // Dirty accounting: a dirty page still in the cache, plus any
1748        // dirty victims handed back for writeback, must conserve the
1749        // original dirty set — no dirty page silently vanished.
1750        let dirty_resident = (PINNED..PINNED + DIRTY)
1751            .filter(|&p| cache.contains(p))
1752            .count();
1753        assert!(
1754            dirty_resident + dirty_written_back.load(Ordering::Relaxed) >= DIRTY as usize,
1755            "dirty pages lost: resident={dirty_resident} written_back={}",
1756            dirty_written_back.load(Ordering::Relaxed)
1757        );
1758
1759        // Note: we deliberately do NOT assert `len <= CAP`. The
1760        // pre-existing fine-grained-lock `insert` checks length and
1761        // evicts in separate steps, so concurrent inserters can
1762        // transiently push a shard past capacity. ShardMeta's bounds
1763        // guards make that overflow safe (over-capacity slots report
1764        // "not visited" / TAG_EMPTY and drain first); enforcing a hard
1765        // cap is out of scope for this lock-free-hit change.
1766    }
1767
1768    /// Legacy single-lock baseline used as a regression check for the
1769    /// sharded `PageCache`. Mirrors the pre-shard cache shape: a single
1770    /// `RwLock<HashMap<u32, Page>>` so every mutation serializes on one
1771    /// writer. We only need the surface used by the concurrency test
1772    /// (`insert` / `get`); keeping it minimal avoids accidentally
1773    /// drifting toward the sharded design and silently flattening the
1774    /// regression signal.
1775    mod legacy_baseline {
1776        use super::Page;
1777        use std::collections::HashMap;
1778        use std::sync::RwLock;
1779
1780        pub struct LegacyPageCache {
1781            entries: RwLock<HashMap<u32, Page>>,
1782        }
1783
1784        impl LegacyPageCache {
1785            pub fn new(_capacity: usize) -> Self {
1786                Self {
1787                    entries: RwLock::new(HashMap::new()),
1788                }
1789            }
1790
1791            pub fn insert(&self, page_id: u32, page: Page) {
1792                let mut entries = self.entries.write().unwrap();
1793                entries.insert(page_id, page);
1794            }
1795
1796            pub fn get(&self, page_id: u32) -> Option<Page> {
1797                let entries = self.entries.read().unwrap();
1798                entries.get(&page_id).cloned()
1799            }
1800        }
1801    }
1802
1803    /// Workload shared by both the sharded and legacy runs of the
1804    /// concurrency property test below. Each worker churns through its
1805    /// own disjoint `page_id` range so any contention seen between
1806    /// workers is purely lock-induced, not data-induced.
1807    fn run_workload<F>(workers: usize, ops_per_worker: usize, run: F) -> std::time::Duration
1808    where
1809        F: Fn(u32, &Page) + Send + Sync + 'static + Clone,
1810    {
1811        use std::sync::Arc;
1812        use std::time::Instant;
1813
1814        let run = Arc::new(run);
1815        let start = Instant::now();
1816        let mut handles = Vec::with_capacity(workers);
1817        for w in 0..workers {
1818            let run = Arc::clone(&run);
1819            handles.push(std::thread::spawn(move || {
1820                let base = (w as u32) * 1_000_000;
1821                let page = make_page(0);
1822                for i in 0..ops_per_worker {
1823                    let id = base + (i as u32);
1824                    run(id, &page);
1825                }
1826            }));
1827        }
1828        for h in handles {
1829            h.join().unwrap();
1830        }
1831        start.elapsed()
1832    }
1833
1834    #[test]
1835    fn test_sharded_cache_scales_concurrently() {
1836        // Property: with disjoint page_id ranges, 10 workers should
1837        // beat 1 worker by a meaningful factor on the sharded cache,
1838        // and the sharded cache should beat the legacy single-lock
1839        // baseline at 10 workers (since the baseline serializes every
1840        // mutation through one global RwLock).
1841        //
1842        // We assert "sub-linear scaling" in a deliberately loose form
1843        // (parallel < 7x serial) so CI flakiness on busy runners
1844        // doesn't false-positive. The legacy comparison is the
1845        // stronger regression signal.
1846        use std::sync::Arc;
1847
1848        const WORKERS: usize = 10;
1849        const OPS: usize = 5_000;
1850        const CAPACITY: usize = 200_000;
1851
1852        // Sharded: 1 worker baseline.
1853        let sharded = Arc::new(PageCache::new(CAPACITY));
1854        let s1 = Arc::clone(&sharded);
1855        let sharded_serial = run_workload(1, OPS * WORKERS, move |id, page| {
1856            s1.insert(id, page.clone());
1857            let _ = s1.get(id);
1858        });
1859
1860        // Sharded: WORKERS parallel.
1861        let sharded = Arc::new(PageCache::new(CAPACITY));
1862        let s2 = Arc::clone(&sharded);
1863        let sharded_parallel = run_workload(WORKERS, OPS, move |id, page| {
1864            s2.insert(id, page.clone());
1865            let _ = s2.get(id);
1866        });
1867
1868        // Legacy: WORKERS parallel.
1869        let legacy = Arc::new(legacy_baseline::LegacyPageCache::new(CAPACITY));
1870        let l2 = Arc::clone(&legacy);
1871        let legacy_parallel = run_workload(WORKERS, OPS, move |id, page| {
1872            l2.insert(id, page.clone());
1873            let _ = l2.get(id);
1874        });
1875
1876        eprintln!(
1877            "page_cache concurrency: sharded 1w={:?} sharded {}w={:?} legacy {}w={:?}",
1878            sharded_serial, WORKERS, sharded_parallel, WORKERS, legacy_parallel
1879        );
1880
1881        // Sub-linear scaling: parallel must not be worse than ~7x the
1882        // serial run. Linear (no scaling) would be ~10x.
1883        assert!(
1884            sharded_parallel.as_nanos() < sharded_serial.as_nanos() * 7,
1885            "sharded cache did not scale: 1w={:?} {}w={:?}",
1886            sharded_serial,
1887            WORKERS,
1888            sharded_parallel
1889        );
1890
1891        // Regression check: sharded must beat the legacy single-lock
1892        // baseline at WORKERS workers. We give the legacy a generous
1893        // 1.2x cushion so a noisy CI box doesn't flip the assertion
1894        // when both designs are within noise (would itself indicate a
1895        // sharding regression worth investigating).
1896        assert!(
1897            sharded_parallel.as_nanos() * 12 < legacy_parallel.as_nanos() * 10,
1898            "sharded cache did not beat legacy baseline: sharded {}w={:?} legacy {}w={:?}",
1899            WORKERS,
1900            sharded_parallel,
1901            WORKERS,
1902            legacy_parallel
1903        );
1904    }
1905}