use crate::{
error::{PERes, PIRes},
index::{
config::IndexOrd,
keeper::{IndexKeeper, IndexLimits, IndexModify},
tree::{
key_changes::KeyChanges,
lock_group::{LockGroup, LockGroups},
nodes::{Leaf, Node, PageIter, PageIterBack, TreeNode, TreeNodeRef, Value},
parent_change::{ChildChanged, ParentChange},
},
},
};
use std::{cmp::Ordering, fmt::Display, ops::Bound};
use self::path::{BottomDepth, Path, TopDepth};
use super::config::IndexTypeInternal;
mod lock_group;
pub mod nodes;
mod parent_change;
mod path;
#[cfg(test)]
mod nodes_tests;
#[cfg(test)]
mod tests;
pub(crate) mod key_changes;
pub(crate) struct ChangeResults {
add: Option<usize>,
remove: Option<usize>,
}
pub trait Index<K, V> {
fn get(&mut self, k: &K) -> PERes<Option<Value<V>>>;
fn iter_from(&mut self, first: Bound<&K>) -> PERes<PageIter<K, V>>;
fn back_iter_from(&mut self, first: Bound<&K>) -> PERes<PageIterBack<K, V>>;
}
pub(crate) trait IndexApply<K, V>: Index<K, V> {
fn apply(&mut self, adds: &[KeyChanges<K, V>]) -> PIRes<()>;
}
impl<K: Display> Display for PosRef<K> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}:{}=>{}", self.k, self.pos, self.node_ref)
}
}
#[derive(PartialEq, Clone, Debug, Eq)]
pub struct PosRef<K> {
pub k: K,
pub pos: usize,
pub node_ref: TreeNodeRef,
}
impl<K: Clone> PosRef<K> {
pub(crate) fn new(k: &K, pos: usize, node_ref: TreeNodeRef) -> PosRef<K> {
PosRef {
k: k.clone(),
pos,
node_ref,
}
}
}
fn split_and_save<K, V, I>(index: &mut I, id: TreeNodeRef) -> PIRes<Vec<(K, TreeNodeRef)>>
where
K: IndexOrd + Clone,
I: IndexModify<K, V>,
{
let top_limit = index.limits().top();
let new_nodes = index.update_with(&id, |node| (true, node.split(top_limit)))?;
let mut ids = Vec::new();
for (k, new_node) in new_nodes {
let saved_id = index.insert(new_node)?;
ids.push((k, saved_id));
}
Ok(ids)
}
fn recompute_path<K, V, I>(index: &mut I, child: &ChildChanged<K>) -> PIRes<Path<K>>
where
K: IndexOrd + Clone,
I: IndexModify<K, V>,
{
let node = index.get_root_refresh()?.expect("Root is there");
let mut path = Path::new(child.search_key().clone());
let mut cur_node = PosRef::new(child.search_key(), 0, node);
loop {
let mut restart = true;
let mut end = false;
if let Some(new) = index
.load_from_node(&cur_node.node_ref, |node, version| {
path.push(cur_node.clone(), version, node.len());
if cur_node.node_ref == child.item().pos_ref().node_ref {
path.compute_depths_start(child.item().bottom_depth());
path.replace_search_key(cur_node.k.clone());
end = true;
return None;
}
if let TreeNode::Node(n) = node {
if let Some(x) = n.find_write(child.search_key()) {
restart = false;
Some(x)
} else {
None
}
} else {
None
}
})?
.flatten()
{
cur_node = new;
}
if end {
break Ok(path);
}
if restart {
let node = index.get_root_refresh()?.expect("Root is there");
cur_node = PosRef::new(child.search_key(), 0, node);
path.clear();
}
}
}
fn recompute_path_depth<K, V, I>(index: &mut I, depth: BottomDepth, k: &K) -> PIRes<Path<K>>
where
K: IndexOrd + Clone,
I: IndexModify<K, V>,
{
let node = index.get_root_refresh()?.expect("Root is there");
let mut path = Path::new(k.clone());
let mut cur_node = PosRef::new(k, 0, node);
loop {
let mut restart = true;
let mut end = false;
if let Some(new) = index
.load_from_node(&cur_node.node_ref, |node, version| {
path.push(cur_node.clone(), version, node.len());
match node {
TreeNode::Node(n) => {
if let Some(x) = n.find_write(k) {
restart = false;
Some(x)
} else {
None
}
}
TreeNode::Leaf(_n) => {
path.compute_depths();
end = true;
None
}
}
})?
.flatten()
{
cur_node = new;
}
if end {
break;
}
if restart {
let node = index.get_root_refresh()?.expect("Root is there");
cur_node = PosRef::new(k, 0, node);
path.clear();
}
}
path.short_to_depth(depth);
Ok(path)
}
fn recompute_path_depth_pre<K, V, I>(index: &mut I, depth: BottomDepth, k: &K) -> PIRes<Path<K>>
where
K: IndexOrd + Clone,
I: IndexModify<K, V>,
{
let node = index.get_root_refresh()?.expect("Root is there");
let mut path = Path::new(k.clone());
let mut cur_node = PosRef::new(k, 0, node);
loop {
let mut restart = true;
let mut end = false;
if let Some(new) = index
.load_from_node(&cur_node.node_ref, |node, version| {
path.push(cur_node.clone(), version, node.len());
match node {
TreeNode::Node(n) => {
if let Some(x) = n.find_pre_write(k) {
restart = false;
Some(x)
} else {
None
}
}
TreeNode::Leaf(_n) => {
path.replace_search_key(cur_node.k.clone());
path.compute_depths();
end = true;
None
}
}
})?
.flatten()
{
cur_node = new;
}
if end {
break;
}
if restart {
let node = index.get_root_refresh()?.expect("Root is there");
cur_node = PosRef::new(k, 0, node);
path.clear();
}
}
path.short_to_depth(depth);
Ok(path)
}
fn lock_parents<K, V, I>(
index: &mut I,
updates: Vec<LockGroup<K>>,
) -> PIRes<(Vec<LockGroup<K>>, Vec<Vec<ToMergeCheck<K>>>)>
where
K: IndexTypeInternal,
V: IndexTypeInternal,
I: IndexModify<K, V>,
{
let mut group_changes = LockGroups::new();
let mut checks_groups = Vec::new();
for update in &updates {
let mut parent_group = update.parent_group(index.limits());
let mut group_len = parent_group.len();
let mut cur = 0;
let mut checks = Vec::new();
let mut retry = 0;
while cur < group_len {
let recompute = {
let parent = &mut parent_group[cur];
if !parent.to_lock() {
cur += 1;
continue;
}
let path = parent.path().clone();
if lock_logic(index, path, &mut group_changes, Some(parent))? {
checks.extend(parent.children_checks());
cur += 1;
retry = 0;
false
} else {
true
}
};
if recompute {
retry += 1;
if retry > 1000000 {
return Err(crate::IndexChangeError::ReachedLimitOfRetry);
}
let mut children = Vec::<ChildChanged<K>>::new();
let mut inserted = 0;
let mut prev_path: Option<Path<K>> = None;
let old_children = parent_group[cur].children().clone();
for mut child in old_children {
let mut path = recompute_path(index, &child)?;
child.replace_item(path.pop().unwrap());
if let Some(pp) = prev_path {
if pp.last_id() != path.last_id() {
if inserted == 0 {
parent_group[cur].replace(pp.clone(), children);
} else {
let element = ParentChange::new_children(pp.clone(), children);
group_len += 1;
parent_group.insert(cur + inserted, element);
}
inserted += 1;
children = Vec::new();
}
}
prev_path = Some(path);
children.push(child);
}
if !children.is_empty() {
if let Some(pp) = prev_path {
if inserted == 0 {
parent_group[cur].replace(pp, children);
} else {
let element = ParentChange::new_children(pp, children);
group_len += 1;
parent_group.insert(cur + inserted, element);
}
}
}
}
}
if !checks.is_empty() {
checks_groups.push(checks);
}
}
group_changes.finish();
Ok((group_changes.groups(), checks_groups))
}
#[derive(Clone)]
pub(crate) struct ToMergeCheck<K> {
parent: PosRef<K>,
child: PosRef<K>,
check_key: bool,
}
impl<K> ToMergeCheck<K> {
fn new(parent: PosRef<K>, child: PosRef<K>, check_key: bool) -> Self {
Self {
parent,
child,
check_key,
}
}
}
fn rebalance<K, V, I>(index: &mut I, to_balance: Vec<ToMergeCheck<K>>) -> PIRes<()>
where
K: IndexOrd + Clone + Display,
I: IndexModify<K, V>,
V: Display,
{
let mut prev: Option<(ToMergeCheck<K>, usize)> = None;
for rmc in &to_balance {
let pos = &rmc.child;
let node_data = index.load_from_node(&pos.node_ref, |n, _| (n.len(), n.get_prev().clone()))?;
if let Some((node_len, prev_key)) = node_data {
if let Some((prev_id, prev_node_len)) = prev {
if node_len < index.limits().bottom() || prev_node_len < index.limits().bottom() {
let mut source_merge = index.delete_to_own(&pos.node_ref)?;
index.update_with(&prev_id.child.node_ref, move |dest_merge| {
let merge_key = source_merge.get_prev().as_ref().unwrap().clone();
dest_merge.merge_right(&merge_key, &mut source_merge);
(true, ())
})?;
let parent_id = rmc.parent.node_ref;
let next_to_set = index.update_with(&parent_id, |parent_node| {
let parent = parent_node.as_mut_node();
let parent_key = parent.find_key(&pos.node_ref).unwrap().clone();
(true, parent.remove_key(&parent_key))
})?;
if let Some(next) = next_to_set {
let prev_parent_id = prev_id.parent.node_ref;
index.update_with(&prev_parent_id, |prev_parent| {
let changed = prev_parent.as_mut_node().swap_next(&next);
(changed, ())
})?;
}
let node_len = index.load_from_node(&prev_id.child.node_ref, |n, _| n.len())?.unwrap();
prev = Some((prev_id, node_len));
} else if rmc.check_key {
let parent_id = rmc.parent.node_ref;
let new_key = prev_key.unwrap();
let prev_changed = index.update_with(&parent_id, |parent| {
let (changed, prev_changed) = parent.as_mut_node().replace_key(&pos.node_ref, &new_key);
(changed, prev_changed)
})?;
if prev_changed {
let prev_parent_id = prev_id.parent.node_ref;
index.update_with(&prev_parent_id, |parent| {
let node = parent.as_mut_node();
let changed = node.swap_next(&new_key);
(changed, ())
})?;
}
prev = Some((rmc.clone(), node_len));
} else {
prev = Some((rmc.clone(), node_len));
}
} else {
prev = Some((rmc.clone(), node_len));
}
}
}
if to_balance.len() == 1 {
let tb = &to_balance[0];
if tb.check_key {
let parent_id = tb.parent.node_ref;
let prev_key = index
.load_from_node(&tb.child.node_ref, |n, _| n.get_prev().clone())?
.unwrap();
let new_key = prev_key.unwrap();
index.update_with(&parent_id, |parent| {
let (changed, _) = parent.as_mut_node().replace_key(&tb.child.node_ref, &new_key);
(changed, ())
})?;
}
}
for rmc in &to_balance {
let pos = &rmc.child;
let top_limit = index.limits().top();
if index.load_from_node(&pos.node_ref, |_, _| ())?.is_some() {
let new_nodes = index.update_with(&pos.node_ref, |node| {
if node.len() > top_limit {
(true, Some(node.split(top_limit)))
} else {
(false, None)
}
})?;
if let Some(new_nodes) = new_nodes {
let mut ids = Vec::new();
for (k, new_node) in new_nodes {
let saved_id = index.insert(new_node)?;
ids.push((k, saved_id));
}
let rp = rmc.parent.node_ref;
index.update_with(&rp, |parent| {
parent.as_mut_node().insert_after_key(&ids);
(true, ())
})?;
}
}
}
Ok(())
}
struct ToLock<K> {
prev: Option<K>,
next: Option<K>,
len: usize,
path: Path<K>,
}
impl<K: Clone + IndexOrd> ToLock<K> {
fn from_node<V>(node: &TreeNode<K, V>, path: Path<K>) -> Self {
Self::new(node.get_prev(), node.get_next(), node.len(), path)
}
fn new(prev: &Option<K>, next: &Option<K>, len: usize, path: Path<K>) -> Self {
debug_assert!(path.last_bottom_depth().is_some());
ToLock {
prev: prev.clone(),
next: next.clone(),
len,
path,
}
}
}
struct ToLockNodes<K> {
to_lock: Vec<ToLock<K>>,
size_limit: isize,
}
impl<K: IndexOrd + Clone> ToLockNodes<K> {
fn new(size_limit: isize) -> Self {
Self {
to_lock: Default::default(),
size_limit,
}
}
fn push(&mut self, to_lock: ToLock<K>) {
self.size_limit -= to_lock.len as isize;
self.to_lock.push(to_lock);
}
fn are_one_after_the_other(&self) -> bool {
let mut prev_next: Option<Option<K>> = None;
for to_lock in &self.to_lock {
if let Some(pr) = prev_next {
if let (Some(f), Some(s)) = (pr, &to_lock.prev) {
if f.cmp(s) != Ordering::Equal {
return false;
}
} else {
return false;
}
}
prev_next = Some(to_lock.next.clone());
}
true
}
fn lock_all<I: IndexModify<K, V>, V>(&self, index: &mut I) -> bool {
let mut locked = Vec::new();
for to_lock in &self.to_lock {
let last = to_lock.path.last().unwrap();
if !index.lock(&last.pos_ref().node_ref, last.version()).unwrap_or(false) {
for to_unlock in locked {
index.unlock(&to_unlock);
}
return false;
}
locked.push(last.pos_ref().node_ref);
}
true
}
pub fn enough_nodes(&self) -> bool {
if self.to_lock.len() >= 2 {
self.size_limit <= 0
} else {
false
}
}
pub fn account_changes(&mut self, pc: &ParentChange<K>) {
self.size_limit += pc.add_count() as isize;
self.size_limit -= pc.remove_count() as isize;
}
}
fn get_prev_to_lock<K, V, I>(index: &mut I, key: &K, bottom_depth: BottomDepth) -> PIRes<Option<ToLock<K>>>
where
K: IndexTypeInternal,
V: IndexTypeInternal,
I: IndexModify<K, V>,
{
let pre_path = recompute_path_depth_pre(index, bottom_depth, key)?;
let pr = pre_path.last_pos_ref().unwrap().clone();
index.load_from_node(&pr.node_ref, |node, _| ToLock::from_node(node, pre_path))
}
fn add_nodes_to_lock<K, V, I>(
index: &mut I,
from: Option<&PosRef<K>>,
mut path: Path<K>,
to_lock: &mut ToLockNodes<K>,
bottom_depth: BottomDepth,
) -> PIRes<(bool, Option<K>)>
where
K: IndexTypeInternal,
V: IndexTypeInternal,
I: IndexModify<K, V>,
{
path.pop();
let parent_item = path.last().unwrap();
let parent = parent_item.pos_ref();
let nodes_pos = if let Some((nodes, next)) = index
.load_from_node(&parent.node_ref, |parent_n, ver| {
if parent_item.version() != ver {
return None;
}
let parent_node = if let TreeNode::Node(parent_node) = parent_n {
parent_node
} else {
return None;
};
let iter = parent_node.iter_from(from)?;
Some((iter.collect::<Vec<_>>(), parent_node.get_next().clone()))
})?
.flatten()
{
Some((nodes, next))
} else {
None
};
if let Some((iter, next)) = nodes_pos {
for next_pos in iter {
if let Some(lock_node) = index.load_from_node(&next_pos.node_ref, |node, version| {
let mut new_path = path.clone();
new_path.finish(next_pos.clone(), version, node.len(), bottom_depth);
ToLock::from_node(node, new_path)
})? {
to_lock.push(lock_node);
} else {
return Ok((false, None));
}
if to_lock.enough_nodes() {
return Ok((true, None));
}
}
Ok((true, next))
} else {
Ok((false, None))
}
}
fn lock_logic<K, V, I>(
index: &mut I,
path: Path<K>,
lock_groups: &mut LockGroups<K>,
pc: Option<&ParentChange<K>>,
) -> PIRes<bool>
where
K: IndexTypeInternal,
V: IndexTypeInternal,
I: IndexModify<K, V>,
{
let top_depth = path.depth();
let cur_item = path.last().unwrap();
let cur_node = cur_item.pos_ref();
let node_check = index
.load_from_node(&cur_node.node_ref, |node, version| {
if node.check_range(path.search_key()) && version == cur_item.version() {
Some(ToLock::from_node(node, path.clone()))
} else {
None
}
})?
.flatten();
let cur_to_lock = if let Some(tl) = node_check {
tl
} else {
return Ok(false);
};
let node_len = cur_to_lock.len;
let node_prev = &cur_to_lock.prev;
let bottom_depth = path.last_bottom_depth().unwrap();
if top_depth > TopDepth(0) {
let mut to_lock = ToLockNodes::new(index.limits().bottom() as isize);
if let Some(parent_change) = pc {
to_lock.account_changes(parent_change);
if parent_change.change_prev_key() {
if let Some(key) = node_prev {
if let Some(lock_node) = get_prev_to_lock(index, key, bottom_depth)? {
to_lock.push(lock_node);
} else {
return Ok(false);
}
}
}
}
to_lock.push(cur_to_lock);
let (success, mut next_key) =
add_nodes_to_lock(index, Some(cur_node), path.clone(), &mut to_lock, bottom_depth)?;
if !success {
return Ok(false);
}
while !to_lock.enough_nodes() && next_key.is_some() {
if let Some(key) = next_key.take() {
let parent_path = recompute_path_depth(index, bottom_depth, &key)?;
let (success, new_next_key) = add_nodes_to_lock(index, None, parent_path, &mut to_lock, bottom_depth)?;
if !success {
return Ok(false);
}
next_key = new_next_key;
}
}
if !to_lock.are_one_after_the_other() {
return Ok(false);
}
if !to_lock.lock_all(index) {
return Ok(false);
}
let first_path = &to_lock.to_lock.first().unwrap().path;
let lock_group = lock_groups.group_for_id(first_path.last_pos_ref().unwrap(), bottom_depth);
for to_lock in to_lock.to_lock {
lock_group.add(to_lock.path, to_lock.len, &pc);
}
Ok(true)
} else if index.lock(&cur_node.node_ref, cur_item.version())? {
let lock_group = lock_groups.group_for_id(cur_node, bottom_depth);
lock_group.add(path, node_len, &pc);
Ok(true)
} else {
Ok(false)
}
}
fn find_and_change_leaf<K, V, I>(change: &KeyChanges<K, V>, updates: &mut LockGroups<K>, index: &mut I) -> PIRes<()>
where
K: IndexTypeInternal,
V: IndexTypeInternal,
I: IndexModify<K, V>,
{
let mut next: Option<PosRef<K>> = None;
let mut path = Path::new(change.k.clone());
let mut retry_count = 0;
let value_mode = index.value_mode();
let index_name = index.index_name().to_owned();
loop {
next = if let Some(cur_node) = &mut next {
let node_ref = &cur_node.node_ref;
if let Some((new_next, reached_leaf)) = index.load_from_node(node_ref, |node, version| match node {
TreeNode::Node(n) => {
path.push(cur_node.clone(), version, node.len());
(n.find_write(&change.k), false)
}
TreeNode::Leaf(_none) => {
path.push(cur_node.clone(), version, node.len());
path.compute_depths();
(None, true)
}
})? {
if reached_leaf {
let locked = lock_logic(index, path, updates, None)?;
if locked {
let cr = index.update_with(node_ref, |loaded| {
let cr = change.apply(loaded.as_mut_leaf(), value_mode.clone(), &index_name);
let changed = if let Ok(changes) = &cr {
changes.add.is_some() || changes.remove.is_some()
} else {
false
};
(changed, cr)
})??;
if cr.add.is_some() || cr.remove.is_some() {
updates.last_set_change(&cur_node.node_ref, cr);
}
break;
} else {
path = Path::new(change.k.clone());
}
None
} else {
new_next
}
} else {
None
}
} else if let Some(r) = index.get_root_refresh()? {
retry_count += 1;
if retry_count > 1000000 {
return Err(crate::IndexChangeError::ReachedLimitOfRetry);
}
path.clear();
Some(PosRef::new(&change.k, 0, r))
} else {
retry_count += 1;
if retry_count > 1000000 {
return Err(crate::IndexChangeError::ReachedLimitOfRetry);
}
path.clear();
index.lock_config()?;
if let Some(r) = index.get_root_refresh()? {
Some(PosRef::new(&change.k, 0, r))
} else {
let mut leaf = Leaf::new();
change.apply(&mut leaf, index.value_mode(), index.index_name())?;
let leaf_ref = index.insert(TreeNode::Leaf(leaf))?;
index.set_root(Some(leaf_ref))?;
break;
}
}
}
Ok(())
}
impl<K, V, T> IndexApply<K, V> for T
where
K: IndexTypeInternal,
V: IndexTypeInternal,
T: IndexModify<K, V>,
T: Index<K, V>,
{
fn apply(&mut self, adds: &[KeyChanges<K, V>]) -> PIRes<()> {
let mut changes_group = LockGroups::new();
for add in adds {
find_and_change_leaf(add, &mut changes_group, self)?;
}
changes_group.finish();
let mut updates = changes_group.groups();
let mut merge_and_save = Vec::new();
while updates.len() > 1 || (updates.len() == 1 && updates[0].is_not_only_root()) {
let (parent_updates, parent_changes) = lock_parents(self, updates)?;
merge_and_save.push(parent_changes);
updates = parent_updates;
}
for parent_changes in merge_and_save {
for update in parent_changes {
rebalance(self, update)?;
}
}
if updates.len() == 1 && updates[0].is_only_root() {
self.lock_config()?;
while let Some(r) = self.get_root_refresh()? {
let vals = self.load_from_node(&r, |node, n_version| {
let len = node.len();
let single = if len == 1 {
if let TreeNode::Node(cn) = node {
Some((cn.get(0), n_version))
} else {
None
}
} else {
None
};
(len, single)
})?;
if let Some((len, single)) = vals {
if len > self.limits().top() {
let ids = split_and_save(self, r)?;
let node = TreeNode::Node(Node::new_from_split(r, &ids));
let cur_id = self.insert(node)?;
self.set_root(Some(cur_id))?;
} else if len == 1 {
if let Some((new, n_version)) = single {
self.delete(&r, n_version)?;
self.set_root(Some(new))?;
} else {
break;
}
} else if len == 0 {
self.set_root(None)?;
} else {
break;
}
}
}
}
Ok(())
}
}
impl<K, V, T> Index<K, V> for T
where
K: IndexTypeInternal,
V: IndexTypeInternal,
T: IndexKeeper<K, V>,
{
fn get(&mut self, k: &K) -> PERes<Option<Value<V>>> {
Ok(if let Some(node) = self.get_root()? {
let mut cur_node = node;
let mut reuse = None;
loop {
match self.load_with(&cur_node, reuse)? {
TreeNode::Node(n) => {
cur_node = n.find(k).node_ref;
reuse = Some(n);
}
TreeNode::Leaf(leaf) => {
break leaf.find(k).map(|el| el.1).ok();
}
}
}
} else {
None
})
}
fn iter_from(&mut self, first: Bound<&K>) -> PERes<PageIter<K, V>> {
let mut path = Vec::new();
let mut iter = if let Some(mut cur_node) = self.get_root()? {
path.push((0, cur_node));
let mut reuse = None;
loop {
match self.load_with(&cur_node, reuse)? {
TreeNode::Node(n) => {
let value = match first {
Bound::Included(f) => {
let found = n.find(f);
(found.pos, found.node_ref)
}
Bound::Excluded(f) => {
let found = n.find(f);
(found.pos, found.node_ref)
}
Bound::Unbounded => (0, n.get(0)),
};
cur_node = value.1;
path.push(value);
reuse = Some(n);
}
TreeNode::Leaf(leaf) => {
break PageIter {
iter: leaf.iter_from(first).peekable(),
};
}
}
}
} else {
PageIter {
iter: Vec::new().into_iter().peekable(),
}
};
while iter.iter.peek().is_none() {
let prev = path.pop();
if let Some((_, node)) = path.last() {
let (pos, _) = prev.unwrap();
match self.load(node)? {
TreeNode::Node(n) => {
if n.len() > pos + 1 {
let mut cur_node = n.get(pos + 1);
loop {
match self.load(&cur_node)? {
TreeNode::Node(n) => {
cur_node = n.get(0);
}
TreeNode::Leaf(leaf) => {
iter = PageIter {
iter: leaf.iter_from(first).peekable(),
};
break;
}
}
}
}
}
TreeNode::Leaf(_leaf) => {
panic!("can't happen");
}
}
} else {
break;
}
}
Ok(iter)
}
fn back_iter_from(&mut self, last: Bound<&K>) -> PERes<PageIterBack<K, V>> {
let mut path = Vec::new();
let mut iter = if let Some(mut cur_node) = self.get_root()? {
path.push((0, cur_node));
let mut reuse = None;
loop {
match self.load_with(&cur_node, reuse)? {
TreeNode::Node(n) => {
let value = match last {
Bound::Included(f) => {
let found = n.find(f);
(found.pos, found.node_ref)
}
Bound::Excluded(f) => {
let found = n.find(f);
(found.pos, found.node_ref)
}
Bound::Unbounded => {
let pos = n.len() - 1;
(pos, n.get(pos))
}
};
cur_node = value.1;
path.push(value);
reuse = Some(n);
}
TreeNode::Leaf(leaf) => {
break PageIterBack {
iter: leaf.back_iter_from(last).peekable(),
};
}
}
}
} else {
PageIterBack {
iter: Vec::new().into_iter().rev().peekable(),
}
};
while iter.iter.peek().is_none() {
let prev = path.pop();
if let Some((_, node)) = path.last() {
let (pos, _) = prev.unwrap();
match self.load(node)? {
TreeNode::Node(n) => {
if pos > 0 {
let mut cur_node = n.get(pos - 1);
loop {
match self.load(&cur_node)? {
TreeNode::Node(n) => {
cur_node = n.get(n.len() - 1);
}
TreeNode::Leaf(leaf) => {
iter = PageIterBack {
iter: leaf.back_iter_from(last).peekable(),
};
break;
}
}
}
}
}
TreeNode::Leaf(_leaf) => {
panic!("can't happen");
}
}
} else {
break;
}
}
Ok(iter)
}
}