<commit_before>//! `pagecache` is a lock-free pagecache and log for building high-performance
//! databases.
#![allow(unsafe_code)]

pub mod constants;
pub mod logger;

mod disk_pointer;
mod header;
mod heap;
mod iobuf;
mod iterator;
mod pagetable;
#[cfg(any(all(not(unix), not(windows)), miri))]
mod parallel_io_polyfill;
#[cfg(all(unix, not(miri)))]
mod parallel_io_unix;
#[cfg(all(windows, not(miri)))]
mod parallel_io_windows;
mod reservation;
mod segment;
mod snapshot;

use std::{collections::BinaryHeap, ops::Deref};

use crate::*;

#[cfg(any(all(not(unix), not(windows)), miri))]
use parallel_io_polyfill::{pread_exact, pread_exact_or_eof, pwrite_all};

#[cfg(all(unix, not(miri)))]
use parallel_io_unix::{pread_exact, pread_exact_or_eof, pwrite_all};

#[cfg(all(windows, not(miri)))]
use parallel_io_windows::{pread_exact, pread_exact_or_eof, pwrite_all};

use self::{
    constants::{
        BATCH_MANIFEST_PID, COUNTER_PID, META_PID,
        PAGE_CONSOLIDATION_THRESHOLD, SEGMENT_CLEANUP_THRESHOLD,
    },
    header::Header,
    iobuf::{roll_iobuf, IoBuf, IoBufs},
    iterator::{raw_segment_iter_from, LogIter},
    pagetable::PageTable,
    segment::{SegmentAccountant, SegmentCleaner, SegmentOp},
};

pub(crate) use self::{
    heap::{Heap, HeapId},
    logger::{
        read_message, read_segment_header, MessageHeader, SegmentHeader,
        SegmentNumber,
    },
    reservation::Reservation,
    snapshot::{read_snapshot_or_default, PageState, Snapshot},
};

pub use self::{
    constants::{MAX_MSG_HEADER_LEN, MAX_SPACE_AMPLIFICATION, SEG_HEADER_LEN},
    disk_pointer::DiskPtr,
    logger::{Log, LogRead},
};

/// A file offset in the database log.
pub type LogOffset = u64;

/// The logical sequence number of an item in the database log.
pub type Lsn = i64;

/// A page identifier.
pub type PageId = u64;

/// Uses a non-varint `Lsn` to mark offsets.
#[derive(Default, Clone, Copy, Ord, PartialOrd, Eq, PartialEq, Debug)]
#[repr(transparent)]
pub struct BatchManifest(pub Lsn);

/// A buffer with an associated offset. Useful for
/// batching many reads over a file segment.
#[derive(Debug)]
pub struct BasedBuf {
    pub buf: Vec<u8>,
    pub offset: LogOffset,
}

/// A byte used to disambiguate log message types
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
#[repr(u8)]
pub enum MessageKind {
    /// The EVIL_BYTE is written as a canary to help
    /// detect torn writes.
    Corrupted = 0,
    /// Indicates that the following buffer corresponds
    /// to a reservation for an in-memory operation that
    /// failed to complete. It should be skipped during
    /// recovery.
    Canceled = 1,
    /// Indicates that the following buffer is used
    /// as padding to fill out the rest of the segment
    /// before sealing it.
    Cap = 2,
    /// Indicates that the following buffer contains
    /// an Lsn for the last write in an atomic writebatch.
    BatchManifest = 3,
    /// Indicates that this page was freed from the pagetable.
    Free = 4,
    /// Indicates that the last persisted ID was at least
    /// this high.
    Counter = 5,
    /// The meta page, stored inline
    InlineMeta = 6,
    /// The meta page, stored heaply
    HeapMeta = 7,
    /// A consolidated page replacement, stored inline
    InlineNode = 8,
    /// A consolidated page replacement, stored heaply
    HeapNode = 9,
    /// A partial page update, stored inline
    InlineLink = 10,
    /// A partial page update, stored heaply
    HeapLink = 11,
}

impl MessageKind {
    pub(crate) const fn into(self) -> u8 {
        self as u8
    }
}

impl From<u8> for MessageKind {
    fn from(byte: u8) -> Self {
        use MessageKind::*;
        match byte {
            0 => Corrupted,
            1 => Canceled,
            2 => Cap,
            3 => BatchManifest,
            4 => Free,
            5 => Counter,
            6 => InlineMeta,
            7 => HeapMeta,
            8 => InlineNode,
            9 => HeapNode,
            10 => InlineLink,
            11 => HeapLink,
            other => {
                debug!("encountered unexpected message kind byte {}", other);
                Corrupted
            }
        }
    }
}

/// The high-level types of stored information
/// about pages and their mutations
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LogKind {
    /// Persisted data containing a page replacement
    Replace,
    /// Persisted immutable update
    Link,
    /// Freeing of a page
    Free,
    /// Some state indicating this should be skipped
    Skip,
    /// Unexpected corruption
    Corrupted,
}

fn log_kind_from_update(update: &Update) -> LogKind {
    match update {
        Update::Free => LogKind::Free,
        Update::Link(..) => LogKind::Link,
        Update::Node(..) | Update::Counter(..) | Update::Meta(..) => {
            LogKind::Replace
        }
    }
}

impl From<MessageKind> for LogKind {
    fn from(kind: MessageKind) -> Self {
        match kind {
            MessageKind::Free => LogKind::Free,
            MessageKind::InlineNode
            | MessageKind::Counter
            | MessageKind::HeapNode
            | MessageKind::InlineMeta
            | MessageKind::HeapMeta => LogKind::Replace,
            MessageKind::InlineLink | MessageKind::HeapLink => LogKind::Link,
            MessageKind::Canceled
            | MessageKind::Cap
            | MessageKind::BatchManifest => LogKind::Skip,
            other => {
                debug!("encountered unexpected message kind byte {:?}", other);
                LogKind::Corrupted
            }
        }
    }
}

fn assert_usize<T>(from: T) -> usize
where
    usize: TryFrom<T, Error = std::num::TryFromIntError>,
{
    usize::try_from(from).expect("lost data cast while converting to usize")
}

// TODO remove this with atomic fetch_max if we increase MSRV to 1.45 or higher
fn bump_atomic_lsn(atomic_lsn: &AtomicLsn, to: Lsn) {
    let mut current = atomic_lsn.load(Acquire);
    loop {
        if current >= to {
            return;
        }
        let last =
            atomic_lsn.compare_exchange_weak(current, to, SeqCst, SeqCst);
        if last.is_ok() {
            // we succeeded.
            return;
        }
        current = last.unwrap_err();
    }
}

use std::convert::{TryFrom, TryInto};

#[inline]
pub(crate) fn lsn_to_arr(number: Lsn) -> [u8; 8] {
    number.to_le_bytes()
}

#[inline]
pub(crate) fn arr_to_lsn(arr: &[u8]) -> Lsn {
    Lsn::from_le_bytes(arr.try_into().unwrap())
}

#[inline]
pub(crate) fn u64_to_arr(number: u64) -> [u8; 8] {
    number.to_le_bytes()
}

#[inline]
pub(crate) fn arr_to_u32(arr: &[u8]) -> u32 {
    u32::from_le_bytes(arr.try_into().unwrap())
}

#[inline]
pub(crate) fn u32_to_arr(number: u32) -> [u8; 4] {
    number.to_le_bytes()
}

#[allow(clippy::needless_pass_by_value)]
#[allow(clippy::needless_return)]
pub(crate) fn decompress(in_buf: Vec<u8>) -> Vec<u8> {
    #[cfg(feature = "compression")]
    {
        use zstd::stream::decode_all;

        let scootable_in_buf = &mut &*in_buf;
        let raw: IVec = IVec::deserialize(scootable_in_buf)
            .expect("this had to be serialized with an extra length frame");
        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.decompress);
        let out_buf = decode_all(&raw[..]).expect(
            "failed to decompress data. \
             This is not expected, please open an issue on \
             https://github.com/spacejam/sled so we can \
             fix this critical issue ASAP. Thank you :)",
        );

        return out_buf;
    }

    #[cfg(not(feature = "compression"))]
    in_buf
}

#[derive(Debug, Clone, Copy)]
pub struct NodeView<'g>(pub(crate) PageView<'g>);

impl<'g> Deref for NodeView<'g> {
    type Target = Node;
    fn deref(&self) -> &Node {
        self.0.as_node()
    }
}

#[derive(Debug, Clone, Copy)]
pub struct MetaView<'g>(PageView<'g>);

impl<'g> Deref for MetaView<'g> {
    type Target = Meta;
    fn deref(&self) -> &Meta {
        self.0.as_meta()
    }
}

#[derive(Debug, Clone, Copy)]
pub struct PageView<'g> {
    pub(crate) read: Shared<'g, Page>,
    pub(crate) entry: &'g Atomic<Page>,
}

impl<'g> Deref for PageView<'g> {
    type Target = Page;

    fn deref(&self) -> &Page {
        unsafe { self.read.deref() }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CacheInfo {
    pub ts: u64,
    pub lsn: Lsn,
    pub pointer: DiskPtr,
    pub log_size: u64,
}

#[cfg(test)]
impl quickcheck::Arbitrary for CacheInfo {
    fn arbitrary<G: quickcheck::Gen>(g: &mut G) -> CacheInfo {
        use rand::Rng;

        CacheInfo {
            ts: g.gen(),
            lsn: g.gen(),
            pointer: DiskPtr::arbitrary(g),
            log_size: g.gen(),
        }
    }
}

/// Update<PageLinkment> denotes a state or a change in a sequence of updates
/// of which a page consists.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "testing", derive(PartialEq))]
pub(crate) enum Update {
    Link(Link),
    Node(Node),
    Free,
    Counter(u64),
    Meta(Meta),
}

impl Update {
    fn as_node(&self) -> &Node {
        match self {
            Update::Node(node) => node,
            other => panic!("called as_node on non-Node: {:?}", other),
        }
    }

    fn as_node_mut(&mut self) -> &mut Node {
        match self {
            Update::Node(node) => node,
            other => panic!("called as_node_mut on non-Node: {:?}", other),
        }
    }

    fn as_link(&self) -> &Link {
        match self {
            Update::Link(link) => link,
            other => panic!("called as_link on non-Link: {:?}", other),
        }
    }

    fn as_meta(&self) -> &Meta {
        if let Update::Meta(meta) = self {
            meta
        } else {
            panic!("called as_meta on {:?}", self)
        }
    }

    fn as_counter(&self) -> u64 {
        if let Update::Counter(counter) = self {
            *counter
        } else {
            panic!("called as_counter on {:?}", self)
        }
    }
}

/// Ensures that any operations that are written to disk between the
/// creation of this guard and its destruction will be recovered
/// atomically. When this guard is dropped, it marks in an earlier
/// reservation where the stable tip must be in order to perform
/// recovery. If this is beyond where the system successfully
/// wrote before crashing, then the recovery will stop immediately
/// before any of the atomic batch can be partially recovered.
///
/// Must call `seal_batch` to complete the atomic batch operation.
///
/// If this is dropped without calling `seal_batch`, the complete
/// recovery effect will not occur.
#[derive(Debug)]
pub struct RecoveryGuard<'a> {
    batch_res: Reservation<'a>,
}

impl<'a> RecoveryGuard<'a> {
    /// Writes the last LSN for a batch into an earlier
    /// reservation, releasing it.
    pub(crate) fn seal_batch(self) -> Result<()> {
        let max_reserved =
            self.batch_res.log.iobufs.max_reserved_lsn.load(Acquire);
        self.batch_res.mark_writebatch(max_reserved).map(|_| ())
    }
}

/// A page consists of a sequence of state transformations
/// with associated storage parameters like disk pos, lsn, time.
#[derive(Debug, Clone)]
pub struct Page {
    update: Option<Update>,
    cache_infos: Vec<CacheInfo>,
}

impl Page {
    fn to_page_state(&self) -> PageState {
        let base = &self.cache_infos[0];
        if self.is_free() {
            PageState::Free(base.lsn, base.pointer)
        } else {
            let mut frags: Vec<(Lsn, DiskPtr, u64)> = vec![];

            for cache_info in self.cache_infos.iter().skip(1) {
                frags.push((
                    cache_info.lsn,
                    cache_info.pointer,
                    cache_info.log_size,
                ));
            }

            PageState::Present {
                base: (base.lsn, base.pointer, base.log_size),
                frags,
            }
        }
    }

    fn as_node(&self) -> &Node {
        self.update.as_ref().unwrap().as_node()
    }

    fn as_meta(&self) -> &Meta {
        self.update.as_ref().unwrap().as_meta()
    }

    fn as_counter(&self) -> u64 {
        self.update.as_ref().unwrap().as_counter()
    }

    fn is_free(&self) -> bool {
        if let Some(Update::Free) = self.update {
            true
        } else {
            false
        }
    }

    fn last_lsn(&self) -> Lsn {
        self.cache_infos.last().map(|ci| ci.lsn).unwrap()
    }

    pub(crate) fn log_size(&self) -> u64 {
        self.cache_infos.iter().map(|ci| ci.log_size).sum()
    }

    fn ts(&self) -> u64 {
        self.cache_infos.last().map_or(0, |ci| ci.ts)
    }

    fn lone_heap_item(&self) -> Option<CacheInfo> {
        if self.cache_infos.len() == 1
            && self.cache_infos[0].pointer.is_heap_item()
        {
            Some(self.cache_infos[0])
        } else {
            None
        }
    }
}

/// A lock-free pagecache which supports linkmented pages
/// for dramatically improving write throughput.
#[derive(Clone)]
pub struct PageCache(Arc<PageCacheInner>);

impl Deref for PageCache {
    type Target = PageCacheInner;

    fn deref(&self) -> &PageCacheInner {
        &self.0
    }
}

pub struct PageCacheInner {
    was_recovered: bool,
    pub(crate) config: RunningConfig,
    inner: PageTable,
    next_pid_to_allocate: Mutex<PageId>,
    // needs to be a sub-Arc because we separate
    // it for async modification in an EBR guard
    free: Arc<Mutex<BinaryHeap<PageId>>>,
    #[doc(hidden)]
    pub log: Log,
    lru: Lru,

    idgen: AtomicU64,
    idgen_persists: AtomicU64,
    idgen_persist_mu: Mutex<()>,

    // fuzzy snapshot-related items
    snapshot_min_lsn: AtomicLsn,
    links: AtomicU64,
    snapshot_lock: Mutex<()>,
}

impl Debug for PageCache {
    fn fmt(
        &self,
        f: &mut fmt::Formatter<'_>,
    ) -> std::result::Result<(), fmt::Error> {
        f.write_str(&*format!(
            "PageCache {{ max: {:?} free: {:?} }}\n",
            *self.next_pid_to_allocate.lock(),
            self.free
        ))
    }
}

#[cfg(feature = "event_log")]
impl Drop for PageCacheInner {
    fn drop(&mut self) {
        trace!("dropping pagecache");

        // we can't as easily assert recovery
        // invariants across failpoints for now
        if self.log.iobufs.config.global_error().is_ok() {
            let mut pages_before_restart = Map::default();

            let guard = pin();

            self.config
                .event_log
                .meta_before_restart(self.get_meta(&guard).deref().clone());

            for pid in 0..*self.next_pid_to_allocate.lock() {
                let pte = self.inner.get(pid, &guard);
                let pointers =
                    pte.cache_infos.iter().map(|ci| ci.pointer).collect();
                pages_before_restart.insert(pid, pointers);
            }

            self.config.event_log.pages_before_restart(pages_before_restart);
        }

        trace!("pagecache dropped");
    }
}

impl PageCache {
    /// Instantiate a new `PageCache`.
    pub(crate) fn start(config: RunningConfig) -> Result<PageCache> {
        trace!("starting pagecache");

        config.reset_global_error();

        // try to pull any existing snapshot off disk, and
        // apply any new data to it to "catch-up" the
        // snapshot before loading it.
        let snapshot = read_snapshot_or_default(&config)?;

        config.heap.gc_unknown_items(&snapshot);

        #[cfg(feature = "testing")]
        {
            // these checks are in place to catch non-idempotent
            // recovery which could trigger feedback loops and
            // emergent behavior.
            trace!(
                "\n\n~~~~ regenerating snapshot for idempotency test ~~~~\n"
            );

            let snapshot2 = read_snapshot_or_default(&config)
                .expect("second read snapshot");
            assert_eq!(
                snapshot.active_segment, snapshot2.active_segment,
                "snapshot active_segment diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot, snapshot2
            );
            assert_eq!(
                snapshot.stable_lsn, snapshot2.stable_lsn,
                "snapshot stable_lsn diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot, snapshot2
            );
            for (pid, (p1, p2)) in
                snapshot.pt.iter().zip(snapshot2.pt.iter()).enumerate()
            {
                assert_eq!(
                    p1, p2,
                    "snapshot pid {} diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                    pid, p1, p2
                );
            }
            assert_eq!(
                snapshot.pt.len(),
                snapshot2.pt.len(),
                "snapshots number of pages diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot.pt,
                snapshot2.pt
            );
            assert_eq!(
                snapshot, snapshot2,
                "snapshots diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot, snapshot2
            );
        }

        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.start_pagecache);

        let cache_capacity = config.cache_capacity;
        let lru = Lru::new(cache_capacity);

        let mut pc = PageCacheInner {
            was_recovered: false,
            config: config.clone(),
            free: Arc::new(Mutex::new(BinaryHeap::new())),
            idgen: AtomicU64::new(0),
            idgen_persist_mu: Mutex::new(()),
            idgen_persists: AtomicU64::new(0),
            inner: PageTable::default(),
            log: Log::start(config, &snapshot)?,
            lru,
            next_pid_to_allocate: Mutex::new(0),
            snapshot_min_lsn: AtomicLsn::new(snapshot.stable_lsn.unwrap_or(0)),
            links: AtomicU64::new(0),
            snapshot_lock: Mutex::new(()),
        };

        // now we read it back in
        pc.load_snapshot(&snapshot)?;

        #[cfg(feature = "testing")]
        {
            // NB this must be before idgen/meta are initialized
            // because they may cas_page on initial page-in.
            let guard = pin();

            let mut pages_after_restart = Map::default();

            for pid in 0..*pc.next_pid_to_allocate.lock() {
                let pte = pc.inner.get(pid, &guard);
                let pointers =
                    pte.cache_infos.iter().map(|ci| ci.pointer).collect();
                pages_after_restart.insert(pid, pointers);
            }

            pc.config.event_log.pages_after_restart(pages_after_restart);
        }

        let mut was_recovered = true;

        {
            // subscope required because pc.begin() borrows pc

            let guard = pin();

            if !pc.inner.contains_pid(META_PID, &guard) {
                // set up meta
                was_recovered = false;

                let meta_update = Update::Meta(Meta::default());

                let (meta_id, _) = pc.allocate_inner(meta_update, &guard)?;

                assert_eq!(
                    meta_id, META_PID,
                    "we expect the meta page to have pid {}, but it had pid {} instead",
                    META_PID, meta_id,
                );
            }

            if !pc.inner.contains_pid(COUNTER_PID, &guard) {
                // set up idgen
                was_recovered = false;

                let counter_update = Update::Counter(0);

                let (counter_id, _) =
                    pc.allocate_inner(counter_update, &guard)?;

                assert_eq!(
                    counter_id, COUNTER_PID,
                    "we expect the counter to have pid {}, but it had pid {} instead",
                    COUNTER_PID, counter_id,
                );
            }

            let (idgen_key, counter) = pc.get_idgen(&guard);
            let idgen_recovery = if was_recovered {
                counter + (2 * pc.config.idgen_persist_interval)
            } else {
                // persist the meta and idgen pages now, so that we don't hand
                // out id 0 again if we crash and recover
                pc.flush()?;
                0
            };
            let idgen_persists = counter / pc.config.idgen_persist_interval
                * pc.config.idgen_persist_interval;

            pc.idgen.store(idgen_recovery, Release);
            pc.idgen_persists.store(idgen_persists, Release);

            if was_recovered {
                // advance pc.idgen_persists and the counter page by one
                // interval, so that when generate_id() is next called, it
                // will advance them further by another interval, and wait for
                // this update to be durable before returning the first ID.
                let necessary_persists =
                    (counter / pc.config.idgen_persist_interval + 1)
                        * pc.config.idgen_persist_interval;
                let counter_update = Update::Counter(necessary_persists);
                let old = pc.idgen_persists.swap(necessary_persists, Release);
                assert_eq!(old, idgen_persists);
                // CAS should never fail because the PageCache is still being constructed.
                pc.cas_page(
                    COUNTER_PID,
                    idgen_key,
                    counter_update,
                    false,
                    &guard,
                )?
                .unwrap();
            }
        }

        pc.was_recovered = was_recovered;

        #[cfg(feature = "event_log")]
        {
            let guard = pin();

            pc.config
                .event_log
                .meta_after_restart(pc.get_meta(&guard).deref().clone());
        }

        trace!("pagecache started");

        Ok(PageCache(Arc::new(pc)))
    }

    /// Try to atomically add a `PageLink` to the page.
    /// Returns `Ok(new_key)` if the operation was successful. Returns
    /// `Err(None)` if the page no longer exists. Returns
    /// `Err(Some(actual_key))` if the atomic link fails.
    pub(crate) fn link<'g>(
        &self,
        pid: PageId,
        mut old: PageView<'g>,
        new: Link,
        guard: &'g Guard,
    ) -> Result<CasResult<'g, Link>> {
        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.link_page);

        trace!("linking pid {} node {:?} with {:?}", pid, old.as_node(), new);

        // A failure injector that fails links randomly
        // during test to ensure interleaving coverage.
        #[cfg(any(test, feature = "lock_free_delays"))]
        {
            use std::cell::RefCell;
            use std::time::{SystemTime, UNIX_EPOCH};

            thread_local! {
                pub static COUNT: RefCell<u32> = RefCell::new(1);
            }

            let time_now =
                SystemTime::now().duration_since(UNIX_EPOCH).unwrap();

            #[allow(clippy::cast_possible_truncation)]
            let fail_seed = std::cmp::max(3, time_now.as_nanos() as u32 % 128);

            let inject_failure = COUNT.with(|c| {
                let mut cr = c.borrow_mut();
                *cr += 1;
                *cr % fail_seed == 0
            });

            if inject_failure {
                debug!(
                    "injecting a randomized failure in the link of pid {}",
                    pid
                );
                if let Some(current_pointer) = self.get(pid, guard)? {
                    return Ok(Err(Some((current_pointer.0, new))));
                } else {
                    return Ok(Err(None));
                }
            }
        }

        let node = old.as_node().apply(&new);

        // see if we should short-circuit replace
        if old.cache_infos.len() >= PAGE_CONSOLIDATION_THRESHOLD {
            log::trace!("skipping link, replacing pid {} with {:?}", pid, node);
            let short_circuit = self.replace(pid, old, &node, guard)?;
            return Ok(short_circuit.map_err(|a| a.map(|b| (b.0, new))));
        }

        log::trace!(
            "applying link of {:?} to pid {:?} resulted in node {:?}",
            new,
            pid,
            node
        );

        let mut new_page = Some(Owned::new(Page {
            update: Some(Update::Node(node)),
            cache_infos: Vec::default(),
        }));

        loop {
            // TODO handle replacement on threshold here instead

            let log_reservation =
                self.log.reserve(LogKind::Link, pid, &new, guard)?;
            let lsn = log_reservation.lsn;
            let pointer = log_reservation.pointer;

            // NB the setting of the timestamp is quite
            // correctness-critical! We use the ts to
            // ensure that fundamentally new data causes
            // high-level link and replace operations
            // to fail when the data in the pagecache
            // actually changes. When we just rewrite
            // the page for the purposes of moving it
            // to a new location on disk, however, we
            // don't want to cause threads that are
            // basing the correctness of their new
            // writes on the unchanged state to fail.
            // Here, we bump it by 1, to signal that
            // the underlying state is fundamentally
            // changing.
            let ts = old.ts() + 1;

            let cache_info = CacheInfo {
                lsn,
                pointer,
                ts,
                log_size: log_reservation.reservation_len() as u64,
            };

            let mut new_cache_infos =
                Vec::with_capacity(old.cache_infos.len() + 1);
            new_cache_infos.extend_from_slice(&old.cache_infos);
            new_cache_infos.push(cache_info);

            let mut page_ptr = new_page.take().unwrap();
            page_ptr.cache_infos = new_cache_infos;

            debug_delay();
            let result =
                old.entry.compare_and_set(old.read, page_ptr, SeqCst, guard);

            match result {
                Ok(new_shared) => {
                    trace!("link of pid {} succeeded", pid);

                    unsafe {
                        guard.defer_destroy(old.read);
                    }

                    assert_ne!(old.last_lsn(), 0);

                    self.log.iobufs.sa_mark_link(pid, cache_info, guard);

                    // NB complete must happen AFTER calls to SA, because
                    // when the iobuf's n_writers hits 0, we may transition
                    // the segment to inactive, resulting in a race otherwise.
                    // FIXME can result in deadlock if a node that holds SA
                    // is waiting to acquire a new reservation blocked by this?
                    log_reservation.complete()?;

                    // possibly evict an item now that our cache has grown
                    let total_page_size =
                        unsafe { new_shared.deref().log_size() };
                    let to_evict =
                        self.lru.accessed(pid, total_page_size, guard);
                    trace!(
                        "accessed pid {} -> paging out pids {:?}",
                        pid,
                        to_evict
                    );
                    if !to_evict.is_empty() {
                        self.page_out(to_evict, guard);
                    }

                    old.read = new_shared;

                    let link_count = self.links.fetch_add(1, Release);

                    if link_count > 0
                        && link_count % self.config.snapshot_after_ops == 0
                    {
                        let s2: PageCache = self.clone();
                        threadpool::spawn(move || {
                            if let Err(e) = s2.take_fuzzy_snapshot() {
                                log::error!(
                                    "failed to write snapshot: {:?}",
                                    e
                                );
                            }
                        })?;
                    }

                    return Ok(Ok(old));
                }
                Err(cas_error) => {
                    log_reservation.abort()?;
                    let actual = cas_error.current;
                    let actual_ts = unsafe { actual.deref().ts() };
                    if actual_ts == old.ts() {
                        trace!(
                            "link of pid {} failed due to movement, retrying",
                            pid
                        );
                        new_page = Some(cas_error.new);

                        old.read = actual;
                    } else {
                        trace!("link of pid {} failed due to new update", pid);
                        let mut page_view = old;
                        page_view.read = actual;
                        return Ok(Err(Some((page_view, new))));
                    }
                }
            }
        }
    }

    fn take_fuzzy_snapshot(self) -> Result<()> {
        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.fuzzy_snapshot);
        let lock = self.snapshot_lock.try_lock();
        if lock.is_none() {
            log::debug!(
                "skipping snapshot because the snapshot lock is already claimed"
            );
            return Ok(());
        }
        let stable_lsn_before: Lsn = self.log.stable_offset();

        // This is how we determine the number of the pages we will snapshot.
        let pid_bound = *self.next_pid_to_allocate.lock();

        let pid_bound_usize = assert_usize(pid_bound);

        let mut page_states = Vec::<PageState>::with_capacity(pid_bound_usize);
        let mut guard = pin();
        for pid in 0..pid_bound {
            if pid % 64 == 0 {
                drop(guard);
                guard = pin();
            }
            'inner: loop {
                let pg_view = self.inner.get(pid, &guard);
                if pg_view.cache_infos.is_empty() {
                    // there is a benign race with the thread
                    // that is allocating this page. the allocating
                    // thread has not yet written the new page to disk,
                    // and it does not yet have any storage tracking
                    // information.
                    std::thread::yield_now();
                } else {
                    let page_state = pg_view.to_page_state();
                    page_states.push(page_state);
                    break 'inner;
                }
            }
        }

        let max_reserved_lsn_after: Lsn =
            self.log.iobufs.max_reserved_lsn.load(Acquire);

        let snapshot = Snapshot {
            stable_lsn: Some(stable_lsn_before),
            active_segment: None,
            pt: page_states,
        };

        self.log.make_stable(max_reserved_lsn_after)?;

        snapshot::write_snapshot(&self.config, &snapshot)?;

        // NB: this must only happen after writing the snapshot to disk
        bump_atomic_lsn(&self.snapshot_min_lsn, stable_lsn_before);<commit_msg>Drop fuzzy snapshot ebr guard before performing any IO<commit_after>//! `pagecache` is a lock-free pagecache and log for building high-performance
//! databases.
#![allow(unsafe_code)]

pub mod constants;
pub mod logger;

mod disk_pointer;
mod header;
mod heap;
mod iobuf;
mod iterator;
mod pagetable;
#[cfg(any(all(not(unix), not(windows)), miri))]
mod parallel_io_polyfill;
#[cfg(all(unix, not(miri)))]
mod parallel_io_unix;
#[cfg(all(windows, not(miri)))]
mod parallel_io_windows;
mod reservation;
mod segment;
mod snapshot;

use std::{collections::BinaryHeap, ops::Deref};

use crate::*;

#[cfg(any(all(not(unix), not(windows)), miri))]
use parallel_io_polyfill::{pread_exact, pread_exact_or_eof, pwrite_all};

#[cfg(all(unix, not(miri)))]
use parallel_io_unix::{pread_exact, pread_exact_or_eof, pwrite_all};

#[cfg(all(windows, not(miri)))]
use parallel_io_windows::{pread_exact, pread_exact_or_eof, pwrite_all};

use self::{
    constants::{
        BATCH_MANIFEST_PID, COUNTER_PID, META_PID,
        PAGE_CONSOLIDATION_THRESHOLD, SEGMENT_CLEANUP_THRESHOLD,
    },
    header::Header,
    iobuf::{roll_iobuf, IoBuf, IoBufs},
    iterator::{raw_segment_iter_from, LogIter},
    pagetable::PageTable,
    segment::{SegmentAccountant, SegmentCleaner, SegmentOp},
};

pub(crate) use self::{
    heap::{Heap, HeapId},
    logger::{
        read_message, read_segment_header, MessageHeader, SegmentHeader,
        SegmentNumber,
    },
    reservation::Reservation,
    snapshot::{read_snapshot_or_default, PageState, Snapshot},
};

pub use self::{
    constants::{MAX_MSG_HEADER_LEN, MAX_SPACE_AMPLIFICATION, SEG_HEADER_LEN},
    disk_pointer::DiskPtr,
    logger::{Log, LogRead},
};

/// A file offset in the database log.
pub type LogOffset = u64;

/// The logical sequence number of an item in the database log.
pub type Lsn = i64;

/// A page identifier.
pub type PageId = u64;

/// Uses a non-varint `Lsn` to mark offsets.
#[derive(Default, Clone, Copy, Ord, PartialOrd, Eq, PartialEq, Debug)]
#[repr(transparent)]
pub struct BatchManifest(pub Lsn);

/// A buffer with an associated offset. Useful for
/// batching many reads over a file segment.
#[derive(Debug)]
pub struct BasedBuf {
    pub buf: Vec<u8>,
    pub offset: LogOffset,
}

/// A byte used to disambiguate log message types
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
#[repr(u8)]
pub enum MessageKind {
    /// The EVIL_BYTE is written as a canary to help
    /// detect torn writes.
    Corrupted = 0,
    /// Indicates that the following buffer corresponds
    /// to a reservation for an in-memory operation that
    /// failed to complete. It should be skipped during
    /// recovery.
    Canceled = 1,
    /// Indicates that the following buffer is used
    /// as padding to fill out the rest of the segment
    /// before sealing it.
    Cap = 2,
    /// Indicates that the following buffer contains
    /// an Lsn for the last write in an atomic writebatch.
    BatchManifest = 3,
    /// Indicates that this page was freed from the pagetable.
    Free = 4,
    /// Indicates that the last persisted ID was at least
    /// this high.
    Counter = 5,
    /// The meta page, stored inline
    InlineMeta = 6,
    /// The meta page, stored heaply
    HeapMeta = 7,
    /// A consolidated page replacement, stored inline
    InlineNode = 8,
    /// A consolidated page replacement, stored heaply
    HeapNode = 9,
    /// A partial page update, stored inline
    InlineLink = 10,
    /// A partial page update, stored heaply
    HeapLink = 11,
}

impl MessageKind {
    pub(crate) const fn into(self) -> u8 {
        self as u8
    }
}

impl From<u8> for MessageKind {
    fn from(byte: u8) -> Self {
        use MessageKind::*;
        match byte {
            0 => Corrupted,
            1 => Canceled,
            2 => Cap,
            3 => BatchManifest,
            4 => Free,
            5 => Counter,
            6 => InlineMeta,
            7 => HeapMeta,
            8 => InlineNode,
            9 => HeapNode,
            10 => InlineLink,
            11 => HeapLink,
            other => {
                debug!("encountered unexpected message kind byte {}", other);
                Corrupted
            }
        }
    }
}

/// The high-level types of stored information
/// about pages and their mutations
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LogKind {
    /// Persisted data containing a page replacement
    Replace,
    /// Persisted immutable update
    Link,
    /// Freeing of a page
    Free,
    /// Some state indicating this should be skipped
    Skip,
    /// Unexpected corruption
    Corrupted,
}

fn log_kind_from_update(update: &Update) -> LogKind {
    match update {
        Update::Free => LogKind::Free,
        Update::Link(..) => LogKind::Link,
        Update::Node(..) | Update::Counter(..) | Update::Meta(..) => {
            LogKind::Replace
        }
    }
}

impl From<MessageKind> for LogKind {
    fn from(kind: MessageKind) -> Self {
        match kind {
            MessageKind::Free => LogKind::Free,
            MessageKind::InlineNode
            | MessageKind::Counter
            | MessageKind::HeapNode
            | MessageKind::InlineMeta
            | MessageKind::HeapMeta => LogKind::Replace,
            MessageKind::InlineLink | MessageKind::HeapLink => LogKind::Link,
            MessageKind::Canceled
            | MessageKind::Cap
            | MessageKind::BatchManifest => LogKind::Skip,
            other => {
                debug!("encountered unexpected message kind byte {:?}", other);
                LogKind::Corrupted
            }
        }
    }
}

fn assert_usize<T>(from: T) -> usize
where
    usize: TryFrom<T, Error = std::num::TryFromIntError>,
{
    usize::try_from(from).expect("lost data cast while converting to usize")
}

// TODO remove this with atomic fetch_max if we increase MSRV to 1.45 or higher
fn bump_atomic_lsn(atomic_lsn: &AtomicLsn, to: Lsn) {
    let mut current = atomic_lsn.load(Acquire);
    loop {
        if current >= to {
            return;
        }
        let last =
            atomic_lsn.compare_exchange_weak(current, to, SeqCst, SeqCst);
        if last.is_ok() {
            // we succeeded.
            return;
        }
        current = last.unwrap_err();
    }
}

use std::convert::{TryFrom, TryInto};

#[inline]
pub(crate) fn lsn_to_arr(number: Lsn) -> [u8; 8] {
    number.to_le_bytes()
}

#[inline]
pub(crate) fn arr_to_lsn(arr: &[u8]) -> Lsn {
    Lsn::from_le_bytes(arr.try_into().unwrap())
}

#[inline]
pub(crate) fn u64_to_arr(number: u64) -> [u8; 8] {
    number.to_le_bytes()
}

#[inline]
pub(crate) fn arr_to_u32(arr: &[u8]) -> u32 {
    u32::from_le_bytes(arr.try_into().unwrap())
}

#[inline]
pub(crate) fn u32_to_arr(number: u32) -> [u8; 4] {
    number.to_le_bytes()
}

#[allow(clippy::needless_pass_by_value)]
#[allow(clippy::needless_return)]
pub(crate) fn decompress(in_buf: Vec<u8>) -> Vec<u8> {
    #[cfg(feature = "compression")]
    {
        use zstd::stream::decode_all;

        let scootable_in_buf = &mut &*in_buf;
        let raw: IVec = IVec::deserialize(scootable_in_buf)
            .expect("this had to be serialized with an extra length frame");
        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.decompress);
        let out_buf = decode_all(&raw[..]).expect(
            "failed to decompress data. \
             This is not expected, please open an issue on \
             https://github.com/spacejam/sled so we can \
             fix this critical issue ASAP. Thank you :)",
        );

        return out_buf;
    }

    #[cfg(not(feature = "compression"))]
    in_buf
}

#[derive(Debug, Clone, Copy)]
pub struct NodeView<'g>(pub(crate) PageView<'g>);

impl<'g> Deref for NodeView<'g> {
    type Target = Node;
    fn deref(&self) -> &Node {
        self.0.as_node()
    }
}

#[derive(Debug, Clone, Copy)]
pub struct MetaView<'g>(PageView<'g>);

impl<'g> Deref for MetaView<'g> {
    type Target = Meta;
    fn deref(&self) -> &Meta {
        self.0.as_meta()
    }
}

#[derive(Debug, Clone, Copy)]
pub struct PageView<'g> {
    pub(crate) read: Shared<'g, Page>,
    pub(crate) entry: &'g Atomic<Page>,
}

impl<'g> Deref for PageView<'g> {
    type Target = Page;

    fn deref(&self) -> &Page {
        unsafe { self.read.deref() }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CacheInfo {
    pub ts: u64,
    pub lsn: Lsn,
    pub pointer: DiskPtr,
    pub log_size: u64,
}

#[cfg(test)]
impl quickcheck::Arbitrary for CacheInfo {
    fn arbitrary<G: quickcheck::Gen>(g: &mut G) -> CacheInfo {
        use rand::Rng;

        CacheInfo {
            ts: g.gen(),
            lsn: g.gen(),
            pointer: DiskPtr::arbitrary(g),
            log_size: g.gen(),
        }
    }
}

/// Update<PageLinkment> denotes a state or a change in a sequence of updates
/// of which a page consists.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "testing", derive(PartialEq))]
pub(crate) enum Update {
    Link(Link),
    Node(Node),
    Free,
    Counter(u64),
    Meta(Meta),
}

impl Update {
    fn as_node(&self) -> &Node {
        match self {
            Update::Node(node) => node,
            other => panic!("called as_node on non-Node: {:?}", other),
        }
    }

    fn as_node_mut(&mut self) -> &mut Node {
        match self {
            Update::Node(node) => node,
            other => panic!("called as_node_mut on non-Node: {:?}", other),
        }
    }

    fn as_link(&self) -> &Link {
        match self {
            Update::Link(link) => link,
            other => panic!("called as_link on non-Link: {:?}", other),
        }
    }

    fn as_meta(&self) -> &Meta {
        if let Update::Meta(meta) = self {
            meta
        } else {
            panic!("called as_meta on {:?}", self)
        }
    }

    fn as_counter(&self) -> u64 {
        if let Update::Counter(counter) = self {
            *counter
        } else {
            panic!("called as_counter on {:?}", self)
        }
    }
}

/// Ensures that any operations that are written to disk between the
/// creation of this guard and its destruction will be recovered
/// atomically. When this guard is dropped, it marks in an earlier
/// reservation where the stable tip must be in order to perform
/// recovery. If this is beyond where the system successfully
/// wrote before crashing, then the recovery will stop immediately
/// before any of the atomic batch can be partially recovered.
///
/// Must call `seal_batch` to complete the atomic batch operation.
///
/// If this is dropped without calling `seal_batch`, the complete
/// recovery effect will not occur.
#[derive(Debug)]
pub struct RecoveryGuard<'a> {
    batch_res: Reservation<'a>,
}

impl<'a> RecoveryGuard<'a> {
    /// Writes the last LSN for a batch into an earlier
    /// reservation, releasing it.
    pub(crate) fn seal_batch(self) -> Result<()> {
        let max_reserved =
            self.batch_res.log.iobufs.max_reserved_lsn.load(Acquire);
        self.batch_res.mark_writebatch(max_reserved).map(|_| ())
    }
}

/// A page consists of a sequence of state transformations
/// with associated storage parameters like disk pos, lsn, time.
#[derive(Debug, Clone)]
pub struct Page {
    update: Option<Update>,
    cache_infos: Vec<CacheInfo>,
}

impl Page {
    fn to_page_state(&self) -> PageState {
        let base = &self.cache_infos[0];
        if self.is_free() {
            PageState::Free(base.lsn, base.pointer)
        } else {
            let mut frags: Vec<(Lsn, DiskPtr, u64)> = vec![];

            for cache_info in self.cache_infos.iter().skip(1) {
                frags.push((
                    cache_info.lsn,
                    cache_info.pointer,
                    cache_info.log_size,
                ));
            }

            PageState::Present {
                base: (base.lsn, base.pointer, base.log_size),
                frags,
            }
        }
    }

    fn as_node(&self) -> &Node {
        self.update.as_ref().unwrap().as_node()
    }

    fn as_meta(&self) -> &Meta {
        self.update.as_ref().unwrap().as_meta()
    }

    fn as_counter(&self) -> u64 {
        self.update.as_ref().unwrap().as_counter()
    }

    fn is_free(&self) -> bool {
        if let Some(Update::Free) = self.update {
            true
        } else {
            false
        }
    }

    fn last_lsn(&self) -> Lsn {
        self.cache_infos.last().map(|ci| ci.lsn).unwrap()
    }

    pub(crate) fn log_size(&self) -> u64 {
        self.cache_infos.iter().map(|ci| ci.log_size).sum()
    }

    fn ts(&self) -> u64 {
        self.cache_infos.last().map_or(0, |ci| ci.ts)
    }

    fn lone_heap_item(&self) -> Option<CacheInfo> {
        if self.cache_infos.len() == 1
            && self.cache_infos[0].pointer.is_heap_item()
        {
            Some(self.cache_infos[0])
        } else {
            None
        }
    }
}

/// A lock-free pagecache which supports linkmented pages
/// for dramatically improving write throughput.
#[derive(Clone)]
pub struct PageCache(Arc<PageCacheInner>);

impl Deref for PageCache {
    type Target = PageCacheInner;

    fn deref(&self) -> &PageCacheInner {
        &self.0
    }
}

pub struct PageCacheInner {
    was_recovered: bool,
    pub(crate) config: RunningConfig,
    inner: PageTable,
    next_pid_to_allocate: Mutex<PageId>,
    // needs to be a sub-Arc because we separate
    // it for async modification in an EBR guard
    free: Arc<Mutex<BinaryHeap<PageId>>>,
    #[doc(hidden)]
    pub log: Log,
    lru: Lru,

    idgen: AtomicU64,
    idgen_persists: AtomicU64,
    idgen_persist_mu: Mutex<()>,

    // fuzzy snapshot-related items
    snapshot_min_lsn: AtomicLsn,
    links: AtomicU64,
    snapshot_lock: Mutex<()>,
}

impl Debug for PageCache {
    fn fmt(
        &self,
        f: &mut fmt::Formatter<'_>,
    ) -> std::result::Result<(), fmt::Error> {
        f.write_str(&*format!(
            "PageCache {{ max: {:?} free: {:?} }}\n",
            *self.next_pid_to_allocate.lock(),
            self.free
        ))
    }
}

#[cfg(feature = "event_log")]
impl Drop for PageCacheInner {
    fn drop(&mut self) {
        trace!("dropping pagecache");

        // we can't as easily assert recovery
        // invariants across failpoints for now
        if self.log.iobufs.config.global_error().is_ok() {
            let mut pages_before_restart = Map::default();

            let guard = pin();

            self.config
                .event_log
                .meta_before_restart(self.get_meta(&guard).deref().clone());

            for pid in 0..*self.next_pid_to_allocate.lock() {
                let pte = self.inner.get(pid, &guard);
                let pointers =
                    pte.cache_infos.iter().map(|ci| ci.pointer).collect();
                pages_before_restart.insert(pid, pointers);
            }

            self.config.event_log.pages_before_restart(pages_before_restart);
        }

        trace!("pagecache dropped");
    }
}

impl PageCache {
    /// Instantiate a new `PageCache`.
    pub(crate) fn start(config: RunningConfig) -> Result<PageCache> {
        trace!("starting pagecache");

        config.reset_global_error();

        // try to pull any existing snapshot off disk, and
        // apply any new data to it to "catch-up" the
        // snapshot before loading it.
        let snapshot = read_snapshot_or_default(&config)?;

        config.heap.gc_unknown_items(&snapshot);

        #[cfg(feature = "testing")]
        {
            // these checks are in place to catch non-idempotent
            // recovery which could trigger feedback loops and
            // emergent behavior.
            trace!(
                "\n\n~~~~ regenerating snapshot for idempotency test ~~~~\n"
            );

            let snapshot2 = read_snapshot_or_default(&config)
                .expect("second read snapshot");
            assert_eq!(
                snapshot.active_segment, snapshot2.active_segment,
                "snapshot active_segment diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot, snapshot2
            );
            assert_eq!(
                snapshot.stable_lsn, snapshot2.stable_lsn,
                "snapshot stable_lsn diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot, snapshot2
            );
            for (pid, (p1, p2)) in
                snapshot.pt.iter().zip(snapshot2.pt.iter()).enumerate()
            {
                assert_eq!(
                    p1, p2,
                    "snapshot pid {} diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                    pid, p1, p2
                );
            }
            assert_eq!(
                snapshot.pt.len(),
                snapshot2.pt.len(),
                "snapshots number of pages diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot.pt,
                snapshot2.pt
            );
            assert_eq!(
                snapshot, snapshot2,
                "snapshots diverged across recoveries.\n\n \
                first: {:?}\n\n
                second: {:?}\n\n",
                snapshot, snapshot2
            );
        }

        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.start_pagecache);

        let cache_capacity = config.cache_capacity;
        let lru = Lru::new(cache_capacity);

        let mut pc = PageCacheInner {
            was_recovered: false,
            config: config.clone(),
            free: Arc::new(Mutex::new(BinaryHeap::new())),
            idgen: AtomicU64::new(0),
            idgen_persist_mu: Mutex::new(()),
            idgen_persists: AtomicU64::new(0),
            inner: PageTable::default(),
            log: Log::start(config, &snapshot)?,
            lru,
            next_pid_to_allocate: Mutex::new(0),
            snapshot_min_lsn: AtomicLsn::new(snapshot.stable_lsn.unwrap_or(0)),
            links: AtomicU64::new(0),
            snapshot_lock: Mutex::new(()),
        };

        // now we read it back in
        pc.load_snapshot(&snapshot)?;

        #[cfg(feature = "testing")]
        {
            // NB this must be before idgen/meta are initialized
            // because they may cas_page on initial page-in.
            let guard = pin();

            let mut pages_after_restart = Map::default();

            for pid in 0..*pc.next_pid_to_allocate.lock() {
                let pte = pc.inner.get(pid, &guard);
                let pointers =
                    pte.cache_infos.iter().map(|ci| ci.pointer).collect();
                pages_after_restart.insert(pid, pointers);
            }

            pc.config.event_log.pages_after_restart(pages_after_restart);
        }

        let mut was_recovered = true;

        {
            // subscope required because pc.begin() borrows pc

            let guard = pin();

            if !pc.inner.contains_pid(META_PID, &guard) {
                // set up meta
                was_recovered = false;

                let meta_update = Update::Meta(Meta::default());

                let (meta_id, _) = pc.allocate_inner(meta_update, &guard)?;

                assert_eq!(
                    meta_id, META_PID,
                    "we expect the meta page to have pid {}, but it had pid {} instead",
                    META_PID, meta_id,
                );
            }

            if !pc.inner.contains_pid(COUNTER_PID, &guard) {
                // set up idgen
                was_recovered = false;

                let counter_update = Update::Counter(0);

                let (counter_id, _) =
                    pc.allocate_inner(counter_update, &guard)?;

                assert_eq!(
                    counter_id, COUNTER_PID,
                    "we expect the counter to have pid {}, but it had pid {} instead",
                    COUNTER_PID, counter_id,
                );
            }

            let (idgen_key, counter) = pc.get_idgen(&guard);
            let idgen_recovery = if was_recovered {
                counter + (2 * pc.config.idgen_persist_interval)
            } else {
                // persist the meta and idgen pages now, so that we don't hand
                // out id 0 again if we crash and recover
                pc.flush()?;
                0
            };
            let idgen_persists = counter / pc.config.idgen_persist_interval
                * pc.config.idgen_persist_interval;

            pc.idgen.store(idgen_recovery, Release);
            pc.idgen_persists.store(idgen_persists, Release);

            if was_recovered {
                // advance pc.idgen_persists and the counter page by one
                // interval, so that when generate_id() is next called, it
                // will advance them further by another interval, and wait for
                // this update to be durable before returning the first ID.
                let necessary_persists =
                    (counter / pc.config.idgen_persist_interval + 1)
                        * pc.config.idgen_persist_interval;
                let counter_update = Update::Counter(necessary_persists);
                let old = pc.idgen_persists.swap(necessary_persists, Release);
                assert_eq!(old, idgen_persists);
                // CAS should never fail because the PageCache is still being constructed.
                pc.cas_page(
                    COUNTER_PID,
                    idgen_key,
                    counter_update,
                    false,
                    &guard,
                )?
                .unwrap();
            }
        }

        pc.was_recovered = was_recovered;

        #[cfg(feature = "event_log")]
        {
            let guard = pin();

            pc.config
                .event_log
                .meta_after_restart(pc.get_meta(&guard).deref().clone());
        }

        trace!("pagecache started");

        Ok(PageCache(Arc::new(pc)))
    }

    /// Try to atomically add a `PageLink` to the page.
    /// Returns `Ok(new_key)` if the operation was successful. Returns
    /// `Err(None)` if the page no longer exists. Returns
    /// `Err(Some(actual_key))` if the atomic link fails.
    pub(crate) fn link<'g>(
        &self,
        pid: PageId,
        mut old: PageView<'g>,
        new: Link,
        guard: &'g Guard,
    ) -> Result<CasResult<'g, Link>> {
        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.link_page);

        trace!("linking pid {} node {:?} with {:?}", pid, old.as_node(), new);

        // A failure injector that fails links randomly
        // during test to ensure interleaving coverage.
        #[cfg(any(test, feature = "lock_free_delays"))]
        {
            use std::cell::RefCell;
            use std::time::{SystemTime, UNIX_EPOCH};

            thread_local! {
                pub static COUNT: RefCell<u32> = RefCell::new(1);
            }

            let time_now =
                SystemTime::now().duration_since(UNIX_EPOCH).unwrap();

            #[allow(clippy::cast_possible_truncation)]
            let fail_seed = std::cmp::max(3, time_now.as_nanos() as u32 % 128);

            let inject_failure = COUNT.with(|c| {
                let mut cr = c.borrow_mut();
                *cr += 1;
                *cr % fail_seed == 0
            });

            if inject_failure {
                debug!(
                    "injecting a randomized failure in the link of pid {}",
                    pid
                );
                if let Some(current_pointer) = self.get(pid, guard)? {
                    return Ok(Err(Some((current_pointer.0, new))));
                } else {
                    return Ok(Err(None));
                }
            }
        }

        let node = old.as_node().apply(&new);

        // see if we should short-circuit replace
        if old.cache_infos.len() >= PAGE_CONSOLIDATION_THRESHOLD {
            log::trace!("skipping link, replacing pid {} with {:?}", pid, node);
            let short_circuit = self.replace(pid, old, &node, guard)?;
            return Ok(short_circuit.map_err(|a| a.map(|b| (b.0, new))));
        }

        log::trace!(
            "applying link of {:?} to pid {:?} resulted in node {:?}",
            new,
            pid,
            node
        );

        let mut new_page = Some(Owned::new(Page {
            update: Some(Update::Node(node)),
            cache_infos: Vec::default(),
        }));

        loop {
            // TODO handle replacement on threshold here instead

            let log_reservation =
                self.log.reserve(LogKind::Link, pid, &new, guard)?;
            let lsn = log_reservation.lsn;
            let pointer = log_reservation.pointer;

            // NB the setting of the timestamp is quite
            // correctness-critical! We use the ts to
            // ensure that fundamentally new data causes
            // high-level link and replace operations
            // to fail when the data in the pagecache
            // actually changes. When we just rewrite
            // the page for the purposes of moving it
            // to a new location on disk, however, we
            // don't want to cause threads that are
            // basing the correctness of their new
            // writes on the unchanged state to fail.
            // Here, we bump it by 1, to signal that
            // the underlying state is fundamentally
            // changing.
            let ts = old.ts() + 1;

            let cache_info = CacheInfo {
                lsn,
                pointer,
                ts,
                log_size: log_reservation.reservation_len() as u64,
            };

            let mut new_cache_infos =
                Vec::with_capacity(old.cache_infos.len() + 1);
            new_cache_infos.extend_from_slice(&old.cache_infos);
            new_cache_infos.push(cache_info);

            let mut page_ptr = new_page.take().unwrap();
            page_ptr.cache_infos = new_cache_infos;

            debug_delay();
            let result =
                old.entry.compare_and_set(old.read, page_ptr, SeqCst, guard);

            match result {
                Ok(new_shared) => {
                    trace!("link of pid {} succeeded", pid);

                    unsafe {
                        guard.defer_destroy(old.read);
                    }

                    assert_ne!(old.last_lsn(), 0);

                    self.log.iobufs.sa_mark_link(pid, cache_info, guard);

                    // NB complete must happen AFTER calls to SA, because
                    // when the iobuf's n_writers hits 0, we may transition
                    // the segment to inactive, resulting in a race otherwise.
                    // FIXME can result in deadlock if a node that holds SA
                    // is waiting to acquire a new reservation blocked by this?
                    log_reservation.complete()?;

                    // possibly evict an item now that our cache has grown
                    let total_page_size =
                        unsafe { new_shared.deref().log_size() };
                    let to_evict =
                        self.lru.accessed(pid, total_page_size, guard);
                    trace!(
                        "accessed pid {} -> paging out pids {:?}",
                        pid,
                        to_evict
                    );
                    if !to_evict.is_empty() {
                        self.page_out(to_evict, guard);
                    }

                    old.read = new_shared;

                    let link_count = self.links.fetch_add(1, Release);

                    if link_count > 0
                        && link_count % self.config.snapshot_after_ops == 0
                    {
                        let s2: PageCache = self.clone();
                        threadpool::spawn(move || {
                            if let Err(e) = s2.take_fuzzy_snapshot() {
                                log::error!(
                                    "failed to write snapshot: {:?}",
                                    e
                                );
                            }
                        })?;
                    }

                    return Ok(Ok(old));
                }
                Err(cas_error) => {
                    log_reservation.abort()?;
                    let actual = cas_error.current;
                    let actual_ts = unsafe { actual.deref().ts() };
                    if actual_ts == old.ts() {
                        trace!(
                            "link of pid {} failed due to movement, retrying",
                            pid
                        );
                        new_page = Some(cas_error.new);

                        old.read = actual;
                    } else {
                        trace!("link of pid {} failed due to new update", pid);
                        let mut page_view = old;
                        page_view.read = actual;
                        return Ok(Err(Some((page_view, new))));
                    }
                }
            }
        }
    }

    fn take_fuzzy_snapshot(self) -> Result<()> {
        #[cfg(feature = "metrics")]
        let _measure = Measure::new(&M.fuzzy_snapshot);
        let lock = self.snapshot_lock.try_lock();
        if lock.is_none() {
            log::debug!(
                "skipping snapshot because the snapshot lock is already claimed"
            );
            return Ok(());
        }
        let stable_lsn_before: Lsn = self.log.stable_offset();

        // This is how we determine the number of the pages we will snapshot.
        let pid_bound = *self.next_pid_to_allocate.lock();

        let pid_bound_usize = assert_usize(pid_bound);

        let mut page_states = Vec::<PageState>::with_capacity(pid_bound_usize);
        let mut guard = pin();
        for pid in 0..pid_bound {
            if pid % 64 == 0 {
                drop(guard);
                guard = pin();
            }
            'inner: loop {
                let pg_view = self.inner.get(pid, &guard);
                if pg_view.cache_infos.is_empty() {
                    // there is a benign race with the thread
                    // that is allocating this page. the allocating
                    // thread has not yet written the new page to disk,
                    // and it does not yet have any storage tracking
                    // information.
                    std::thread::yield_now();
                } else {
                    let page_state = pg_view.to_page_state();
                    page_states.push(page_state);
                    break 'inner;
                }
            }
        }
        drop(guard);

        let max_reserved_lsn_after: Lsn =
            self.log.iobufs.max_reserved_lsn.load(Acquire);

        let snapshot = Snapshot {
            stable_lsn: Some(stable_lsn_before),
            active_segment: None,
            pt: page_states,
        };

        self.log.make_stable(max_reserved_lsn_after)?;

        snapshot::write_snapshot(&self.config, &snapshot)?;

        // NB: this must only happen after writing the snapshot to disk
        bump_atomic_lsn(&self.snapshot_min_lsn, stable_lsn_before);
{'fraction_non_alphanumeric': 0.11617446037447945, 'fraction_numerical': 0.005118097720698096, 'mean_word_length': 1.6879859864991882, 'pattern_counts': {'":': 0, '<': 99, '<?xml version=': 0, '>': 235, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1417912', 'n_tokens_mistral': 18792, 'n_tokens_neox': 18149, 'n_words': 4692}