scc 3.7.0

use std::cmp::Ordering;
use std::fmt;
use std::mem::{forget, needs_drop};
use std::ops::Bound::{Excluded, Included, Unbounded};
use std::ops::{Deref, RangeBounds};
use std::ptr;
use std::sync::atomic::AtomicU32;
use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release};
#[cfg(not(feature = "loom"))]
use std::sync::atomic::{AtomicPtr, AtomicUsize};

#[cfg(feature = "loom")]
use loom::sync::atomic::{AtomicPtr, AtomicUsize};
use saa::Lock;

use crate::data_block::DataBlock;
use crate::utils::{cold_path, likely, snapshot};
use crate::{Comparable, Guard};

/// [`Leaf`] is an ordered array of key-value pairs.
///
/// A constructed key-value pair entry is never dropped until the entire [`Leaf`] instance is
/// dropped.
pub struct Leaf<K, V> {
    /// [`Array`] containing the key-value pairs.
    array: Array<K, V>,
    /// Pointer to the previous [`Leaf`].
    pub(super) prev: AtomicPtr<Leaf<K, V>>,
    /// Pointer to the next [`Leaf`].
    pub(super) next: AtomicPtr<Leaf<K, V>>,
    /// Lock to protect the linked list.
    pub(super) lock: Lock,
}

/// A fixed size ordered array of key-value pairs.
pub struct Array<K, V> {
    /// Metadata for entry and array states.
    ///
    /// The state of each entry is as follows.
    /// * `0`: `unreachable`.
    /// * `1 - array_size`: `rank`.
    ///
    /// The entry state transitions as follows.
    /// * `unreachable -> rank -> unreachable`.
    metadata: AtomicUsize,
    /// [`DataBlock`] containing the key-value pairs.
    data_block: DataBlock<K, V, { DIMENSION.len as usize }>,
    /// Bitmap representing the occupancy state of each entry in the data block.
    bitmap: AtomicU32,
}

/// The number of entries and number of state bits per entry.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct Dimension {
    /// Length of an [`Array`].
    pub len: u8,
    /// Number of bits required per entry metadata.
    pub bits: u8,
}

/// Insertion result.
pub enum InsertResult<K, V> {
    /// Insertion succeeded.
    Success,
    /// Duplicate key found.
    Duplicate(K, V),
    /// No vacant slot for the key.
    Full(K, V),
}

/// Remove result.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RemoveResult {
    /// Remove succeeded.
    Success,
    /// Remove succeeded and cleanup required.
    Retired,
    /// Remove failed.
    Fail,
    /// The [`Leaf`] is frozen.
    Frozen,
}

/// Array entry iterator.
#[derive(Debug)]
pub struct ArrayIter {
    /// Snapshot of the metadata of [`Array`].
    metadata: usize,
    /// Rank to position mapping.
    pos: [u8; DIMENSION.len as usize + 1],
}

/// Array entry iterator, reversed.
#[derive(Debug)]
pub struct ArrayRevIter {
    /// Snapshot of the metadata of [`Array`].
    metadata: usize,
    /// Rank to position mapping.
    pos: [u8; DIMENSION.len as usize + 1],
}

/// Leaf entry iterator.
pub struct Iter<'l, K, V> {
    /// Reference to the [`Leaf`] being iterated.
    leaf: &'l Leaf<K, V>,
    /// Array entry iterator.
    array_iter: ArrayIter,
}

/// Leaf entry iterator, reversed.
pub struct RevIter<'l, K, V> {
    /// Reference to the [`Leaf`] being iterated.
    leaf: &'l Leaf<K, V>,
    /// Array entry reverse iterator.
    array_rev_iter: ArrayRevIter,
}

/// Emulates `RangeBounds::contains`.
#[inline]
pub(crate) fn range_contains<K, Q, R: RangeBounds<Q>>(range: &R, key: &K) -> bool
where
    Q: Comparable<K> + ?Sized,
{
    (match range.start_bound() {
        Included(start) => start.compare(key).is_le(),
        Excluded(start) => start.compare(key).is_lt(),
        Unbounded => true,
    }) && (match range.end_bound() {
        Included(end) => end.compare(key).is_ge(),
        Excluded(end) => end.compare(key).is_gt(),
        Unbounded => true,
    })
}

impl<K, V> Leaf<K, V> {
    /// Creates a new empty [`Leaf`].
    #[inline]
    #[cfg(not(feature = "loom"))]
    pub(super) const fn new() -> Leaf<K, V> {
        Leaf {
            array: Array::new(),
            prev: AtomicPtr::new(ptr::null_mut()),
            next: AtomicPtr::new(ptr::null_mut()),
            lock: Lock::new(),
        }
    }

    /// Creates a new empty [`Leaf`].
    #[inline]
    #[cfg(feature = "loom")]
    pub(super) fn new() -> Leaf<K, V> {
        Leaf {
            array: Array::new(),
            prev: AtomicPtr::new(ptr::null_mut()),
            next: AtomicPtr::new(ptr::null_mut()),
            lock: Lock::new(),
        }
    }

    /// Creates a new empty [`Leaf`] in a frozen state.
    #[inline]
    #[cfg(not(feature = "loom"))]
    pub(super) const fn new_frozen() -> Leaf<K, V> {
        Leaf {
            array: Array::new_frozen(),
            prev: AtomicPtr::new(ptr::null_mut()),
            next: AtomicPtr::new(ptr::null_mut()),
            lock: Lock::new(),
        }
    }

    /// Creates a new empty [`Leaf`].
    #[inline]
    #[cfg(feature = "loom")]
    pub(super) fn new_frozen() -> Leaf<K, V> {
        Leaf {
            array: Array::new_frozen(),
            prev: AtomicPtr::new(ptr::null_mut()),
            next: AtomicPtr::new(ptr::null_mut()),
            lock: Lock::new(),
        }
    }

    /// Updates the linked list state around this entry.
    #[inline]
    pub(super) fn update_link<F>(&self, f: F)
    where
        F: FnOnce(
            Option<&AtomicPtr<Self>>,
            Option<&AtomicPtr<Self>>,
            *const Leaf<K, V>,
            *const Leaf<K, V>,
        ),
    {
        let mut prev_ptr = self.prev.load(Acquire);
        loop {
            if let Some(prev) = unsafe { prev_ptr.as_ref() } {
                prev.lock.lock_sync();
            }
            self.lock.lock_sync();
            let prev_check = self.prev.load(Acquire);
            if prev_check == prev_ptr {
                break;
            }
            if let Some(prev) = unsafe { prev_ptr.as_ref() } {
                prev.lock.release_lock();
            }
            self.lock.release_lock();
            prev_ptr = prev_check;
        }
        let prev = unsafe { prev_ptr.as_ref() };
        let next_ptr = self.next.load(Acquire);
        let next = unsafe { next_ptr.as_ref() };
        if let Some(next_link) = next {
            next_link.lock.lock_sync();
        }

        // Check consistency before modifying the linked list, because this leaf may have been
        // deleted by `remove_range` operations and `prev` and `next` may have connected to other
        // leaves.
        if prev.is_none_or(|p| ptr::eq(p.next.load(Relaxed), self))
            && next.is_none_or(|n| ptr::eq(n.prev.load(Relaxed), self))
        {
            f(
                prev.map(|p| &p.next),
                next.map(|n| &n.prev),
                prev_ptr,
                next_ptr,
            );
        }

        if let Some(prev_link) = prev {
            let released = prev_link.lock.release_lock();
            debug_assert!(released);
        }
        let released = self.lock.release_lock();
        debug_assert!(released);
        if let Some(next_link) = next {
            let released = next_link.lock.release_lock();
            debug_assert!(released);
        }
    }

    /// Deletes itself from the linked list.
    #[inline]
    pub(super) fn unlink(&self) {
        self.update_link(
            #[inline]
            |prev_next, next_prev, prev_ptr, next_ptr| {
                // `self`, on the other hand, keeps its pointers.
                if let Some(prev_next) = prev_next {
                    prev_next.store(next_ptr.cast_mut(), Release);
                }
                if let Some(next_prev) = next_prev {
                    next_prev.store(prev_ptr.cast_mut(), Release);
                }
            },
        );
    }

    /// Splices two leaves into the linked list.
    ///
    /// All the leaves between `left` and `right` are assumed to be `cleared` and bound to be
    /// unreachable.
    #[inline]
    pub(super) fn splice_link(
        left: Option<&Leaf<K, V>>,
        right: Option<&Leaf<K, V>>,
        _guard: &Guard,
    ) {
        let locked = left.is_none_or(|o| o.lock.lock_sync());
        debug_assert!(locked);
        let locked = right.is_none_or(|o| o.lock.lock_sync());
        debug_assert!(locked);

        if let Some(left) = left {
            let next = right.map_or(ptr::null(), ptr::from_ref).cast_mut();
            left.next.store(next, Release);
        }
        if let Some(right) = right {
            let prev = left.map_or(ptr::null(), ptr::from_ref).cast_mut();
            right.prev.store(prev, Release);
        }

        let released = left.is_none_or(|o| o.lock.release_lock());
        debug_assert!(released);
        let released = right.is_none_or(|o| o.lock.release_lock());
        debug_assert!(released);
    }
}

impl<K, V> fmt::Debug for Leaf<K, V>
where
    K: Clone + fmt::Debug + Ord,
    V: fmt::Debug,
{
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("Leaf { ")?;
        let ptr: *const Self = ptr::addr_of!(*self);
        write!(f, "addr: {ptr:?}")?;
        write!(f, ", array: {:?}", &self.array)?;
        write!(f, ", prev: {:?}", self.prev.load(Relaxed))?;
        write!(f, ", next: {:?}", self.next.load(Relaxed))?;
        f.write_str(" }")
    }
}

impl<K, V> Deref for Leaf<K, V> {
    type Target = Array<K, V>;

    #[inline]
    fn deref(&self) -> &Self::Target {
        &self.array
    }
}

impl<K, V> Array<K, V> {
    /// Creates a new [`Array`].
    #[cfg(not(feature = "loom"))]
    #[inline]
    pub(super) const fn new() -> Array<K, V> {
        Array {
            metadata: AtomicUsize::new(0),
            data_block: DataBlock::new(),
            bitmap: AtomicU32::new(0),
        }
    }

    #[cfg(feature = "loom")]
    #[inline]
    pub(super) fn new() -> Array<K, V> {
        Array {
            metadata: AtomicUsize::new(0),
            data_block: DataBlock::new(),
            bitmap: AtomicU32::new(0),
        }
    }

    /// Creates a new [`Array`] in a frozen state.
    #[cfg(not(feature = "loom"))]
    #[inline]
    pub(super) const fn new_frozen() -> Array<K, V> {
        Array {
            metadata: AtomicUsize::new(Dimension::FROZEN),
            data_block: DataBlock::new(),
            bitmap: AtomicU32::new(0),
        }
    }

    #[cfg(feature = "loom")]
    #[inline]
    pub(super) fn new_frozen() -> Array<K, V> {
        Array {
            metadata: AtomicUsize::new(Dimension::FROZEN),
            data_block: DataBlock::new(),
            bitmap: AtomicU32::new(0),
        }
    }

    /// Returns `true` if the [`Array`] has no reachable entry.
    #[inline]
    pub(super) fn is_empty(&self) -> bool {
        self.metadata.load(Relaxed) & (!Dimension::state_mask()) == 0
    }

    /// Returns `true` if the [`Array`] is full or retired.
    ///
    /// This may return `true` even when there is a slot if there was an insertion failure.
    #[inline]
    pub(super) fn is_full(&self) -> bool {
        self.bitmap.load(Relaxed).trailing_ones() == u32::from(DIMENSION.len)
            || Dimension::is_retired(self.metadata.load(Relaxed))
    }

    /// Returns `true` if the [`Array`] has retired.
    #[inline]
    pub(super) fn is_retired(&self) -> bool {
        Dimension::is_retired(self.metadata.load(Acquire))
    }

    /// Returns the current metadata.
    #[inline]
    pub(super) fn metadata(&self) -> usize {
        self.metadata.load(Acquire)
    }

    /// Returns a reference to the key at the given position.
    #[inline]
    pub(super) const fn key(&self, pos: u8) -> &K {
        unsafe { &*self.data_block.key_ptr(pos as usize) }
    }

    /// Returns a reference to the key at the given position.
    #[inline]
    pub(super) const fn val(&self, pos: u8) -> &V {
        unsafe { &*self.data_block.val_ptr(pos as usize) }
    }

    /// Inserts a key-value pair at the specified position without checking the metadata when the
    /// leaf is frozen.
    ///
    /// `rank` is calculated as `bitmap_pos + 1`.
    #[inline]
    pub(super) fn insert_unchecked(&self, key: K, val: V, bitmap_pos: u8) -> u8 {
        debug_assert!(bitmap_pos < DIMENSION.len);

        let pos = DIMENSION.slot_pos(bitmap_pos);
        debug_assert!(pos < DIMENSION.len);

        self.data_block.write(pos as usize, key, val);
        self.bitmap
            .store(self.bitmap.load(Relaxed) | (1 << bitmap_pos), Relaxed);

        let metadata = self.metadata.load(Relaxed);
        debug_assert!(Dimension::is_frozen(metadata));

        self.metadata.store(
            metadata | ((bitmap_pos as usize + 1) << (pos * DIMENSION.bits)),
            Release,
        );
        pos
    }

    /// Removes the entry at the specified position without checking the metadata.
    #[inline]
    pub(super) fn remove_unchecked(&self, mut metadata: usize, pos: u8) -> RemoveResult {
        loop {
            let mut new_metadata = metadata & !DIMENSION.rank_mask(pos);
            if new_metadata == 0 {
                new_metadata |= Dimension::RETIRED;
            }
            match self
                .metadata
                .compare_exchange(metadata, new_metadata, AcqRel, Acquire)
            {
                Ok(_) => {
                    if new_metadata == Dimension::RETIRED {
                        return RemoveResult::Retired;
                    }
                    return RemoveResult::Success;
                }
                Err(current_metadata) => {
                    if current_metadata & DIMENSION.rank_mask(pos) == 0 {
                        return RemoveResult::Fail;
                    } else if Dimension::is_frozen(current_metadata) {
                        return RemoveResult::Frozen;
                    }
                    metadata = current_metadata;
                }
            }
        }
    }

    /// Compares the given metadata value with the current one.
    ///
    /// Metadata validation is needed to avoid possible data races between writers and readers. For
    /// instance, when a writer is splitting a node where a reader tries to read an entry in the
    /// affected key range.
    ///
    ///  - Writer: start to insert an intermediate low key leaf.
    ///  - Reader: read the metadata that does not contain the intermediate low key leaf.
    ///  - Writer: insert the intermediate low key leaf.
    ///  - Writer: replace the high key leaf pointer.
    ///  - Reader: read the new high key leaf pointer.
    ///  - Reader: *cannot read keys in the intermediate low key leaf*.
    ///
    /// This kind of data races can be resolved by the reader validating the metadata after reading
    /// the pointer.
    #[inline]
    pub(super) fn validate(&self, metadata: usize) -> bool {
        // `Relaxed` is sufficient as long as the caller has load-acquired its contents.
        self.metadata.load(Relaxed) == metadata
    }

    /// Freezes the [`Array`] temporarily.
    #[inline]
    pub(super) fn freeze<const RELAXED: bool>(&self) -> usize {
        let metadata = if RELAXED {
            let metadata = self.metadata.load(Acquire);
            self.metadata.store(metadata | Dimension::FROZEN, Release);
            metadata
        } else {
            self.metadata.fetch_or(Dimension::FROZEN, AcqRel)
        };
        debug_assert!(!Dimension::is_frozen(metadata));
        metadata
    }

    /// Unfreezes the [`Array`].
    #[inline]
    pub(super) fn unfreeze(&self) {
        let metadata = self.metadata.load(Relaxed);
        debug_assert!(Dimension::is_frozen(metadata));
        self.metadata.store(Dimension::unfreeze(metadata), Release);
    }

    /// Distributes entries to other arrays after freezing itself.
    ///
    /// `dist` is a function to distribute entries to other containers where the first argument is
    /// the key, the second argument is the value, the third argument is the boundary, and the
    /// fourth argument is `rank - 1`.
    #[inline]
    pub(super) fn distribute<
        P: FnOnce(u8, usize, bool) -> bool,
        F: FnMut(&K, &V, u8, u8),
        const RELAXED: bool,
    >(
        &self,
        prepare: P,
        mut dist: F,
    ) -> bool {
        let metadata = self.freeze::<RELAXED>();
        let bitmap = self.bitmap.load(Relaxed);
        let (boundary, len, ordered) = Self::optimal_boundary(metadata, bitmap);
        if !prepare(boundary, len, ordered) {
            // Do nothing if the preparation fails.
            return false;
        }

        for (i, pos) in ArrayIter::with_metadata(metadata).enumerate() {
            #[allow(clippy::cast_possible_truncation)]
            dist(self.key(pos), self.val(pos), boundary, i as u8);
        }
        true
    }

    /// Iterates over reachable positions.
    #[inline]
    pub(crate) fn for_each_pos<P: FnOnce(bool) -> bool, F: FnMut(u8)>(&self, p: P, mut f: F) {
        let metadata = self.metadata.load(Acquire);
        if p(Dimension::is_frozen(metadata)) {
            let mut mutable_metadata = metadata & !Dimension::state_mask();
            for pos in 0..DIMENSION.len {
                if mutable_metadata == 0 {
                    break;
                }
                let rank = DIMENSION.rank_first(mutable_metadata);
                if rank != 0 {
                    f(pos);
                }
                mutable_metadata >>= DIMENSION.bits;
            }
        }
    }

    /// Iterates over initialized entries, including unreachable ones.
    #[inline]
    pub(crate) fn for_each_all<
        E,
        P: FnOnce(bool, bool) -> bool,
        F: FnMut(u8, u8, Option<(&K, &V)>, bool) -> Result<(), E>,
    >(
        &self,
        p: P,
        mut f: F,
    ) -> Result<(), E> {
        let metadata = self.metadata.load(Acquire);
        if p(
            Dimension::is_frozen(metadata),
            Dimension::is_retired(metadata),
        ) {
            let mut bitmap = self.bitmap.load(Relaxed);
            let mut bitmap_pos = 0;
            while bitmap != 0 {
                let pos = DIMENSION.slot_pos(bitmap_pos);
                let rank = DIMENSION.rank_at(metadata, pos);
                if (bitmap & 1) == 0 {
                    f(pos, rank, None, false)?;
                } else {
                    let entry = (self.key(pos), self.val(pos));
                    if rank == 0 {
                        f(pos, rank, Some(entry), true)?;
                    } else {
                        f(pos, rank, Some(entry), false)?;
                    }
                }
                bitmap_pos += 1;
                bitmap >>= 1;
            }
        }
        Ok(())
    }

    /// Returns the recommended number of entries that the left-side array should store when an
    /// [`Array`] is split, and the number of valid entries in the [`Array`].
    ///
    /// Returns `true` as the third result value when all the entries are ordered.
    ///
    /// Returns a number in `[1, len)` that represents the recommended number of entries in
    /// the left-side node. The number is calculated as follows for each adjacent slot:
    /// - Initial `score = 1`.
    /// - Rank increased: `score += 2`.
    /// - Take `score.div_ceil(2)` if `score < len` or `score / 2` if `score >= len`.
    ///
    /// For instance, when the length of an [`Array`] is 7,
    /// - Returns 6 for `rank = [1, 2, 3, 4, 5, 6, 7]`.
    /// - Returns 5 for `rank = [2, 1, 3, 4, 5, 6, 7]`.
    /// - Returns 4 for `rank = [6, 2, 4, 7, 1, 3, 5]`.
    /// - Returns 3 for `rank = [6, 2, 4, 1, 7, 3, 5]`.
    /// - Returns 3 for `rank = [6, 2, 4, 1, 7, 5, 3]`.
    /// - Returns 2 for `rank = [7, 6, 5, 4, 3, 1, 2]`.
    /// - Returns 1 for `rank = [7, 6, 5, 4, 3, 2, 1]`.
    #[inline]
    const fn optimal_boundary(metadata: usize, mut bitmap: u32) -> (u8, usize, bool) {
        let mut boundary = 1;
        let mut prev_rank = 0;
        let mut len = 0;
        let mut bitmap_pos = 0;
        while bitmap != 0 {
            if (bitmap & 1) == 1 {
                let pos = DIMENSION.slot_pos(bitmap_pos);
                let rank = DIMENSION.rank_at(metadata, pos);
                if rank != 0 {
                    len += 1;
                    if prev_rank != 0 && prev_rank < rank {
                        boundary += 2;
                    }
                    prev_rank = rank;
                }
            }
            bitmap_pos += 1;
            bitmap >>= 1;
        }

        if len > (3 * DIMENSION.len) / 4 {
            // Recommend the caller to split the leaf if the load factor is higher than `3/4`.
            let ordered = boundary == DIMENSION.len * 2 - 1;
            if boundary < len {
                boundary = boundary.div_ceil(2);
            } else {
                boundary /= 2;
            }
            (boundary, len as usize, ordered)
        } else {
            // Expect the caller would reorganize the leaf.
            (DIMENSION.len, len as usize, false)
        }
    }

    /// Builds a rank to position map from metadata.
    #[inline]
    const fn build_index(mut mutable_metadata: usize) -> [u8; DIMENSION.len as usize + 1] {
        let mut index = [u8::MAX; DIMENSION.len as usize + 1];
        *at_mut(&mut index, 0) = 0;
        let mut pos = 0;
        mutable_metadata &= !Dimension::state_mask();
        while mutable_metadata != 0 {
            let rank = DIMENSION.rank_first(mutable_metadata);
            if rank != 0 {
                *at_mut(&mut index, rank as usize) = pos;
            }
            pos += 1;
            mutable_metadata >>= DIMENSION.bits;
        }
        index
    }
}

impl<K, V> Array<K, V>
where
    K: Ord,
{
    /// Inserts a key-value pair.
    #[inline]
    pub(super) fn insert<const UPSERT: bool>(&self, key: K, val: V) -> InsertResult<K, V> {
        let mut bitmap_pos = DIMENSION.len;
        if UPSERT {
            // `Acquire` is needed to synchronize with previous failed insertion attempts.
            #[allow(clippy::cast_possible_truncation)]
            if self
                .bitmap
                .fetch_update(AcqRel, Acquire, |bitmap| {
                    bitmap_pos = bitmap.trailing_ones() as u8;
                    (bitmap_pos != DIMENSION.len).then_some(bitmap | (1 << bitmap_pos))
                })
                .is_err()
            {
                return InsertResult::Full(key, val);
            }

            // Write the key and value into the data block.
            self.data_block.write(
                DIMENSION.slot_pos(bitmap_pos) as usize,
                snapshot(&key),
                snapshot(&val),
            );
        }

        let mut metadata = self.metadata.load(Acquire);
        loop {
            if (metadata & Dimension::state_mask()) != 0 {
                cold_path();
                if bitmap_pos != DIMENSION.len {
                    self.bitmap.fetch_and(!(1 << bitmap_pos), Release);
                }
                return InsertResult::Full(key, val);
            }

            let mut min_max_rank = u8::MAX;
            let mut max_min_rank = 0;
            let mut new_metadata = metadata;
            let mut mutable_metadata = metadata;
            for pos in 0..DIMENSION.len {
                if mutable_metadata == 0 {
                    break;
                }
                let rank = DIMENSION.rank_first(mutable_metadata);
                if rank < min_max_rank && rank > max_min_rank {
                    match key.compare(self.key(pos)) {
                        Ordering::Less => {
                            min_max_rank = min_max_rank.min(rank);
                            new_metadata += 1 << (pos * DIMENSION.bits);
                        }
                        Ordering::Equal => {
                            if UPSERT {
                                max_min_rank = rank - 1;
                                new_metadata = metadata & (!(DIMENSION.rank_mask(pos)));
                                break;
                            }
                            if bitmap_pos != DIMENSION.len {
                                self.bitmap.fetch_and(!(1 << bitmap_pos), Release);
                            }
                            return InsertResult::Duplicate(key, val);
                        }
                        Ordering::Greater => max_min_rank = max_min_rank.max(rank),
                    }
                } else if rank > min_max_rank {
                    new_metadata += 1 << (pos * DIMENSION.bits);
                }
                mutable_metadata >>= DIMENSION.bits;
            }

            if !UPSERT && bitmap_pos == DIMENSION.len {
                // `Acquire` is needed to synchronize with previous failed insertion attempts.
                #[allow(clippy::cast_possible_truncation)]
                if self
                    .bitmap
                    .fetch_update(AcqRel, Acquire, |bitmap| {
                        bitmap_pos = bitmap.trailing_ones() as u8;
                        (bitmap_pos != DIMENSION.len).then_some(bitmap | (1 << bitmap_pos))
                    })
                    .is_err()
                {
                    return InsertResult::Full(key, val);
                }

                // Write the key and value into the data block.
                self.data_block.write(
                    DIMENSION.slot_pos(bitmap_pos) as usize,
                    snapshot(&key),
                    snapshot(&val),
                );
            }

            new_metadata |=
                (max_min_rank as usize + 1) << (DIMENSION.slot_pos(bitmap_pos) * DIMENSION.bits);
            if let Err(current_metadata) =
                self.metadata
                    .compare_exchange(metadata, new_metadata, AcqRel, Acquire)
            {
                metadata = current_metadata;
                continue;
            }

            // The key-value pair was moved to the array.
            forget((key, val));
            return InsertResult::Success;
        }
    }

    /// Removes the key if the condition is met.
    #[inline]
    pub(super) fn remove_if<Q, F: FnMut(&V) -> bool>(
        &self,
        key: &Q,
        condition: &mut F,
    ) -> RemoveResult
    where
        Q: Comparable<K> + ?Sized,
    {
        let metadata = self.metadata.load(Acquire);
        if Dimension::is_frozen(metadata) {
            cold_path();
            return RemoveResult::Frozen;
        }
        if let Some(pos) = self.search_slot(key, metadata) {
            if condition(self.val(pos)) {
                return self.remove_unchecked(metadata, pos);
            }
        }
        RemoveResult::Fail
    }

    /// Removes a range of entries.
    ///
    /// Returns the number of remaining entries.
    #[inline]
    pub(super) fn remove_range<Q, R: RangeBounds<Q>>(&self, range: &R)
    where
        Q: Comparable<K> + ?Sized,
    {
        let mut mutable_metadata = self.metadata.load(Acquire) & (!Dimension::state_mask());
        for pos in 0..DIMENSION.len {
            if mutable_metadata == 0 {
                break;
            }
            let rank = DIMENSION.rank_first(mutable_metadata);
            if rank != 0 {
                let k = self.key(pos);
                if range_contains(range, k) {
                    self.remove_unchecked(self.metadata.load(Acquire), pos);
                }
            }
            mutable_metadata >>= DIMENSION.bits;
        }
    }

    /// Returns an entry containing the specified key.
    #[inline]
    pub(super) fn search_entry<Q>(&self, key: &Q) -> Option<(&K, &V)>
    where
        Q: Comparable<K> + ?Sized,
    {
        self.search_slot(key, self.metadata.load(Acquire))
            .map(|i| (self.key(i), self.val(i)))
    }

    /// Returns the value associated with the specified key.
    #[inline]
    pub(super) fn search_val<Q>(&self, key: &Q) -> Option<&V>
    where
        Q: Comparable<K> + ?Sized,
    {
        self.search_slot(key, self.metadata.load(Acquire))
            .map(|i| self.val(i))
    }

    /// Returns the minimum entry among those that are not `Ordering::Less` than the given key.
    ///
    /// It additionally returns the current version of its metadata so the caller can validate the
    /// correctness of the result.
    #[allow(clippy::inline_always)]
    #[inline(always)]
    pub(super) fn min_greater_equal<Q>(&self, key: &Q) -> (Option<&V>, usize)
    where
        Q: Comparable<K> + ?Sized,
    {
        let metadata = self.metadata.load(Acquire);
        let mut min_max_rank = u8::MAX;
        let mut max_min_rank = 0;
        let mut min_max_pos = DIMENSION.len;
        let mut mutable_metadata = metadata;
        for pos in 0..DIMENSION.len {
            if mutable_metadata == 0 {
                break;
            }
            let rank = DIMENSION.rank_first(mutable_metadata);
            if rank < min_max_rank && rank > max_min_rank {
                match key.compare(self.key(pos)) {
                    Ordering::Less => {
                        if min_max_rank > rank {
                            min_max_rank = rank;
                            min_max_pos = pos;
                        }
                    }
                    Ordering::Equal => return (Some(self.val(pos)), metadata),
                    Ordering::Greater => max_min_rank = max_min_rank.max(rank),
                }
            }
            mutable_metadata >>= DIMENSION.bits;
        }
        (
            (min_max_pos != DIMENSION.len).then(|| self.val(min_max_pos)),
            metadata,
        )
    }

    /// Searches for a slot in which the key is stored.
    #[allow(clippy::inline_always)]
    #[inline(always)]
    fn search_slot<Q>(&self, key: &Q, mut mutable_metadata: usize) -> Option<u8>
    where
        Q: Comparable<K> + ?Sized,
    {
        mutable_metadata &= !Dimension::state_mask();
        let mut min_max_rank = u8::MAX;
        let mut max_min_rank = 0;
        for pos in 0..DIMENSION.len {
            if mutable_metadata == 0 {
                break;
            }
            let rank = DIMENSION.rank_first(mutable_metadata);
            if rank < min_max_rank && rank > max_min_rank {
                match key.compare(self.key(pos)) {
                    Ordering::Less => min_max_rank = min_max_rank.min(rank),
                    Ordering::Equal => return Some(pos),
                    Ordering::Greater => max_min_rank = max_min_rank.max(rank),
                }
            }
            mutable_metadata >>= DIMENSION.bits;
        }
        None
    }
}

impl<K, V> fmt::Debug for Array<K, V>
where
    K: Clone + fmt::Debug + Ord,
    V: fmt::Debug,
{
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("Array { ")?;
        let mut state = (false, false);
        self.for_each_all(
            |frozen, retired| {
                state = (frozen, retired);
                true
            },
            |i, rank, entry, removed| {
                if let Some(entry) = entry {
                    write!(f, "{i}: ({rank}, removed: {removed}, {entry:?}), ")?;
                } else {
                    write!(f, "{i}: (none), ")?;
                }
                Ok(())
            },
        )?;
        write!(f, "frozen: {}, ", state.0)?;
        write!(f, "retired: {}", state.1)?;
        f.write_str(" }")
    }
}

impl<K, V> Drop for Array<K, V> {
    #[inline]
    fn drop(&mut self) {
        if needs_drop::<(K, V)>() {
            let metadata = self.metadata.load(Acquire);
            let is_frozen = Dimension::is_frozen(metadata);
            let mut bitmap = self.bitmap.load(Relaxed);
            let mut bitmap_pos = 0;
            while bitmap != 0 {
                if (bitmap & 1) == 1 {
                    let pos = DIMENSION.slot_pos(bitmap_pos);
                    let rank = DIMENSION.rank_at(metadata, pos);
                    if !is_frozen || rank == 0 {
                        // `self` being frozen means that reachable values have copied to another
                        // array, and they should not be dropped here.
                        self.data_block.drop_in_place(pos as usize);
                    }
                }
                bitmap_pos += 1;
                bitmap >>= 1;
            }
        }
    }
}

impl Dimension {
    /// Returns the maximum key slot position when the leaf is full, entries are ordered, and no
    /// entries have been removed.
    #[inline]
    pub(super) const fn ideal_max_key_pos(self) -> u8 {
        self.len - 1
    }

    /// Flag indicating that the [`Array`] is frozen.
    const FROZEN: usize = 1_usize << (usize::BITS - 1);

    /// Flag indicating that the [`Array`] is retired.
    const RETIRED: usize = 1_usize << (usize::BITS - 2);

    /// Returns a bit mask for an array state.
    #[inline]
    const fn state_mask() -> usize {
        Self::RETIRED | Self::FROZEN
    }

    /// Returns `true` if the [`Array`] is frozen.
    #[inline]
    const fn is_frozen(metadata: usize) -> bool {
        metadata & Self::FROZEN != 0
    }

    /// Updates the metadata to represent a non-frozen state.
    #[inline]
    const fn unfreeze(metadata: usize) -> usize {
        metadata & (!Self::FROZEN)
    }

    /// Returns `true` if the [`Leaf`] is retired.
    #[inline]
    const fn is_retired(metadata: usize) -> bool {
        metadata & Self::RETIRED != 0
    }

    /// Returns a bit mask for an entry.
    #[inline]
    const fn rank_mask(self, pos: u8) -> usize {
        ((1_usize << self.bits) - 1) << (pos * self.bits)
    }

    /// Returns the rank of the first entry.
    #[allow(clippy::cast_possible_truncation)]
    #[inline]
    const fn rank_first(self, metadata: usize) -> u8 {
        (metadata % (1_usize << self.bits)) as u8
    }

    /// Returns the rank of the entry at the position.
    #[inline]
    const fn rank_at(self, metadata: usize, pos: u8) -> u8 {
        self.rank_first(metadata >> (pos * self.bits))
    }

    /// Returns the slot position corresponding to the bitmap position.
    #[inline]
    const fn slot_pos(self, bitmap_pos: u8) -> u8 {
        let unit_len = self.bits.next_power_of_two();
        if bitmap_pos < self.len - self.len % unit_len {
            return (bitmap_pos - bitmap_pos % unit_len) + (bitmap_pos + 1) % unit_len;
        }
        bitmap_pos
    }
}

/// The maximum number of entries and the number of metadata bits per entry in a [`Leaf`].
///
/// * `M`: The maximum number of entries.
/// * `B`: The minimum number of bits to express the state of an entry.
/// * `1`: The number of special states of an entry: unreachable.
/// * `2`: The number of special states of an [`Array`]: frozen, and retired.
/// * `U`: `usize::BITS`.
/// * `Eq1 = M + 1 <= 2^B`: `B` bits represent at least `M + 1` states.
/// * `Eq2 = B * M + 2 <= U`: `M entries + 2` special state.
/// * `Eq3 = Ceil(Log2(M + 2)) * M + 2 <= U`: derived from `Eq1` and `Eq2`.
///
/// Therefore, when `U = 64 => M = 15 / B = 4`, and `U = 32 => M = 7 / B = 3`.
pub const DIMENSION: Dimension = match usize::BITS / 8 {
    1 => Dimension { len: 2, bits: 2 },
    2 => Dimension { len: 4, bits: 3 },
    // `bits = 3` is sufficient, but `4` is more efficient.
    4 => Dimension { len: 7, bits: 4 },
    8 => Dimension { len: 15, bits: 4 },
    _ => Dimension { len: 25, bits: 5 },
};

impl ArrayIter {
    /// Creates a new [`ArrayIter`].
    #[inline]
    pub(super) fn new<K, V>(array: &Array<K, V>) -> ArrayIter {
        let metadata = array.metadata.load(Acquire);
        Self::with_metadata(metadata)
    }

    /// Clones the iterator.
    #[inline]
    pub(super) const fn clone(&self) -> ArrayIter {
        ArrayIter {
            metadata: self.metadata,
            pos: self.pos,
        }
    }

    /// Rewinds the iterator to the beginning.
    #[inline]
    pub(super) const fn rewind(&mut self) {
        *at_mut(&mut self.pos, 0) = 0;
    }

    /// Converts itself into a [`ArrayRevIter`].
    #[inline]
    pub(super) const fn rev(self) -> ArrayRevIter {
        ArrayRevIter {
            metadata: self.metadata,
            pos: self.pos,
        }
    }

    /// Returns the snapshot of leaf metadata that the [`Iter`] took.
    #[inline]
    pub(super) const fn metadata(&self) -> usize {
        self.metadata
    }

    /// Creates a new [`ArrayIter`] with the supplied metadata.
    #[inline]
    const fn with_metadata(metadata: usize) -> ArrayIter {
        let pos = Array::<(), ()>::build_index(metadata);
        ArrayIter { metadata, pos }
    }
}

impl Iterator for ArrayIter {
    type Item = u8;

    #[inline]
    fn next(&mut self) -> Option<Self::Item> {
        let mut rank = *at(&self.pos, 0) + 1;
        while likely(rank != DIMENSION.len + 1) {
            let pos = *at(&self.pos, rank as usize);
            if pos != u8::MAX {
                *at_mut(&mut self.pos, 0) = rank;
                return Some(pos);
            }
            rank += 1;
        }
        None
    }
}

impl<'l, K, V> Iter<'l, K, V> {
    /// Rewinds the iterator to the beginning.
    #[inline]
    pub(crate) const fn rewind(&mut self) {
        self.array_iter.rewind();
    }

    /// Creates a new [`Iter`].
    #[inline]
    pub(super) fn new(leaf: &'l Leaf<K, V>) -> Iter<'l, K, V> {
        Self {
            leaf,
            array_iter: ArrayIter::new(&leaf.array),
        }
    }

    /// Clones the iterator.
    #[inline]
    pub(super) const fn clone(&self) -> Iter<'l, K, V> {
        Iter {
            leaf: self.leaf,
            array_iter: self.array_iter.clone(),
        }
    }

    /// Converts itself into a [`RevIter`].
    #[inline]
    pub(super) const fn rev(self) -> RevIter<'l, K, V> {
        RevIter {
            leaf: self.leaf,
            array_rev_iter: self.array_iter.rev(),
        }
    }

    /// Returns a reference to the entry that the iterator is currently pointing to.
    #[inline]
    pub(super) const fn get(&self) -> Option<(&'l K, &'l V)> {
        let rank = *at(&self.array_iter.pos, 0);
        if likely(rank != 0) {
            let pos = *at(&self.array_iter.pos, rank as usize);
            return Some((self.leaf.array.key(pos), self.leaf.array.val(pos)));
        }
        None
    }

    /// Returns a reference to the max key.
    #[inline]
    pub(super) fn max_key(&self) -> Option<&'l K> {
        let mut rank = DIMENSION.len;
        while likely(rank != 0) {
            let pos = *at(&self.array_iter.pos, rank as usize);
            if pos != u8::MAX {
                return Some(self.leaf.key(pos));
            }
            rank -= 1;
        }
        None
    }

    /// Jumps to the min entry of the next non-empty leaf.
    #[inline]
    pub(super) fn jump(&mut self, _guard: &'l Guard) -> Option<(&'l K, &'l V)>
    where
        K: Ord,
    {
        let max_key = self.get().map(|(k, _)| k);
        let mut found_unlinked = false;
        while let Some(leaf) = unsafe { self.leaf.next.load(Acquire).as_ref() } {
            let metadata = leaf.metadata.load(Acquire);
            found_unlinked |= !ptr::eq(leaf.prev.load(Relaxed), self.leaf);

            self.leaf = leaf;
            self.array_iter = ArrayIter::with_metadata(metadata);

            // Data race resolution:
            //  - T1:                remove(L1) -> range(L0) ->              traverse(L1)
            //  - T2: unlink(L0) ->                             delete(L0)
            //  - T3:                                                        insertSmall(L1)
            //
            // T1 must not see T3's insertion while it still needs to observe its own deletion.
            // Therefore, keys that are smaller than the max key in the current leaf should be
            // filtered out here.
            for (k, v) in self.by_ref() {
                if likely(!found_unlinked) || max_key.is_none_or(|max| max < k) {
                    return Some((k, v));
                }
            }
        }
        None
    }
}

impl<K, V> fmt::Debug for Iter<'_, K, V> {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Iter")
            .field("leaf", &ptr::addr_of!(*self.leaf))
            .field("prev", &self.leaf.prev.load(Relaxed))
            .field("next", &self.leaf.next.load(Relaxed))
            .field("array_iter", &self.array_iter)
            .finish()
    }
}

impl<'l, K, V> Iterator for Iter<'l, K, V> {
    type Item = (&'l K, &'l V);

    #[inline]
    fn next(&mut self) -> Option<Self::Item> {
        self.array_iter
            .next()
            .map(|i| (self.leaf.key(i), self.leaf.val(i)))
    }
}

impl ArrayRevIter {
    /// Creates a new [`ArrayRevIter`].
    #[inline]
    pub(super) fn new<K, V>(array: &Array<K, V>) -> ArrayRevIter {
        let metadata = array.metadata.load(Acquire);
        Self::with_metadata(metadata)
    }

    /// Clones the iterator.
    #[inline]
    pub(super) const fn clone(&self) -> ArrayRevIter {
        ArrayRevIter {
            metadata: self.metadata,
            pos: self.pos,
        }
    }

    /// Rewinds the iterator to the beginning.
    #[inline]
    pub(super) const fn rewind(&mut self) {
        *at_mut(&mut self.pos, 0) = 0;
    }

    /// Converts itself into a [`ArrayIter`].
    #[inline]
    pub(super) const fn rev(self) -> ArrayIter {
        ArrayIter {
            metadata: self.metadata,
            pos: self.pos,
        }
    }

    /// Returns the snapshot of leaf metadata that the [`ArrayRevIter`] took.
    #[inline]
    pub(super) const fn metadata(&self) -> usize {
        self.metadata
    }

    /// Creates a new [`ArrayRevIter`] with the supplied metadata.
    #[inline]
    const fn with_metadata(metadata: usize) -> ArrayRevIter {
        let pos = Array::<(), ()>::build_index(metadata);
        ArrayRevIter { metadata, pos }
    }
}

impl Iterator for ArrayRevIter {
    type Item = u8;

    #[inline]
    fn next(&mut self) -> Option<Self::Item> {
        let mut rank = *at(&self.pos, 0);
        if likely(rank != 0) {
            rank -= 1;
        } else {
            rank = DIMENSION.len;
        }
        while likely(rank != 0) {
            let pos = *at(&self.pos, rank as usize);
            if pos != u8::MAX {
                *at_mut(&mut self.pos, 0) = rank;
                return Some(pos);
            }
            rank -= 1;
        }
        None
    }
}

impl<'l, K, V> RevIter<'l, K, V> {
    /// Rewinds the iterator to the beginning.
    #[inline]
    pub(crate) const fn rewind(&mut self) {
        self.array_rev_iter.rewind();
    }

    /// Creates a new [`RevIter`].
    #[inline]
    pub(super) fn new(leaf: &'l Leaf<K, V>) -> RevIter<'l, K, V> {
        Self {
            leaf,
            array_rev_iter: ArrayRevIter::new(&leaf.array),
        }
    }

    /// Clones the iterator.
    #[inline]
    pub(super) const fn clone(&self) -> RevIter<'l, K, V> {
        RevIter {
            leaf: self.leaf,
            array_rev_iter: self.array_rev_iter.clone(),
        }
    }

    /// Converts itself into an [`Iter`].
    #[inline]
    pub(super) const fn rev(self) -> Iter<'l, K, V> {
        Iter {
            leaf: self.leaf,
            array_iter: self.array_rev_iter.rev(),
        }
    }

    /// Returns a reference to the entry that the iterator is currently pointing to.
    #[inline]
    pub(super) const fn get(&self) -> Option<(&'l K, &'l V)> {
        let rank = *at(&self.array_rev_iter.pos, 0);
        if likely(rank != 0) {
            let pos = *at(&self.array_rev_iter.pos, rank as usize);
            return Some((self.leaf.array.key(pos), self.leaf.array.val(pos)));
        }
        None
    }

    /// Returns a reference to the min key entry.
    #[inline]
    pub(super) fn min_key(&self) -> Option<&'l K> {
        let mut rank = 1;
        while likely(rank != DIMENSION.len + 1) {
            let pos = *at(&self.array_rev_iter.pos, rank as usize);
            if pos != u8::MAX {
                return Some(self.leaf.key(pos));
            }
            rank += 1;
        }
        None
    }

    /// Jumps to the max entry of the prev non-empty leaf.
    #[inline]
    pub(super) fn jump(&mut self, _guard: &'l Guard) -> Option<(&'l K, &'l V)>
    where
        K: Ord,
    {
        let min_key = self.get().map(|(k, _)| k);
        let mut found_unlinked = false;
        while let Some(leaf) = unsafe { self.leaf.prev.load(Acquire).as_ref() } {
            let metadata = leaf.metadata.load(Acquire);
            found_unlinked |= !ptr::eq(leaf.next.load(Relaxed), self.leaf);

            self.leaf = leaf;
            self.array_rev_iter = ArrayRevIter::with_metadata(metadata);

            // See `Iter::jump`.
            for (k, v) in self.by_ref() {
                if likely(!found_unlinked) || min_key.is_none_or(|min| min > k) {
                    return Some((k, v));
                }
            }
        }
        None
    }
}

impl<K, V> fmt::Debug for RevIter<'_, K, V> {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("RevIter")
            .field("leaf", &ptr::addr_of!(*self.leaf))
            .field("prev", &self.leaf.prev.load(Relaxed))
            .field("next", &self.leaf.next.load(Relaxed))
            .field("array_rev_iter", &self.array_rev_iter)
            .finish()
    }
}

impl<'l, K, V> Iterator for RevIter<'l, K, V> {
    type Item = (&'l K, &'l V);

    #[inline]
    fn next(&mut self) -> Option<Self::Item> {
        self.array_rev_iter
            .next()
            .map(|i| (self.leaf.key(i), self.leaf.val(i)))
    }
}

/// Gets a reference to an entry in an array.
#[inline]
const fn at<T>(array: &[T], index: usize) -> &T {
    unsafe { &*array.as_ptr().add(index) }
}

/// Gets a mutable reference to an entry in an array.
#[inline]
const fn at_mut<T>(array: &mut [T], index: usize) -> &mut T {
    unsafe { &mut *array.as_mut_ptr().add(index) }
}

#[cfg(not(feature = "loom"))]
#[cfg(test)]
mod test {
    use std::sync::Arc;
    use std::sync::atomic::AtomicBool;

    use proptest::prelude::*;
    use tokio::sync::Barrier;

    use super::*;

    #[test]
    fn array() {
        let array: Array<String, String> = Array::new();
        assert!(matches!(
            array.insert::<false>("MY GOODNESS!".to_owned(), "OH MY GOD!!".to_owned()),
            InsertResult::Success
        ));
        assert!(matches!(
            array.insert::<false>("GOOD DAY".to_owned(), "OH MY GOD!!".to_owned()),
            InsertResult::Success
        ));
        assert_eq!(array.search_entry("MY GOODNESS!").unwrap().1, "OH MY GOD!!");
        assert_eq!(array.search_entry("GOOD DAY").unwrap().1, "OH MY GOD!!");

        for pos in 0..DIMENSION.len {
            if let InsertResult::Full(k, v) =
                array.insert::<false>(pos.to_string(), pos.to_string())
            {
                assert_eq!(pos + 2, DIMENSION.len);
                assert_eq!(k, pos.to_string());
                assert_eq!(v, pos.to_string());
                break;
            }
            assert_eq!(
                array.search_entry(&pos.to_string()).unwrap(),
                (&pos.to_string(), &pos.to_string())
            );
        }

        for pos in 0..DIMENSION.len {
            let result = array.remove_if(&pos.to_string(), &mut |_| pos >= 10);
            if pos >= 10 && pos + 2 < DIMENSION.len {
                assert_eq!(result, RemoveResult::Success);
            } else {
                assert_eq!(result, RemoveResult::Fail);
            }
        }

        assert_eq!(
            array.remove_if("GOOD DAY", &mut |v| v == "OH MY"),
            RemoveResult::Fail
        );
        assert_eq!(
            array.remove_if("GOOD DAY", &mut |v| v == "OH MY GOD!!"),
            RemoveResult::Success
        );
        assert!(array.search_entry("GOOD DAY").is_none());
        assert_eq!(
            array.remove_if("MY GOODNESS!", &mut |_| true),
            RemoveResult::Success
        );
        assert!(array.search_entry("MY GOODNESS!").is_none());
        assert!(array.search_entry("1").is_some());
        assert!(matches!(
            array.insert::<false>("1".to_owned(), "1".to_owned()),
            InsertResult::Duplicate(..)
        ));
        assert!(matches!(
            array.insert::<false>("100".to_owned(), "100".to_owned()),
            InsertResult::Full(..)
        ));

        let mut iter = ArrayIter::new(&array);
        for pos in 0..DIMENSION.len {
            if let Some(e) = iter.next() {
                assert_eq!(array.key(e), &pos.to_string());
                assert_eq!(array.val(e), &pos.to_string());
                assert_ne!(
                    array.remove_if(&pos.to_string(), &mut |_| true),
                    RemoveResult::Fail
                );
            } else {
                break;
            }
        }

        assert!(matches!(
            array.insert::<false>("200".to_owned(), "200".to_owned()),
            InsertResult::Full(..)
        ));
    }

    #[test]
    fn iter_rev_iter() {
        let leaf: Leaf<usize, usize> = Leaf::new();
        for pos in 0..DIMENSION.len as usize {
            if pos % 2 == 0 {
                assert!(matches!(
                    leaf.insert::<false>(pos * 1024 + 1, pos),
                    InsertResult::Success
                ));
            } else {
                assert!(matches!(
                    leaf.insert::<false>(pos * 2, pos),
                    InsertResult::Success
                ));
            }
        }
        assert!(matches!(
            leaf.remove_if(&6, &mut |_| true),
            RemoveResult::Success
        ));

        let mut iter = Iter::new(&leaf);
        assert_eq!(iter.next(), Some((&1, &0)));
        let rev_iter = iter.rev();
        assert_eq!(rev_iter.get(), Some((&1, &0)));
        iter = rev_iter.rev();
        assert_eq!(iter.get(), Some((&1, &0)));

        let mut prev_key = 0;
        let mut sum = 0;
        for (key, _) in Iter::new(&leaf) {
            assert_ne!(*key, 6);
            assert!(prev_key < *key);
            prev_key = *key;
            sum += *key;
        }
        prev_key = usize::MAX;

        for (key, _) in RevIter::new(&leaf) {
            assert_ne!(*key, 6);
            assert!(prev_key > *key);
            prev_key = *key;
            sum -= *key;
        }
        assert_eq!(sum, 0);
    }

    #[test]
    fn calculate_boundary() {
        let leaf: Leaf<usize, usize> = Leaf::new();
        for i in 0..DIMENSION.len as usize {
            assert!(matches!(leaf.insert::<false>(i, i), InsertResult::Success));
        }
        assert_eq!(
            Array::<usize, usize>::optimal_boundary(
                leaf.metadata.load(Relaxed),
                leaf.bitmap.load(Relaxed)
            ),
            (DIMENSION.len - 1, DIMENSION.len as usize, true)
        );

        let leaf: Leaf<usize, usize> = Leaf::new();
        for i in (0..DIMENSION.len as usize).rev() {
            assert!(matches!(leaf.insert::<false>(i, i), InsertResult::Success));
        }
        assert_eq!(
            Array::<usize, usize>::optimal_boundary(
                leaf.metadata.load(Relaxed),
                leaf.bitmap.load(Relaxed)
            ),
            (1, DIMENSION.len as usize, false)
        );

        let leaf: Leaf<usize, usize> = Leaf::new();
        for i in 0..DIMENSION.len as usize {
            if i < DIMENSION.len as usize / 2 {
                assert!(matches!(
                    leaf.insert::<false>(usize::MAX - i, usize::MAX - i),
                    InsertResult::Success
                ));
            } else {
                assert!(matches!(leaf.insert::<false>(i, i), InsertResult::Success));
            }
        }
        if usize::BITS == 32 {
            // `7, 6, 5, 1, 2, 3, 4`: `3`.
            assert_eq!(
                Array::<usize, usize>::optimal_boundary(
                    leaf.metadata.load(Relaxed),
                    leaf.bitmap.load(Relaxed)
                ),
                (3, DIMENSION.len as usize, false)
            );
        } else {
            // `14, 13, 12, 11, 10, 9, 8, 1, 2, 3, 4, 5, 6, 7`: `7`.
            assert_eq!(
                Array::<usize, usize>::optimal_boundary(
                    leaf.metadata.load(Relaxed),
                    leaf.bitmap.load(Relaxed)
                ),
                (7, DIMENSION.len as usize, false)
            );
        }
    }

    #[test]
    fn optimal_boundary() {
        if size_of::<usize>() == 8 {
            let data = [
                (
                    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14],
                    14,
                    1,
                    true,
                ),
                (
                    [0, 1, 2, 3, 4, 5, 6, 7, 9, 8, 10, 11, 12, 13, 14],
                    13,
                    2,
                    false,
                ),
                (
                    [0, 1, 2, 3, 4, 5, 6, 7, 9, 8, 11, 10, 12, 13, 14],
                    12,
                    3,
                    false,
                ),
                (
                    [1, 0, 2, 3, 4, 6, 5, 7, 9, 8, 11, 10, 12, 14, 13],
                    9,
                    6,
                    false,
                ),
                (
                    [1, 0, 3, 2, 4, 6, 5, 7, 9, 8, 11, 10, 12, 14, 13],
                    8,
                    7,
                    false,
                ),
                (
                    [7, 6, 5, 4, 3, 2, 1, 0, 8, 9, 10, 11, 12, 13, 14],
                    7,
                    7,
                    false,
                ),
            ];
            for (array, expect, rev_expect, ordered) in data {
                for rev in [false, true] {
                    let leaf = Leaf::new();
                    if rev {
                        for k in array.iter().rev() {
                            leaf.insert::<true>(*k, *k);
                        }
                    } else {
                        for k in array {
                            leaf.insert::<true>(k, k);
                        }
                    }
                    let optimal_boundary = Array::<usize, usize>::optimal_boundary(
                        leaf.metadata.load(Relaxed),
                        leaf.bitmap.load(Relaxed),
                    );
                    assert_eq!(
                        optimal_boundary.0,
                        if rev { rev_expect } else { expect },
                        "{rev} {expect}"
                    );
                    assert_eq!(optimal_boundary.1, 15);
                    assert!((ordered && !rev) == optimal_boundary.2);
                }
            }
        }
    }

    #[test]
    fn special() {
        let leaf: Leaf<usize, usize> = Leaf::new();
        assert!(matches!(
            leaf.insert::<false>(11, 17),
            InsertResult::Success
        ));
        assert!(matches!(
            leaf.insert::<false>(19, 17),
            InsertResult::Success
        ));
        assert!(matches!(
            leaf.insert::<false>(32, 17),
            InsertResult::Success
        ));
        assert!(matches!(
            leaf.insert::<false>(20, 17),
            InsertResult::Success
        ));
        assert!(matches!(leaf.insert::<false>(1, 17), InsertResult::Success));
        assert!(matches!(leaf.insert::<false>(0, 17), InsertResult::Success));

        let leaf1 = Leaf::new_frozen();
        let leaf2 = Leaf::new_frozen();
        leaf.distribute::<_, _, true>(
            |_, _, _| true,
            |k, v, _, i| {
                if *k < 20 {
                    leaf1.insert_unchecked(*k, *v, i);
                } else {
                    leaf2.insert_unchecked(*k, *v, i - 4);
                }
            },
        );
        leaf1.unfreeze();
        leaf2.unfreeze();
        assert_eq!(leaf1.search_entry(&0), Some((&0, &17)));
        assert_eq!(leaf1.search_entry(&1), Some((&1, &17)));
        assert_eq!(leaf2.search_entry(&20), Some((&20, &17)));
        assert_eq!(leaf2.search_entry(&32), Some((&32, &17)));
        assert!(matches!(leaf.insert::<false>(1, 7), InsertResult::Full(..)));
        assert_eq!(leaf.remove_if(&17, &mut |_| true), RemoveResult::Frozen);
        assert!(matches!(leaf.insert::<false>(3, 5), InsertResult::Full(..)));

        leaf.unfreeze();
        assert!(matches!(leaf.insert::<false>(3, 7), InsertResult::Success));

        assert_eq!(leaf.remove_if(&1, &mut |_| true), RemoveResult::Success);

        let mut invalid = 0;
        let mut valid = 0;
        assert!(
            leaf.for_each_all(
                |frozen, retired| {
                    assert!(!frozen && !retired);
                    true
                },
                |_, _, entry, removed| -> Result<(), ()> {
                    if removed {
                        assert_eq!(*entry.unwrap().0, 1);
                        invalid += 1;
                    } else if entry.is_some() {
                        valid += 1;
                    }
                    Ok(())
                }
            )
            .is_ok()
        );
        assert_eq!(invalid, 1);
        assert_eq!(valid, 6);

        assert_eq!(leaf.remove_if(&0, &mut |_| true), RemoveResult::Success);
        assert_eq!(leaf.remove_if(&3, &mut |_| true), RemoveResult::Success);
        assert_eq!(leaf.remove_if(&11, &mut |_| true), RemoveResult::Success);
        assert_eq!(leaf.remove_if(&19, &mut |_| true), RemoveResult::Success);
        assert_eq!(leaf.remove_if(&20, &mut |_| true), RemoveResult::Success);
        assert_eq!(leaf.remove_if(&32, &mut |_| true), RemoveResult::Retired);

        assert!(matches!(leaf.insert::<false>(5, 3), InsertResult::Full(..)));
    }

    proptest! {
        #[cfg_attr(miri, ignore)]
        #[test]
        fn general(insert in 0_usize..DIMENSION.len as usize, remove in 0_usize..DIMENSION.len as usize) {
            let array: Array<usize, usize> = Array::new();
            assert!(array.is_empty());
            for i in 0..insert {
                assert!(matches!(array.insert::<false>(i, i), InsertResult::Success));
            }
            assert!(array.is_empty() == (insert == 0));
            for i in 0..insert {
                assert!(matches!(array.insert::<false>(i, i), InsertResult::Duplicate(..)));
                assert!(!array.is_empty());
                let result = array.min_greater_equal(&i);
                assert_eq!(result.0, Some(&i));
            }
            for i in 0..insert {
                assert_eq!(array.search_entry(&i).unwrap(), (&i, &i));
            }
            if insert == DIMENSION.len as usize {
                assert!(matches!(array.insert::<false>(usize::MAX, usize::MAX), InsertResult::Full(..)));
            }
            for i in 0..remove {
                if i < insert {
                    if i == insert - 1 {
                        assert!(matches!(array.remove_if(&i, &mut |_| true), RemoveResult::Retired));
                        for i in 0..insert {
                            assert!(matches!(array.insert::<false>(i, i), InsertResult::Full(..)));
                        }
                    } else {
                        assert!(matches!(array.remove_if(&i, &mut |_| true), RemoveResult::Success));
                    }
                } else {
                    assert!(matches!(array.remove_if(&i, &mut |_| true), RemoveResult::Fail));
                    assert!(array.is_empty());
                }
            }
        }

        #[cfg_attr(miri, ignore)]
        #[test]
        fn range(start in 0_usize..DIMENSION.len as usize, end in 0_usize..DIMENSION.len as usize) {
            let array: Array<usize, usize> = Array::new();
            for i in 1..DIMENSION.len as usize - 1 {
                prop_assert!(matches!(array.insert::<false>(i, i), InsertResult::Success));
            }
            array.remove_range(&(start..end));
            for i in 1..DIMENSION.len as usize - 1 {
                prop_assert!(array.search_entry(&i).is_none() == (start..end).contains(&i));
            }
            prop_assert!(array.search_entry(&0).is_none());
            prop_assert!(array.search_entry(&(DIMENSION.len as usize - 1)).is_none());
        }
    }

    #[cfg_attr(miri, ignore)]
    #[tokio::test(flavor = "multi_thread", worker_threads = 16)]
    async fn update() {
        let num_excess = 3;
        let num_tasks = DIMENSION.len as usize + num_excess;
        for _ in 0..256 {
            let barrier = Arc::new(Barrier::new(num_tasks));
            let leaf: Arc<Leaf<usize, usize>> = Arc::new(Leaf::new());
            let full: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0));
            let retire: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0));
            let mut task_handles = Vec::with_capacity(num_tasks);
            for t in 1..=num_tasks {
                let barrier_clone = barrier.clone();
                let leaf_clone = leaf.clone();
                let full_clone = full.clone();
                let retire_clone = retire.clone();
                task_handles.push(tokio::spawn(async move {
                    barrier_clone.wait().await;
                    let inserted = match leaf_clone.insert::<false>(t, t) {
                        InsertResult::Success => {
                            assert_eq!(leaf_clone.search_entry(&t).unwrap(), (&t, &t));
                            true
                        }
                        InsertResult::Duplicate(_, _) => unreachable!(),
                        InsertResult::Full(k, v) => {
                            assert_eq!(k, v);
                            assert_eq!(k, t);
                            full_clone.fetch_add(1, Relaxed);
                            false
                        }
                    };
                    {
                        let mut prev = 0;
                        let mut iter = Iter::new(&leaf_clone);
                        for e in iter.by_ref() {
                            assert_eq!(e.0, e.1);
                            assert!(*e.0 > prev);
                            prev = *e.0;
                        }
                    }

                    barrier_clone.wait().await;
                    assert_eq!((*full_clone).load(Relaxed), num_excess);
                    if inserted {
                        assert_eq!(leaf_clone.search_entry(&t).unwrap(), (&t, &t));
                    }
                    {
                        let iter = Iter::new(&leaf_clone);
                        assert_eq!(iter.count(), DIMENSION.len as usize);
                    }

                    barrier_clone.wait().await;
                    match leaf_clone.remove_if(&t, &mut |_| true) {
                        RemoveResult::Success => assert!(inserted),
                        RemoveResult::Fail => assert!(!inserted),
                        RemoveResult::Frozen => unreachable!(),
                        RemoveResult::Retired => {
                            assert!(inserted);
                            assert_eq!(retire_clone.swap(1, Relaxed), 0);
                        }
                    }
                }));
            }
            for r in futures::future::join_all(task_handles).await {
                assert!(r.is_ok());
            }
        }
    }

    #[cfg_attr(miri, ignore)]
    #[tokio::test(flavor = "multi_thread", worker_threads = 16)]
    async fn durability() {
        let num_tasks = 16_usize;
        let workload_size = 8_usize;
        for _ in 0..16 {
            for k in 0..=workload_size {
                let barrier = Arc::new(Barrier::new(num_tasks));
                let leaf: Arc<Leaf<usize, usize>> = Arc::new(Leaf::new());
                let inserted: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
                let mut task_handles = Vec::with_capacity(num_tasks);
                for _ in 0..num_tasks {
                    let barrier_clone = barrier.clone();
                    let leaf_clone = leaf.clone();
                    let inserted_clone = inserted.clone();
                    task_handles.push(tokio::spawn(async move {
                        {
                            barrier_clone.wait().await;
                            if let InsertResult::Success = leaf_clone.insert::<false>(k, k) {
                                assert!(!inserted_clone.swap(true, Relaxed));
                            }
                        }
                        {
                            barrier_clone.wait().await;
                            for i in 0..workload_size {
                                if i != k {
                                    let _result = leaf_clone.insert::<false>(i, i);
                                }
                                assert!(!leaf_clone.is_retired());
                                assert_eq!(leaf_clone.search_entry(&k).unwrap(), (&k, &k));
                            }
                            for i in 0..workload_size {
                                let _result = leaf_clone.remove_if(&i, &mut |v| *v != k);
                                assert_eq!(leaf_clone.search_entry(&k).unwrap(), (&k, &k));
                            }
                        }
                    }));
                }
                for r in futures::future::join_all(task_handles).await {
                    assert!(r.is_ok());
                }
                assert!((*inserted).load(Relaxed));
            }
        }
    }
}