yo-kv 0.3.20

//! Variable length bytes belonging to one collection, back to back.
//!
//! Two things inside a collection are the same problem. A set or a hash interns
//! its member and field names so that writing the same field again touches no
//! name bytes (`05` section 3), and a hash in the native band has to put its
//! values somewhere that is not one allocation per field, because one allocation
//! per field is the thing the element per row layout exists to avoid. Both want
//! a stretch of bytes with an offset handed back, both want a rewrite to leave
//! the old bytes behind rather than move everything after them, and both want
//! those bytes back eventually.
//!
//! ```text
//!   bytes                                    dead
//! +-------+---------+-------+---------+     bytes nothing points at, counted
//! | name  | oldval  | name  | value   |     here and given back once they
//! +-------+---------+-------+---------+     outnumber the ones that are live
//!    ^ at, len              ^ at, len
//! ```
//!
//! # It does not hold the references
//!
//! [`Blob::push`] hands back an offset and nothing else, and [`Blob::read`] takes
//! an offset and a length. The reference is the caller's to shape, which is not
//! ceremony, because the two callers want different shapes and neither of them
//! is a compromise with the other.
//!
//! [`crate::Elements`] has eight spare bits in a row it is trying to keep at
//! eight bytes, so it keeps a name's length in those and its reference is the
//! offset on its own. A hash value has no spare bits anywhere and cannot be
//! capped at two hundred and fifty five bytes either, because Redis lets one be
//! half a gigabyte, so [`Blob::push_sized`] writes the length into the blob in
//! front of the bytes and its reference is also the offset on its own. One byte
//! of prefix for a value under two hundred and fifty five bytes and five for
//! anything longer, which is a byte per field against the four a length beside
//! the offset would cost.
//!
//! [`Span`] is here for the callers that have no reason to pack it tighter.
//!
//! # Giving the dead bytes back
//!
//! A rewrite appends and abandons, so a hash whose values are written over and
//! over holds every value it ever had until something clears up. That something
//! is [`Blob::compact`], which the owner runs when [`Blob::worth_compacting`]
//! says so and drives itself, because the owner is the only thing that knows
//! where its references are. Until then the dead bytes are counted and reported
//! rather than pretended away, which is the rule the arena follows and for the
//! same reason: a number `INFO memory` can show is a leak you can see.
//!
//! Half is the line, with a floor of four kilobytes under it. Below the half the
//! copy costs more than the bytes are worth, and below the floor there are not
//! enough bytes to be worth a copy at any ratio at all.

/// Where something is in a blob, for a caller with no reason to pack it tighter.
///
/// Eight bytes. A hash value uses this, because a value can be as long as the
/// 512 MiB Redis puts on everything and there is no shorter length that holds
/// it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Span {
    /// Where the bytes start.
    pub at: u32,
    /// How many of them there are.
    pub len: u32,
}

/// Dead bytes below this are left alone whatever the ratio says.
const FLOOR: usize = 4096;

/// The shortest run whose length goes in front of it in five bytes, not one.
const LONG: usize = 255;

/// How many bytes a long run's length prefix takes, the marker included.
const LONG_PREFIX: usize = 5;

/// How far past `at` the bytes start, and how many of them there are.
///
/// The prefix is one byte holding the length, or the marker followed by the
/// length in four. Two hundred and fifty five is the marker rather than a
/// length, so a run of exactly that many bytes takes the long form and pays four
/// bytes it did not have to. That is one length out of the whole range and it
/// buys a check that is a compare against a constant.
#[inline]
fn sized_head(bytes: &[u8], at: usize) -> (usize, usize) {
    let head = usize::from(bytes[at]);
    if head < LONG {
        return (1, head);
    }
    let head: [u8; 4] = bytes[at + 1..at + LONG_PREFIX]
        .try_into()
        .expect("four bytes of length behind the marker");
    (LONG_PREFIX, u32::from_le_bytes(head) as usize)
}

/// Bytes belonging to one collection, appended to and occasionally rebuilt.
#[derive(Debug, Default, Clone)]
pub struct Blob {
    bytes: Vec<u8>,
    dead: usize,
}

impl Blob {
    /// An empty blob that has not allocated anything.
    ///
    /// A collection is made by its first write, so the empty case is the common
    /// one and it does not deserve an allocation.
    #[must_use]
    pub const fn new() -> Blob {
        Blob {
            bytes: Vec::new(),
            dead: 0,
        }
    }

    /// An empty blob with room already taken.
    #[must_use]
    pub fn with_capacity(n: usize) -> Blob {
        Blob {
            bytes: Vec::with_capacity(n),
            dead: 0,
        }
    }

    /// Every byte here, live and dead together.
    #[inline]
    #[must_use]
    pub const fn len(&self) -> usize {
        self.bytes.len()
    }

    /// Whether nothing has ever been written.
    #[inline]
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.bytes.is_empty()
    }

    /// Bytes nothing points at any more.
    #[inline]
    #[must_use]
    pub const fn dead(&self) -> usize {
        self.dead
    }

    /// What this costs, which is the allocation and not the used part of it.
    #[inline]
    #[must_use]
    pub fn memory_bytes(&self) -> usize {
        self.bytes.capacity()
    }

    /// Append `bytes` and say where they went.
    ///
    /// # Panics
    ///
    /// If the blob would pass four gigabytes, which no collection reaches
    /// without passing a row limit first.
    #[inline]
    pub fn push(&mut self, bytes: &[u8]) -> u32 {
        let at = u32::try_from(self.bytes.len()).expect("the blob is under 4 GiB");
        // By [`crate::grow`]'s policy and not by `Vec`'s, for the same reason
        // the row array above it grows that way: a blob holding the names of a
        // large collection is megabytes, and half of a doubled one is air.
        crate::grow::reserve(&mut self.bytes, bytes.len());
        self.bytes.extend_from_slice(bytes);
        at
    }

    /// Append `bytes` and say where they went, as a [`Span`].
    ///
    /// # Panics
    ///
    /// If the blob would pass four gigabytes, or `bytes` is longer than one.
    #[inline]
    pub fn push_span(&mut self, bytes: &[u8]) -> Span {
        Span {
            at: self.push(bytes),
            len: u32::try_from(bytes.len()).expect("no one value is 4 GiB"),
        }
    }

    /// Append `bytes` behind their own length and say where the length went.
    ///
    /// For the caller that has nowhere else to keep a length. The offset alone
    /// is the whole reference, which is four bytes rather than the eight a
    /// [`Span`] costs, against one byte in the blob for anything under two
    /// hundred and fifty five and five for anything longer.
    ///
    /// # Panics
    ///
    /// If the blob would pass four gigabytes, or `bytes` is longer than one.
    pub fn push_sized(&mut self, bytes: &[u8]) -> u32 {
        if bytes.len() < LONG {
            let head = [u8::try_from(bytes.len()).expect("under LONG")];
            let at = self.push(&head);
            self.push(bytes);
            return at;
        }
        let len = u32::try_from(bytes.len()).expect("no one value is 4 GiB");
        let mut head = [0u8; LONG_PREFIX];
        head[0] = u8::try_from(LONG).expect("LONG is one byte");
        head[1..].copy_from_slice(&len.to_le_bytes());
        let at = self.push(&head);
        self.push(bytes);
        at
    }

    /// The bytes a [`Blob::push_sized`] offset points at.
    ///
    /// # Panics
    ///
    /// If `at` is not the start of a run that was pushed with its length.
    #[inline]
    #[must_use]
    pub fn sized(&self, at: u32) -> &[u8] {
        let at = at as usize;
        let (skip, len) = sized_head(&self.bytes, at);
        &self.bytes[at + skip..at + skip + len]
    }

    /// How long a [`Blob::push_sized`] run is, without reading it.
    ///
    /// This is what `HSTRLEN` asks. It used to be free, because the length was
    /// in the reference, and now it is one byte off the front of the value. That
    /// byte is on the same cache line as the value itself, so the answer costs
    /// the miss the caller would have taken to read the value anyway.
    #[inline]
    #[must_use]
    pub fn sized_len(&self, at: u32) -> usize {
        sized_head(&self.bytes, at as usize).1
    }

    /// Say that a [`Blob::push_sized`] run is not pointed at any more.
    #[inline]
    pub fn release_sized(&mut self, at: u32) {
        let (skip, len) = sized_head(&self.bytes, at as usize);
        self.release(skip + len);
    }

    /// The `len` bytes at `at`.
    ///
    /// # Panics
    ///
    /// If they are not inside the blob, which means a reference was kept across
    /// a [`Blob::compact`] without being moved.
    #[inline]
    #[must_use]
    pub fn read(&self, at: u32, len: usize) -> &[u8] {
        let at = at as usize;
        &self.bytes[at..at + len]
    }

    /// The bytes a [`Span`] points at.
    #[inline]
    #[must_use]
    pub fn span(&self, span: Span) -> &[u8] {
        self.read(span.at, span.len as usize)
    }

    /// Say that `len` bytes are not pointed at any more.
    ///
    /// This frees nothing. It moves the number that decides when
    /// [`Blob::compact`] is worth running.
    #[inline]
    pub const fn release(&mut self, len: usize) {
        self.dead += len;
    }

    /// Say that a [`Span`] is not pointed at any more.
    #[inline]
    pub const fn release_span(&mut self, span: Span) {
        self.release(span.len as usize);
    }

    /// Throw everything away and keep the allocation.
    #[inline]
    pub fn clear(&mut self) {
        self.bytes.clear();
        self.dead = 0;
    }

    /// Whether the dead bytes are worth a rebuild.
    #[inline]
    #[must_use]
    pub const fn worth_compacting(&self) -> bool {
        self.dead >= FLOOR && self.dead * 2 >= self.bytes.len()
    }

    /// Rebuild, keeping only what `keep` points at.
    ///
    /// The owner walks its own references and calls [`Keep::moved`] on each,
    /// which copies those bytes into the new blob and rewrites the offset in
    /// place. Anything not offered is gone. A reference the owner forgets to
    /// offer becomes a reference into a blob that moved underneath it, and
    /// [`Blob::read`] turns that into a panic rather than into wrong bytes.
    ///
    /// The order the owner walks in becomes the order in the new blob, so
    /// walking in row order leaves a sequential read sequential.
    pub fn compact<F>(&mut self, keep: F)
    where
        F: FnOnce(&mut Keep<'_>),
    {
        let fresh = {
            let mut k = Keep {
                old: &self.bytes,
                fresh: Vec::with_capacity(self.bytes.len() - self.dead),
            };
            keep(&mut k);
            k.fresh
        };
        self.bytes = fresh;
        self.dead = 0;
    }
}

/// A rebuild in progress, handed to the owner so it can move its references.
#[derive(Debug)]
pub struct Keep<'a> {
    old: &'a [u8],
    fresh: Vec<u8>,
}

impl Keep<'_> {
    /// Carry the `len` bytes at `*at` over, and point `at` at where they landed.
    ///
    /// # Panics
    ///
    /// If they are not inside the old blob, which is the mistake
    /// [`Blob::read`] catches and it is caught here for the same reason.
    #[inline]
    pub fn moved(&mut self, at: &mut u32, len: usize) {
        let from = *at as usize;
        let to = u32::try_from(self.fresh.len()).expect("the blob only shrinks here");
        self.fresh.extend_from_slice(&self.old[from..from + len]);
        *at = to;
    }

    /// The same for a [`Span`], whose length does not change.
    #[inline]
    pub fn moved_span(&mut self, span: &mut Span) {
        let len = span.len as usize;
        self.moved(&mut span.at, len);
    }

    /// The same for a [`Blob::push_sized`] run, prefix and all.
    ///
    /// The length is in the bytes being moved rather than in the reference, so
    /// it is read off the old copy, which is the whole reason [`Keep::peek`] is
    /// here.
    #[inline]
    pub fn moved_sized(&mut self, at: &mut u32) {
        let (skip, len) = sized_head(self.old, *at as usize);
        self.moved(at, skip + len);
    }

    /// The `len` bytes at `at`, as they were before the rebuild started.
    ///
    /// A reference whose length is written into the bytes rather than held
    /// beside them has to read those bytes to know how many to carry over, and
    /// the blob it would normally read from is half moved by the time it is
    /// asked. This is the old copy, which is still whole.
    ///
    /// # Panics
    ///
    /// If they are not inside the old blob.
    #[inline]
    #[must_use]
    pub fn peek(&self, at: u32, len: usize) -> &[u8] {
        let at = at as usize;
        &self.old[at..at + len]
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn what_goes_in_comes_back_out() {
        let mut b = Blob::new();
        let one = b.push_span(b"field");
        let two = b.push_span(b"");
        let three = b.push_span(b"a longer value than the first one");

        assert_eq!(b.span(one), b"field");
        assert_eq!(b.span(two), b"");
        assert_eq!(b.span(three), b"a longer value than the first one");
        assert_eq!(b.len(), 5 + 33);
        assert_eq!(b.dead(), 0);
    }

    #[test]
    fn a_length_written_in_front_reads_back_at_every_length() {
        let mut b = Blob::new();
        let lens = [0usize, 1, 2, 100, 253, 254, 255, 256, 257, 70_000];
        let at: Vec<u32> = lens
            .iter()
            .enumerate()
            .map(|(i, &n)| {
                let byte = u8::try_from(i).expect("ten of them");
                b.push_sized(&vec![byte; n])
            })
            .collect();

        for (i, (&n, &at)) in lens.iter().zip(&at).enumerate() {
            let byte = u8::try_from(i).expect("ten of them");
            assert_eq!(b.sized_len(at), n, "the length came back wrong");
            assert_eq!(b.sized(at), &vec![byte; n][..], "the bytes came back wrong");
        }

        // One byte of prefix under the marker and five at it and above.
        let short: usize = lens.iter().filter(|&&n| n < 255).map(|&n| n + 1).sum();
        let long: usize = lens.iter().filter(|&&n| n >= 255).map(|&n| n + 5).sum();
        assert_eq!(b.len(), short + long);
    }

    #[test]
    fn a_run_that_carries_its_own_length_moves_with_it() {
        let mut b = Blob::new();
        // Long and short mixed, so the rebuild has to read both prefix forms off
        // the copy it is reading from rather than the one it is writing.
        let mut live: Vec<u32> = Vec::new();
        for i in 0..100u32 {
            let n = if i % 3 == 0 { 300 } else { 40 };
            let byte = u8::try_from(i % 251).expect("under 251");
            let first = b.push_sized(&vec![byte; n]);
            b.release_sized(first);
            live.push(b.push_sized(&vec![byte; n]));
        }
        assert!(b.worth_compacting());
        let before = b.len();

        b.compact(|k| {
            for at in &mut live {
                k.moved_sized(at);
            }
        });

        assert_eq!(b.dead(), 0);
        assert_eq!(b.len() * 2, before, "the dead half went and no more");
        for (i, &at) in live.iter().enumerate() {
            let i = u32::try_from(i).expect("a hundred of them");
            let n = if i % 3 == 0 { 300 } else { 40 };
            let byte = u8::try_from(i % 251).expect("under 251");
            assert_eq!(b.sized(at), &vec![byte; n][..], "a reference moved wrongly");
        }
    }

    #[test]
    fn a_rewrite_leaves_the_old_bytes_behind_and_says_so() {
        let mut b = Blob::new();
        let old = b.push_span(b"before");
        b.release_span(old);
        let new = b.push_span(b"after");

        assert_eq!(b.span(new), b"after");
        assert_eq!(b.dead(), 6, "the old bytes are still there and counted");
        assert_eq!(b.len(), 11);
    }

    #[test]
    fn the_dead_bytes_come_back_and_the_live_ones_move() {
        let mut b = Blob::new();
        // Twenty kilobytes written, half of it abandoned, which is over the
        // floor and at the ratio.
        let mut live: Vec<Span> = Vec::new();
        for i in 0..100u32 {
            let bytes = vec![b'a' + u8::try_from(i % 26).expect("under 26"); 100];
            let first = b.push_span(&bytes);
            b.release_span(first);
            live.push(b.push_span(&bytes));
        }
        assert_eq!(b.dead(), 10_000);
        assert!(b.worth_compacting());

        let want: Vec<Vec<u8>> = live.iter().map(|&s| b.span(s).to_vec()).collect();
        b.compact(|k| {
            for span in &mut live {
                k.moved_span(span);
            }
        });

        assert_eq!(b.dead(), 0);
        assert_eq!(b.len(), 10_000, "only the live half survived");
        for (span, bytes) in live.iter().zip(&want) {
            assert_eq!(b.span(*span), &bytes[..], "a reference moved wrongly");
        }
    }

    #[test]
    fn a_small_or_mostly_live_blob_is_left_alone() {
        let mut b = Blob::new();
        b.push(&vec![0u8; 100_000]);
        b.release(3000);
        assert!(!b.worth_compacting(), "under the floor, whatever the ratio");

        let mut c = Blob::new();
        c.push(&vec![0u8; 100_000]);
        c.release(40_000);
        assert!(!c.worth_compacting(), "over the floor and under the half");
        c.release(10_000);
        assert!(c.worth_compacting(), "and at the half it is worth doing");
    }

    #[test]
    fn clearing_keeps_the_allocation_and_forgets_the_dead() {
        let mut b = Blob::with_capacity(1024);
        b.push(b"something");
        b.release(4);
        b.clear();

        assert!(b.is_empty());
        assert_eq!(b.dead(), 0);
        assert!(b.memory_bytes() >= 1024, "the allocation stayed");
    }
}