armdb 0.6.0

//! Resumable, read-only forward cursor over a `SkipList`.
//!
//! `finger_seek` returns the first node that is **not before** the probe key in
//! list order — the list-order analogue of `find_first_ge` — but resumes the
//! descent from the predecessors saved by the previous seek instead of always
//! starting at `head`. For a batch sorted in list order this turns `M·log N`
//! into roughly `M·log(N/M) + M`. Read-only: one seize guard for the whole walk,
//! no tower mutation, no helping-CAS (helping outside `write_lock` would race the
//! linker's plain store — see `presearch`).

use std::cmp::Ordering;
use std::sync::atomic::Ordering as AtomicOrdering;

use super::node::{MAX_HEIGHT, SkipNode};
use super::{SkipList, strip_mark};

/// Predecessor staircase carried between successive `finger_seek` calls.
pub(crate) struct Finger<N: SkipNode> {
    preds: [*mut N; MAX_HEIGHT],
}

impl<N: SkipNode> SkipList<N> {
    /// A fresh finger positioned before every node (all predecessors = head).
    pub(crate) fn finger(&self) -> Finger<N> {
        Finger {
            preds: [self.head; MAX_HEIGHT],
        }
    }

    /// Follow `current`'s level-`level` pointer, skipping logically-removed
    /// (marked) nodes. Read-only — mirrors the inner skip in `find_first_ge`.
    /// Marked nodes' outgoing pointers stay valid (unlink does not clear them),
    /// and the seize guard keeps them allocated.
    #[inline]
    fn next_unmarked(&self, current: *mut N, level: usize) -> *mut N {
        let mut next = strip_mark(unsafe { (*current).tower(level).load(AtomicOrdering::Acquire) });
        while !next.is_null() && unsafe { &*next }.is_marked() {
            next = strip_mark(unsafe { (*next).tower(level).load(AtomicOrdering::Acquire) });
        }
        next
    }

    /// Is `candidate` at or after `current` in list order? Decides whether
    /// adopting the saved predecessor moves the cursor forward (never backward).
    /// `head` is special-cased: it always sorts first regardless of `reversed`
    /// (its key is empty, which `key_cmp` would mis-order under `reversed`).
    #[inline]
    fn at_or_after(&self, current: *mut N, candidate: *mut N) -> bool {
        if candidate == current || current == self.head {
            return true;
        }
        if candidate == self.head {
            return false;
        }
        self.key_cmp(
            unsafe { &*candidate }.key_bytes(),
            unsafe { &*current }.key_bytes(),
        ) != Ordering::Less
    }

    /// First node not before `key` in list order, resuming from the finger.
    pub(crate) fn finger_seek<'g>(
        &self,
        finger: &mut Finger<N>,
        key: &[u8],
        guard: &'g seize::LocalGuard<'_>,
    ) -> Option<&'g N> {
        let _ = guard; // lifetime anchor: result valid while the guard is alive
        let h = self.height.load(AtomicOrdering::Relaxed);
        let mut current = self.head;
        for level in (0..h).rev() {
            // Adopt the saved predecessor at this level when it is a tighter
            // (>=) lower bound than `current`. The saved staircase is monotonic
            // and, for non-decreasing probe keys, a valid lower bound — so this
            // never overshoots the true predecessor and never moves backward.
            let saved = finger.preds[level];
            if self.at_or_after(current, saved) {
                current = saved;
            }
            // Walk forward while the next node is strictly before `key`.
            loop {
                let next = self.next_unmarked(current, level);
                if next.is_null() {
                    break;
                }
                if self.key_cmp(unsafe { &*next }.key_bytes(), key) == Ordering::Less {
                    current = next;
                } else {
                    break;
                }
            }
            finger.preds[level] = current;
        }
        let result = self.next_unmarked(current, 0);
        if result.is_null() {
            None
        } else {
            Some(unsafe { &*result })
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::DiskLoc;
    use crate::skiplist::SkipList;
    use crate::skiplist::node::ConstNode;

    type TestNode = ConstNode<[u8; 4], 8>;
    type TestList = SkipList<TestNode>;

    fn disk() -> DiskLoc {
        DiskLoc::new(0, 0, 0)
    }

    fn n(key: [u8; 4], h: u8) -> *mut TestNode {
        TestNode::alloc(key, [0u8; 8], disk(), h)
    }

    fn key(b: u8) -> [u8; 4] {
        [0, 0, 0, b]
    }

    /// Build a list with the given (key-byte, height) pairs.
    fn build(reversed: bool, items: &[(u8, u8)]) -> TestList {
        let list = TestList::new(reversed);
        {
            let guard = list.collector().enter();
            for &(b, h) in items {
                list.insert(n(key(b), h), &guard);
            }
        }
        list
    }

    /// Oracle: a fresh-from-head finger_seek must equal find_first_ge for every key.
    #[test]
    fn finger_matches_find_first_ge_asc() {
        let list = build(false, &[(2, 1), (4, 2), (6, 2), (8, 1)]);
        let guard = list.collector().enter();
        for probe in 0u8..=10 {
            let mut finger = list.finger();
            let got = list.finger_seek(&mut finger, &key(probe), &guard);
            let oracle = list.find_first_ge(&key(probe), &guard);
            let got_ptr = got.map_or(std::ptr::null(), |r| r as *const TestNode);
            assert_eq!(got_ptr, oracle as *const TestNode, "probe={probe}");
        }
    }

    /// A single forward sweep over a sorted batch must return the same nodes as
    /// per-key fresh seeks (proves the resume path doesn't skip or overshoot).
    #[test]
    fn finger_forward_sweep_matches_oracle_asc() {
        let list = build(false, &[(1, 1), (3, 2), (5, 1), (7, 3), (9, 1)]);
        let guard = list.collector().enter();
        let probes: Vec<u8> = vec![0, 1, 3, 4, 5, 9, 10]; // sorted ASC
        let mut finger = list.finger();
        for &p in &probes {
            let got = list.finger_seek(&mut finger, &key(p), &guard);
            let oracle = list.find_first_ge(&key(p), &guard);
            let got_ptr = got.map_or(std::ptr::null(), |r| r as *const TestNode);
            assert_eq!(got_ptr, oracle as *const TestNode, "p={p}");
        }
    }

    /// Duplicate consecutive probes return the same node and never move backward.
    #[test]
    fn finger_duplicates_and_no_backward() {
        let list = build(false, &[(2, 2), (4, 1), (6, 2)]);
        let guard = list.collector().enter();
        let mut finger = list.finger();
        let a = list.finger_seek(&mut finger, &key(4), &guard).unwrap().key;
        let b = list.finger_seek(&mut finger, &key(4), &guard).unwrap().key;
        assert_eq!(a, key(4));
        assert_eq!(b, key(4));
        // forward to 6, then probe 5 again would violate the contract; instead
        // probe 6 (>= 4) stays correct.
        let c = list.finger_seek(&mut finger, &key(6), &guard).unwrap().key;
        assert_eq!(c, key(6));
    }

    /// Reversed (DESC) list: list order is descending, so probes must be issued
    /// in descending byte order. finger_seek must match find_first_ge there too.
    #[test]
    fn finger_matches_find_first_ge_desc() {
        let list = build(true, &[(2, 1), (4, 2), (6, 2), (8, 1)]);
        let guard = list.collector().enter();
        // DESC list order: 8,6,4,2. Probe in descending byte order.
        for probe in (0u8..=10).rev() {
            let mut finger = list.finger();
            let got = list.finger_seek(&mut finger, &key(probe), &guard);
            let oracle = list.find_first_ge(&key(probe), &guard);
            let got_ptr = got.map_or(std::ptr::null(), |r| r as *const TestNode);
            assert_eq!(got_ptr, oracle as *const TestNode, "probe={probe}");
        }
    }

    /// Sparse vs dense: a one-element list and an empty list behave.
    #[test]
    fn finger_edge_empty_and_single() {
        let empty = build(false, &[]);
        let guard = empty.collector().enter();
        let mut f = empty.finger();
        assert!(empty.finger_seek(&mut f, &key(5), &guard).is_none());

        let single = build(false, &[(5, 1)]);
        let guard2 = single.collector().enter();
        let mut f2 = single.finger();
        assert_eq!(
            single.finger_seek(&mut f2, &key(1), &guard2).unwrap().key,
            key(5)
        );
        let mut f3 = single.finger();
        assert!(single.finger_seek(&mut f3, &key(9), &guard2).is_none());
    }

    /// Concurrent removal: a marked node is skipped, results still match a fresh
    /// scan after the removal is observed under the same guard.
    #[test]
    fn finger_skips_marked_nodes() {
        let list = build(false, &[(2, 2), (4, 1), (6, 2), (8, 1)]);
        let guard = list.collector().enter();
        list.remove(&key(4), &guard);
        let mut finger = list.finger();
        // probe 3 should now land on 6 (4 removed), matching find_first_ge.
        let got = list.finger_seek(&mut finger, &key(3), &guard);
        let oracle = list.find_first_ge(&key(3), &guard);
        let got_ptr = got.map_or(std::ptr::null(), |r| r as *const TestNode);
        assert_eq!(got_ptr, oracle as *const TestNode);
    }
}