armdb 0.12.0

//! Reusable reverse cursor state for read-only skip-list descent.
//!
//! [`SkipList::back_step`] and [`SkipList::back_seek_last`] only read tower
//! pointers: they never mutate a tower or help complete a CAS. For every
//! level, a saved entry is either null, the list head, or a predecessor whose
//! height includes that level. A saved non-head entry is used as a descent
//! start only while it is live and strictly before the probe; marked, equal,
//! and after-probe entries are rejected. The head is the special valid start
//! at every level.
//!
//! Unlinking does not invalidate a node's outgoing pointers while the same
//! caller-supplied [`seize::LocalGuard`] that observed the pointer remains
//! live, so this module may read a stale saved predecessor and its tower under
//! that guard. Reusing a finger with non-increasing probes gives the expected
//! amortized benefit; after a null result it is intended for terminal use.
//! Correctness does not depend on either performance precondition.

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

#[cfg(miri)]
use std::sync::{Mutex, OnceLock, mpsc};

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

#[cfg(miri)]
struct BackStepPause {
    arrived: mpsc::SyncSender<()>,
    resume: mpsc::Receiver<()>,
}

#[cfg(miri)]
static BACK_STEP_PAUSE: OnceLock<Mutex<Option<BackStepPause>>> = OnceLock::new();

/// Installs the Miri fixture pause after its reader checks a saved
/// predecessor's liveness. The hook pauses every eligible checkpoint until
/// cleared; this fixture expects one pause and clears it after the reader joins.
#[cfg(miri)]
pub fn test_pause_back_step_after_liveness(
    arrived: mpsc::SyncSender<()>,
    resume: mpsc::Receiver<()>,
) {
    *BACK_STEP_PAUSE
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap() = Some(BackStepPause { arrived, resume });
}

/// Removes the Miri-only `back_step` pause after its reader has joined.
#[cfg(miri)]
pub fn test_clear_back_step_pause() {
    *BACK_STEP_PAUSE
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap() = None;
}

#[cfg(miri)]
fn pause_back_step_after_liveness() {
    // This hook pauses every eligible checkpoint until the fixture clears it
    // after the reader joins. The current fixture expects one such pause.
    let pause = BACK_STEP_PAUSE
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap();
    if let Some(pause) = pause.as_ref() {
        pause.arrived.send(()).unwrap();
        pause.resume.recv().unwrap();
    }
}

pub(crate) struct BackFinger<N: SkipNode> {
    preds: [*mut N; MAX_HEIGHT],
}

impl<N: SkipNode> BackFinger<N> {
    pub(crate) const fn empty() -> Self {
        Self {
            preds: [std::ptr::null_mut(); MAX_HEIGHT],
        }
    }
}

impl<N: SkipNode> SkipList<N> {
    pub(crate) fn back_step(
        &self,
        finger: &mut BackFinger<N>,
        key: &[u8],
        guard: &seize::LocalGuard<'_>,
    ) -> *mut N {
        let _ = guard;
        let h = self.height.load(AtomicOrdering::Relaxed);
        let mut start = None;
        for level in 0..h {
            let saved = finger.preds[level];
            if saved.is_null() {
                continue;
            }
            if saved != self.head {
                // SAFETY: `saved` was captured while `guard` was live. The
                // same guard pins its allocation even if a concurrent writer
                // unlinks it; the module invariant also makes it tall enough
                // for this saved level.
                let saved_ref = unsafe { &*saved };
                if saved_ref.is_marked()
                    || self.key_cmp(saved_ref.key_bytes(), key) != Ordering::Less
                {
                    continue;
                }
                #[cfg(miri)]
                pause_back_step_after_liveness();
            }
            start = Some((saved, level));
            break;
        }
        let (mut current, start_level) = start.unwrap_or((self.head, h - 1));
        for level in (0..=start_level).rev() {
            loop {
                // SAFETY: `current` is head or a node reached while this
                // caller's `guard` is live. Its height includes `level`, and
                // the guard pins the allocation while its outgoing tower is
                // read, including after unlink.
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(AtomicOrdering::Acquire) });
                while !next.is_null() {
                    // SAFETY: the non-null `next` was loaded from a tower
                    // under this same guard. Its allocation remains pinned
                    // and its height includes `level`, so `is_marked` may
                    // read it and its outgoing pointer is readable.
                    if !unsafe { &*next }.is_marked() {
                        break;
                    }
                    // SAFETY: `next` was loaded from a tower under this same
                    // guard. Its allocation remains pinned and its height
                    // includes `level`, so this tower load is readable.
                    next =
                        strip_mark(unsafe { (*next).tower(level).load(AtomicOrdering::Acquire) });
                }
                if next.is_null() {
                    break;
                }
                // SAFETY: the non-null `next` was loaded from a tower under
                // this same guard. Its allocation remains pinned, so
                // `key_bytes` may read it even after unlink.
                if self.key_cmp(unsafe { &*next }.key_bytes(), key) != Ordering::Less {
                    break;
                }
                current = next;
            }
            finger.preds[level] = current;
        }
        if current == self.head {
            std::ptr::null_mut()
        } else {
            current
        }
    }

    pub(crate) fn back_seek_last(
        &self,
        finger: &mut BackFinger<N>,
        guard: &seize::LocalGuard<'_>,
    ) -> *mut N {
        let _ = guard;
        let mut current = self.head;
        let h = self.height.load(AtomicOrdering::Relaxed);
        for level in (0..h).rev() {
            loop {
                // SAFETY: `current` is head or a node reached while this
                // caller's `guard` is live. Its height includes `level`, and
                // the guard pins the allocation while its outgoing tower is
                // read, including after unlink.
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(AtomicOrdering::Acquire) });
                while !next.is_null() {
                    // SAFETY: the non-null `next` was loaded from a tower
                    // under this same guard. Its allocation remains pinned
                    // and its height includes `level`, so `is_marked` may
                    // read it and its outgoing pointer is readable.
                    if !unsafe { &*next }.is_marked() {
                        break;
                    }
                    // SAFETY: `next` was loaded from a tower under this same
                    // guard. Its allocation remains pinned and its height
                    // includes `level`, so this tower load is readable.
                    next =
                        strip_mark(unsafe { (*next).tower(level).load(AtomicOrdering::Acquire) });
                }
                if next.is_null() {
                    break;
                }
                current = next;
            }
            finger.preds[level] = current;
        }
        if current == self.head {
            std::ptr::null_mut()
        } else {
            current
        }
    }
}

#[cfg(test)]
mod tests {
    use std::cmp::Ordering;

    use crate::DiskLoc;
    use crate::skiplist::SkipList;
    use crate::skiplist::node::{ConstNode, SkipNode};

    use super::BackFinger;

    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]
    }

    fn key_u32(value: u32) -> [u8; 4] {
        value.to_be_bytes()
    }

    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
    }

    fn assert_fresh_steps_match(list: &TestList, probes: impl IntoIterator<Item = u8>) {
        let guard = list.collector().enter();
        for probe in probes {
            let mut finger = BackFinger::empty();
            let got = list.back_step(&mut finger, &key(probe), &guard);
            let oracle = list.find_last_lt(&key(probe), &guard);
            assert_eq!(
                got as *const TestNode, oracle as *const TestNode,
                "probe={probe}"
            );
        }
    }

    #[test]
    fn back_step_matches_find_last_lt_asc() {
        assert_fresh_steps_match(&build(false, &[(2, 1), (4, 2), (6, 2), (8, 1)]), 0..=10);
    }

    #[test]
    fn back_step_matches_find_last_lt_desc() {
        assert_fresh_steps_match(
            &build(true, &[(2, 1), (4, 2), (6, 2), (8, 1)]),
            (0..=10).rev(),
        );
    }

    #[test]
    fn back_seek_last_matches_find_last() {
        for list in [
            build(false, &[(2, 1), (4, 2), (6, 2), (8, 1)]),
            build(true, &[(2, 1), (4, 2), (6, 2), (8, 1)]),
            build(false, &[]),
        ] {
            let guard = list.collector().enter();
            let mut finger = BackFinger::empty();
            assert_eq!(
                list.back_seek_last(&mut finger, &guard) as *const TestNode,
                list.find_last(&guard) as *const TestNode,
            );
        }
    }

    #[test]
    fn back_finger_handles_empty_and_single_lists() {
        let empty = build(false, &[]);
        let guard = empty.collector().enter();
        let mut finger = BackFinger::empty();
        assert!(empty.back_seek_last(&mut finger, &guard).is_null());
        assert!(empty.back_step(&mut finger, &key(5), &guard).is_null());

        let single = build(false, &[(5, 1)]);
        let guard = single.collector().enter();
        let mut finger = BackFinger::empty();
        assert_eq!(
            unsafe { (*single.back_seek_last(&mut finger, &guard)).key },
            key(5)
        );
        assert!(single.back_step(&mut finger, &key(5), &guard).is_null());
    }

    #[test]
    fn back_step_rejects_a_marked_saved_predecessor() {
        let list = build(false, &[(2, 1), (4, 3), (6, 1), (8, 1)]);
        let guard = list.collector().enter();
        let mut finger = BackFinger::empty();
        assert_eq!(
            unsafe { (*list.back_seek_last(&mut finger, &guard)).key },
            key(8)
        );
        assert_eq!(
            unsafe { (*list.back_step(&mut finger, &key(8), &guard)).key },
            key(6)
        );
        assert!(list.remove(&key(4), &guard).is_some());
        let got = list.back_step(&mut finger, &key(6), &guard);
        let oracle = list.find_last_lt(&key(6), &guard);
        assert_eq!(got as *const TestNode, oracle as *const TestNode);
    }

    #[test]
    fn back_finger_sweep_matches_repeated_find_last_lt() {
        for list in [
            build(false, &[(2, 1), (4, 3), (6, 2), (8, 1)]),
            build(true, &[(2, 1), (4, 3), (6, 2), (8, 1)]),
        ] {
            let guard = list.collector().enter();
            let mut finger = BackFinger::empty();
            let mut current = list.back_seek_last(&mut finger, &guard);
            while !current.is_null() {
                let key_bytes = unsafe { (*current).key_bytes() };
                let next = list.back_step(&mut finger, key_bytes, &guard);
                let oracle = list.find_last_lt(key_bytes, &guard);
                assert_eq!(next as *const TestNode, oracle as *const TestNode);
                if !next.is_null() {
                    assert_eq!(
                        list.key_cmp(unsafe { (*next).key_bytes() }, key_bytes),
                        Ordering::Less
                    );
                }
                current = next;
            }
        }
    }

    #[test]
    fn back_finger_stress_10k_deterministic_deletions_matches_oracle() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        for i in 0u32..10_000 {
            let height = 1 + ((i + 1).trailing_zeros() / 2).min(15) as u8;
            list.insert(n(key_u32(i), height), &guard);
        }

        let mut state = 0xD1CE_F1A6_5EED_0001u64;
        let mut finger = BackFinger::empty();
        let mut current = list.back_seek_last(&mut finger, &guard);
        while !current.is_null() {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            if state & 0x7 == 0 {
                let removed = key_u32((state as usize % 10_000) as u32);
                let _ = list.remove(&removed, &guard);
            }

            let current_key = unsafe { (*current).key_bytes() };
            let next = list.back_step(&mut finger, current_key, &guard);
            let oracle = list.find_last_lt(current_key, &guard);
            assert_eq!(next as *const TestNode, oracle as *const TestNode);
            if !next.is_null() {
                assert_eq!(
                    list.key_cmp(unsafe { (*next).key_bytes() }, current_key),
                    Ordering::Less
                );
            }
            current = next;
        }
    }
}