tablestg 0.4.10

use crate::*;

// Bucket that can store small variable size records ( up to 255 bytes ).

/// For reading Bucket.
pub struct Reader<'a> {
    pub data: &'a [u8],
}

/// For writing Bucket.
pub struct Writer<'a> {
    pub data: &'a mut [u8],
}

/// Iteration position.
pub struct Pos {
    slot: usize,
    entry: u8,
}

impl Pos {
    pub fn start() -> Self {
        Self { slot: 0, entry: 0 }
    }
}

/// Number of slots.
const SLOTS: usize = 103; // Fairly arbitrary, a larger number (e.g. 253) reduces collisions, but takes up space.

// Page offsets,
// USED, FREE are single byte fields.
// SLOT is SLOTS bytes.
// This is followed by data area.
// This is followed by free space.
// THis is followed by entry array ( which grows backwards ).

/// Offset of USED byte, which stores the number of allocated entries.
const USED: usize = 0;
/// Offset of FREE byte, first free entry.
const FREE: usize = USED + 1;
/// Offser of DALL bytes (2 bytes)
const DALL: usize = FREE + 1;
/// Offset of slot array.
const SLOT: usize = DALL + 2;
/// Offset of data array.
const DATA: usize = SLOT + SLOTS;

/// Size of entry.
const ESIZE: usize = 3;

impl<'a> Reader<'a> {
    pub fn new(data: &'a [u8]) -> Self {
        Self { data }
    }

    /// Get offset and length of slice of data associated with key.
    pub fn get<K: DKey>(&self, key: &K, hash: u64, ps: &mut PageSet) -> Option<(usize, usize)> {
        if self.data.is_empty() {
            return None;
        }
        let slot = (hash % SLOTS as u64) as usize;
        let mut entry = self.data[SLOT + slot];
        while entry != 0 {
            let voff = self.get_voff(entry);
            let len = self.data[voff] as usize;
            let v = &self.data[voff + 1..voff + 1 + len];
            if key.ok(v, ps) {
                return Some((voff + 1, len));
            }
            entry = self.next(entry);
        }
        None
    }

    /// Get next (offset, length) pair, pos moves to next row.
    pub fn iter_next(&self, pos: &mut Pos) -> Option<(usize, usize)> {
        if self.data.is_empty() {
            return None;
        }
        while pos.entry == 0 {
            if pos.slot == SLOTS {
                return None;
            }
            pos.entry = self.data[SLOT + pos.slot];
            pos.slot += 1;
        }
        let voff = self.get_voff(pos.entry);
        pos.entry = self.next(pos.entry);
        let len = self.data[voff] as usize;
        Some((voff + 1, len))
    }

    /// Returns a new Vec of specified size with same records but compacted.
    /// Size must be sufficient to hold records.
    pub fn compact(&self, size: usize) -> Vec<u8> {
        let mut result = vec![0; size];
        let mut w = Writer::new(&mut result);
        let mut pos = Pos::start();
        while let Some((off, len)) = self.iter_next(&mut pos) {
            let ud = &self.data[off..off + len];
            w.insert_at(pos.slot - 1, ud);
        }
        // println!("compact used={} old size={} new size={}", self.data[USED], self.data.len(), result.len() );
        result
    }

    /// Entry offset
    fn eoff(&self, entry: u8) -> usize {
        assert!(entry > 0);
        self.data.len() - (entry as usize) * ESIZE
    }

    /// Next entry
    fn next(&self, entry: u8) -> u8 {
        self.data[self.eoff(entry)]
    }

    /// Get value offset from entry array.
    fn get_voff(&self, entry: u8) -> usize {
        let off = self.eoff(entry) + 1;
        DATA + self.data[off] as usize + (self.data[off + 1] as usize) * 256
    }
}

impl<'a> Writer<'a> {
    pub fn new(data: &'a mut [u8]) -> Self {
        Self { data }
    }

    /// Insert into Hash Map Page.
    ///
    /// Pre-condition : page must have sufficient space. No duplicates!
    pub fn insert(&mut self, user_data: &[u8], hash: u64) {
        debug_assert!(self.space(user_data.len()) == 0);
        let slot = (hash % SLOTS as u64) as usize;
        self.insert_at(slot, user_data);
    }

    /// Extra space needed add a new record (row) of specified size (without compacting)?
    pub fn space(&self, rsize: usize) -> usize {
        let used = DATA + self.get2(DALL) + (self.data[USED] as usize) * 3;
        let reqd = used + 1 + rsize + (if self.data[FREE] == 0 { ESIZE } else { 0 });
        if reqd > self.data.len() {
            reqd - self.data.len()
        } else {
            0
        }
    }

    fn insert_at(&mut self, slot: usize, user_data: &[u8]) {
        let n = user_data.len();
        let new_entry = self.alloc_entry();
        let voff = self.alloc_data(1 + n);

        self.data[voff] = n as u8;
        self.data[voff + 1..voff + 1 + n].copy_from_slice(user_data);

        self.set_voff(new_entry, voff);

        let old_entry = self.data[SLOT + slot];
        self.set_next(new_entry, old_entry);
        self.data[SLOT + slot] = new_entry;
    }

    /// Remove a key.
    pub fn remove<K: DKey>(&mut self, key: &K, hash: u64, ps: &mut PageSet) -> bool {
        let slot = (hash % SLOTS as u64) as usize;
        let mut entry = self.data[SLOT + slot];
        let mut prev = None;
        while entry != 0 {
            let voff = self.get_voff(entry);
            let len = self.data[voff] as usize;
            let v = &self.data[voff + 1..voff + 1 + len];

            let next = self.next(entry);
            if key.ok(v, ps) {
                // Remove entry.
                if let Some(prev) = prev {
                    self.set_next(prev, next);
                } else {
                    self.data[SLOT + slot] = next;
                }
                // Put entry in free chain.
                self.set_next(entry, self.data[FREE]);
                self.data[FREE] = entry;

                // ToDo : increment free data total
                return true;
            }
            prev = Some(entry);
            entry = next;
        }
        false
    }

    fn alloc_entry(&mut self) -> u8 {
        let result = self.data[FREE];
        if result != 0 {
            self.data[FREE] = self.next(result);
            return result;
        }
        let result = self.data[USED];
        self.data[USED] += 1;
        result + 1
    }

    fn eoff(&self, entry: u8) -> usize {
        assert!(entry > 0);
        self.data.len() - (entry as usize) * ESIZE
    }

    fn next(&self, entry: u8) -> u8 {
        self.data[self.eoff(entry)]
    }

    fn set_next(&mut self, entry: u8, val: u8) {
        self.data[self.eoff(entry)] = val;
    }

    /// Get value offset from entry array.
    fn get_voff(&self, entry: u8) -> usize {
        let off = self.eoff(entry) + 1;
        self.get2(off)
    }

    /// Set value offset from entry array.
    fn set_voff(&mut self, entry: u8, voff: usize) {
        let voff = voff - DATA;
        let off = self.eoff(entry) + 1;
        self.set2(off, voff);
    }

    fn alloc_data(&mut self, size: usize) -> usize {
        let result = self.get2(DALL);
        self.set2(DALL, result + size);
        DATA + result
    }

    fn get2(&self, off: usize) -> usize {
        self.data[off] as usize + (self.data[off + 1] as usize) * 256
    }

    fn set2(&mut self, off: usize, to: usize) {
        self.data[off] = (to % 256) as u8;
        self.data[off + 1] = ((to / 256) % 256) as u8;
    }
}