pub(crate) fn path_for_key<'g, K: AsRef<[u8]>>(
        &self,
        key: K,
        guard: &'g Guard,
    ) -> Result<Vec<(&'g Frag, TreePtr<'g>)>, ()> {
        let _measure = Measure::new(&M.tree_traverse);

        let mut cursor = self.root.load(SeqCst);
        let mut path: Vec<(&'g Frag, TreePtr<'g>)> = vec![];

        // unsplit_parent is used for tracking need
        // to complete partial splits.
        let mut unsplit_parent: Option<usize> = None;

        let mut not_found_loops = 0;
        loop {
            if cursor == usize::max_value() {
                // this collection has been explicitly removed
                return Err(Error::CollectionNotFound(
                    self.tree_id.clone(),
                ));
            }
            let get_cursor = self
                .context
                .pagecache
                .get(cursor, guard)
                .map_err(|e| e.danger_cast())?;

            if get_cursor.is_free() || get_cursor.is_allocated() {
                // restart search from the tree's root
                not_found_loops += 1;
                debug_assert_ne!(
                    not_found_loops, 10_000,
                    "cannot find pid {} in path_for_key",
                    cursor
                );
                cursor = self.root.load(SeqCst);
                continue;
            }

            let (frag, cas_key) = match get_cursor {
                PageGet::Materialized(node, cas_key) => {
                    (node, cas_key)
                }
                broken => {
                    return Err(Error::ReportableBug(format!(
                        "got non-base node while traversing tree: {:?}",
                        broken
                    )));
                }
            };

            let node = frag.unwrap_base();

            // TODO this may need to change when handling (half) merges
            assert!(
                node.lo.as_ref() <= key.as_ref(),
                "overshot key somehow"
            );

            // half-complete split detect & completion
            // (when hi is empty, it means it's unbounded)
            if !node.hi.is_empty() && node.hi.as_ref() <= key.as_ref()
            {
                // we have encountered a child split, without
                // having hit the parent split above.
                cursor = node.next.expect(
                    "if our hi bound is not Inf (inity), \
                     we should have a right sibling",
                );
                if unsplit_parent.is_none() && !path.is_empty() {
                    unsplit_parent = Some(path.len() - 1);
                }
                continue;
            } else if let Some(idx) = unsplit_parent.take() {
                // we have found the proper page for
                // our split.
                let (parent_frag, parent_ptr) = &path[idx];
                let parent_node = parent_frag.unwrap_base();

                let ps = Frag::ParentSplit(ParentSplit {
                    at: node.lo.clone(),
                    to: node.id,
                });

                let link = self.context.pagecache.link(
                    parent_node.id,
                    parent_ptr.clone(),
                    ps,
                    guard,
                );
                match link {
                    Ok(_new_key) => {
                        // TODO set parent's cas_key (not this cas_key) to
                        // new_key in the path, along with updating the
                        // parent's node in the path vec. if we don't do
                        // both, we lose the newly appended parent split.
                    }
                    Err(Error::CasFailed(_)) => {}
                    Err(other) => return Err(other.danger_cast()),
                }
            }

            path.push((frag, cas_key.clone()));

            match path
                .last()
                .expect("we just pushed to path, so it's not empty")
                .0
                .unwrap_base()
                .data
            {
                Data::Index(ref ptrs) => {
                    let old_cursor = cursor;

                    let search = binary_search_lub(
                        ptrs,
                        |&(ref k, ref _v)| {
                            prefix_cmp_encoded(
                                k,
                                key.as_ref(),
                                &node.lo,
                            )
                        },
                    );

                    // This might be none if ord is Less and we're
                    // searching for the empty key
                    let index =
                        search.expect("failed to traverse index");

                    cursor = ptrs[index].1;

                    if cursor == old_cursor {
                        panic!("stuck in page traversal loop");
                    }
                }
                Data::Leaf(_) => {
                    break;
                }
            }
        }

        Ok(path)
    }
/// returns the traversal path, completing any observed
/// partially complete splits or merges along the way.
returns the traversal path, completing any observed
partially complete splits or merges along the way.
returns the traversal path completing any observed partially complete splits or merges along the way .
pub(crate) fn path_for_key<'g, K: AsRef<[u8]>>(
        &self,
        key: K,
        guard: &'g Guard,
    ) -> Result<Vec<(&'g Frag, TreePtr<'g>)>, ()> {
        let _measure = Measure::new(&M.tree_traverse);
        let mut cursor = self.root.load(SeqCst);
        let mut path: Vec<(&'g Frag, TreePtr<'g>)> = vec![];
        // unsplit_parent is used for tracking need
        // to complete partial splits.
        let mut unsplit_parent: Option<usize> = None;
        let mut not_found_loops = 0;
        loop {
            if cursor == usize::max_value() {
                // this collection has been explicitly removed
                return Err(Error::CollectionNotFound(
                    self.tree_id.clone(),
                ));
            }
            let get_cursor = self
                .context
                .pagecache
                .get(cursor, guard)
                .map_err(|e| e.danger_cast())?;
            if get_cursor.is_free() || get_cursor.is_allocated() {
                // restart search from the tree's root
                not_found_loops += 1;
                debug_assert_ne!(
                    not_found_loops, 10_000,
                    "cannot find pid {} in path_for_key",
                    cursor
                );
                cursor = self.root.load(SeqCst);
                continue;
            }
            let (frag, cas_key) = match get_cursor {
                PageGet::Materialized(node, cas_key) => {
                    (node, cas_key)
                }
                broken => {
                    return Err(Error::ReportableBug(format!(
                        "got non-base node while traversing tree: {:?}",
                        broken
                    )));
                }
            };
            let node = frag.unwrap_base();
            // TODO this may need to change when handling (half) merges
            assert!(
                node.lo.as_ref() <= key.as_ref(),
                "overshot key somehow"
            );
            // half-complete split detect & completion
            // (when hi is empty, it means it's unbounded)
            if !node.hi.is_empty() && node.hi.as_ref() <= key.as_ref()
            {
                // we have encountered a child split, without
                // having hit the parent split above.
                cursor = node.next.expect(
                    "if our hi bound is not Inf (inity), \
                     we should have a right sibling",
                );
                if unsplit_parent.is_none() && !path.is_empty() {
                    unsplit_parent = Some(path.len() - 1);
                }
                continue;
            } else if let Some(idx) = unsplit_parent.take() {
                // we have found the proper page for
                // our split.
                let (parent_frag, parent_ptr) = &path[idx];
                let parent_node = parent_frag.unwrap_base();
                let ps = Frag::ParentSplit(ParentSplit {
                    at: node.lo.clone(),
                    to: node.id,
                });
                let link = self.context.pagecache.link(
                    parent_node.id,
                    parent_ptr.clone(),
                    ps,
                    guard,
                );
                match link {
                    Ok(_new_key) => {
                        // TODO set parent's cas_key (not this cas_key) to
                        // new_key in the path, along with updating the
                        // parent's node in the path vec. if we don't do
                        // both, we lose the newly appended parent split.
                    }
                    Err(Error::CasFailed(_)) => {}
                    Err(other) => return Err(other.danger_cast()),
                }
            }
            path.push((frag, cas_key.clone()));
            match path
                .last()
                .expect("we just pushed to path, so it's not empty")
                .0
                .unwrap_base()
                .data
            {
                Data::Index(ref ptrs) => {
                    let old_cursor = cursor;
                    let search = binary_search_lub(
                        ptrs,
                        |&(ref k, ref _v)| {
                            prefix_cmp_encoded(
                                k,
                                key.as_ref(),
                                &node.lo,
                            )
                        },
                    );
                    // This might be none if ord is Less and we're
                    // searching for the empty key
                    let index =
                        search.expect("failed to traverse index");
                    cursor = ptrs[index].1;
                    if cursor == old_cursor {
                        panic!("stuck in page traversal loop");
                    }
                }
                Data::Leaf(_) => {
                    break;
                }
            }
        }
        Ok(path)
    }
pub ( crate ) fn path_for_key < ' g , K : AsRef < [ u8 ] > > ( & self , key : K , guard : & ' g Guard , ) -> Result < Vec < ( & ' g Frag , TreePtr < ' g > ) > , ( ) > { let _measure = Measure :: new ( & M . tree_traverse ) ; let mut cursor = self . root . load ( SeqCst ) ; let mut path : Vec < ( & ' g Frag , TreePtr < ' g > ) > = vec ! [ ] ; // unsplit_parent is used for tracking need // to complete partial splits. let mut unsplit_parent : Option < usize > = None ; let mut not_found_loops = 0 ; loop { if cursor == usize :: max_value ( ) { // this collection has been explicitly removed return Err ( Error :: CollectionNotFound ( self . tree_id . clone ( ) , ) ) ; } let get_cursor = self . context . pagecache . get ( cursor , guard ) . map_err ( | e | e . danger_cast ( ) ) ? ; if get_cursor . is_free ( ) || get_cursor . is_allocated ( ) { // restart search from the tree's root not_found_loops += 1 ; debug_assert_ne ! ( not_found_loops 10_000 " " cursor ) ; cursor = self . root . load ( SeqCst ) ; continue ; } let ( frag , cas_key ) = match get_cursor { PageGet :: Materialized ( node , cas_key ) => { ( node , cas_key ) } broken => { return Err ( Error :: ReportableBug ( format ! ( " " broken ) ) ) ; } } ; let node = frag . unwrap_base ( ) ; // TODO this may need to change when handling (half) merges assert ! ( node lo as_ref ( ) key as_ref ( ) " " ) ; // half-complete split detect & completion // (when hi is empty, it means it's unbounded) if ! node . hi . is_empty ( ) && node . hi . as_ref ( ) <= key . as_ref ( ) { // we have encountered a child split, without // having hit the parent split above. cursor = node . next . expect ( " \
 " , ) ; if unsplit_parent . is_none ( ) && ! path . is_empty ( ) { unsplit_parent = Some ( path . len ( ) - 1 ) ; } continue ; } else if let Some ( idx ) = unsplit_parent . take ( ) { // we have found the proper page for // our split. let ( parent_frag , parent_ptr ) = & path [ idx ] ; let parent_node = parent_frag . unwrap_base ( ) ; let ps = Frag :: ParentSplit ( ParentSplit { at : node . lo . clone ( ) , to : node . id , } ) ; let link = self . context . pagecache . link ( parent_node . id , parent_ptr . clone ( ) , ps , guard , ) ; match link { Ok ( _new_key ) => { // TODO set parent's cas_key (not this cas_key) to // new_key in the path, along with updating the // parent's node in the path vec. if we don't do // both, we lose the newly appended parent split. } Err ( Error :: CasFailed ( _ ) ) => { } Err ( other ) => return Err ( other . danger_cast ( ) ) , } } path . push ( ( frag , cas_key . clone ( ) ) ) ; match path . last ( ) . expect ( " " ) . 0 . unwrap_base ( ) . data { Data :: Index ( ref ptrs ) => { let old_cursor = cursor ; let search = binary_search_lub ( ptrs , | & ( ref k , ref _v ) | { prefix_cmp_encoded ( k , key . as_ref ( ) , & node . lo , ) } , ) ; // This might be none if ord is Less and we're // searching for the empty key let index = search . expect ( " " ) ; cursor = ptrs [ index ] . 1 ; if cursor == old_cursor { panic ! ( " " ) ; } } Data :: Leaf ( _ ) => { break ; } } } Ok ( path ) }
returns the traversal path, completing any observed
partially complete splits or merges along the way.
returns the traversal path completing any observed partially complete splits or merges along the way .
&self
key
K
guard
&'g Guard
&self
key
K
guard
&'g Guard