use async_trait::async_trait;

use async_trait::*;

mod slice;
pub use slice::*;

#[async_trait(?Send)]

#[async_trait]

    fn compare(&self, _b: &Self) -> core::cmp::Ordering {

    async fn compare<T: LoadPage>(&self, _txn: &T, _b: &Self) -> core::cmp::Ordering {

    async fn drop<T: AllocPage>(&self, txn: &mut T) -> Result<(), T::Error> {

    async unsafe fn drop<T: AllocPage>(&self, txn: &mut T) -> Result<(), T::Error> {

        #[async_trait::async_trait]

            fn compare(&self, b: &Self) -> core::cmp::Ordering {

            async fn compare<T: LoadPage>(&self, _txn: &T, b: &Self) -> core::cmp::Ordering {

        #[async_trait::async_trait]

            unsafe fn write_to_page(&self, p: *mut u8) {
                core::ptr::copy_nonoverlapping(self, p as *mut Self, 1)

            async unsafe fn write_to_page<T: AllocPage>(&self, _txn: &mut T, p: Mut<u8>) {
                core::ptr::copy_nonoverlapping(self, p.0 as *mut Self, 1)

            unsafe fn from_raw_ptr<'a>(p: *const u8) -> &'a Self {
                &*(p as *const Self)

            async unsafe fn from_raw_ptr<'a, T>(_: &T, p: Const<u8>) -> &'a Self {
                &*(p.0 as *const Self)

#[derive(Clone, Copy)]
pub struct Mut<T>(pub *mut T);
unsafe impl<T> Send for Mut<T> {}
impl<T> Mut<T> {
    pub unsafe fn new(p: *mut T) -> Self {
        Mut(p)
    }
    pub fn into_inner(self) -> *mut T {
        self.0
    }
}
#[derive(Clone, Copy)]
pub struct Const<T>(pub *const T);
unsafe impl<T> Send for Const<T> {}
impl<T> Const<T> {
    pub unsafe fn new(p: *const T) -> Self {
        Const(p)
    }
    pub fn into_inner(self) -> *const T {
        self.0
    }
}

#[async_trait]

    unsafe fn write_to_page(&self, p: *mut u8);

    async unsafe fn write_to_page<T: AllocPage>(&self, txn: &mut T, p: Mut<u8>);

    unsafe fn from_raw_ptr<'a>(p: *const u8) -> &'a Self;

    async unsafe fn from_raw_ptr<'a, T>(_: &T, p: Const<u8>) -> &'a Self;

#[async_trait]

    fn compare(&self, b: &Self) -> core::cmp::Ordering {

    async fn compare<T: LoadPage>(&self, _txn: &T, b: &Self) -> core::cmp::Ordering {

#[async_trait]

    unsafe fn from_raw_ptr<'a>(p: *const u8) -> &'a Self {
        let len = u16::from_le(*(p as *const u16));

    async unsafe fn from_raw_ptr<'a, T>(_: &T, p: Const<u8>) -> &'a Self {
        let len = u16::from_le(*(p.0 as *const u16));

        core::slice::from_raw_parts(p.add(2), len as usize)

        core::slice::from_raw_parts(p.0.add(2), len as usize)

    unsafe fn write_to_page(&self, p: *mut u8) {

    async unsafe fn write_to_page<T: AllocPage>(&self, _txn: &mut T, p: Mut<u8>) {

        *(p as *mut u16) = (self.len() as u16).to_le();
        core::ptr::copy_nonoverlapping(self.as_ptr(), p.add(2), self.len())

        *(p.0 as *mut u16) = (self.len() as u16).to_le();
        core::ptr::copy_nonoverlapping(self.as_ptr(), p.0.add(2), self.len())

            hasher.update(core::slice::from_raw_parts(
                self.data.add(8),
                PAGE_SIZE - 8,
            ));

            hasher.update(core::slice::from_raw_parts(self.data.add(8), PAGE_SIZE - 8));

#[cfg(not(feature = "crc32"))]
pub unsafe fn clear_dirty_(data: *mut u8) {
    *data &= 0xfe
}

#[cfg(feature = "crc32")]
pub unsafe fn clear_dirty_(data: *mut u8, hasher: &crc32fast::Hasher) {
    *data &= 0xfe;
    let crc = (data as *mut u32).add(1);
    *crc = self.0.crc(hasher)
}

#[async_trait(?Send)]

#[async_trait]

    type Error;

    type Error: core::fmt::Debug + Send;
    /// Loading a page.
    async unsafe fn load_page(&self, off: u64) -> Result<CowPage, Self::Error>;

    async fn load_page(&self, off: u64) -> Result<CowPage, Self::Error>;

    async unsafe fn load_page_contiguous(
        &self,
        _off: u64,
        _length: u64,
    ) -> Result<CowPage, Self::Error> {
        unimplemented!()
    }

#[async_trait(?Send)]

#[async_trait]

    async fn alloc_page(&mut self) -> Result<MutPage, Self::Error>;

    async unsafe fn alloc_page(&mut self) -> Result<MutPage, Self::Error>;
    /// Allocate a new page, and mark it as not requiring cleaning of the dirty bit.
    async unsafe fn alloc_page_nodirty(&mut self) -> Result<MutPage, Self::Error> {
        unimplemented!()
    }
    /// Allocate a new block.
    async unsafe fn alloc_contiguous(&mut self, length: u64) -> Result<MutPage, Self::Error>;

    async fn decr_rc(&mut self, off: u64) -> Result<usize, Self::Error>;

    async unsafe fn decr_rc(&mut self, off: u64) -> Result<usize, Self::Error>;

    async fn decr_rc_owned(&mut self, off: u64) -> Result<usize, Self::Error>;

    async unsafe fn decr_rc_owned(&mut self, off: u64) -> Result<usize, Self::Error>;

    let mut cursor = Cursor::new(txn, db).await?;

    let mut cursor = unsafe { Cursor::new(txn, db).await? };

    ).await?;

    )
    .await?;

    db.db = put_cascade(txn, &mut cursor, put, &mut free).await?.0.offset;
    for f in &free[..p] {
        if *f & 1 != 0 {
            txn.decr_rc_owned((*f) ^ 1).await?;
        } else if *f > 0 {
            txn.decr_rc(*f).await?;

    db.db = put_cascade(txn, &mut cursor, put, &mut free)
        .await?
        .0
        .offset;
    unsafe {
        for f in &free[..p] {
            if *f & 1 != 0 {
                txn.decr_rc_owned((*f) ^ 1).await?;
            } else if *f > 0 {
                txn.decr_rc(*f).await?;
            }

        'a,

    'a,

    P: BTreeMutPage<K, V>,

    P: BTreeMutPage<K, V> + Send,

                    ).await?;

                    )
                    .await?;

                    let mut p = txn.alloc_page().await?;

                    let mut p = unsafe { txn.alloc_page().await? };

                    ).await? {

                    )
                    .await?
                    {

                    put = Put::Ok(P::update_left_child(
                        txn,
                        curs.page,
                        is_owned,
                        &curs.cursor,
                        page.0.offset,
                    ).await?)

                    put = Put::Ok(
                        P::update_left_child(txn, curs.page, is_owned, &curs.cursor, page.0.offset)
                            .await?,
                    )

    P: BTreePage<K, V>,

    P: BTreePage<K, V> + Send,

        while let Some((k, v, r)) = P::next(cur.page.as_page(), &mut c) {

        while let Some((k, v, r)) = P::next(txn, cur.page.as_page(), &mut c).await {

#[async_trait(?Send)]

#[async_trait]

    fn current<'a>(

    async fn current<'a, T: LoadPage>(
        txn: &T,

                let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize));

                let (k, v) = {
                    let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize));
                    (Const(k as *const u8), Const(v as *const u8))
                };

                    K::from_raw_ptr(k as *const u8),
                    V::from_raw_ptr(v as *const u8),

                    K::from_raw_ptr(txn, k).await,
                    V::from_raw_ptr(txn, v).await,

                let (k, v) = read::<K, V>(page.data.as_ptr().add((off & 0xfff) as usize));

                let (k, v) = {
                    let (k, v) = read::<K, V>(page.data.as_ptr().add((off & 0xfff) as usize));
                    (Const(k as *const u8), Const(v as *const u8))
                };

                    K::from_raw_ptr(k as *const u8),
                    V::from_raw_ptr(v as *const u8),

                    K::from_raw_ptr(txn, k).await,
                    V::from_raw_ptr(txn, v).await,

            let off = unsafe { *(page.data.as_ptr().offset(HDR as isize + c.cur * 8) as *const u64) };

            let off =
                unsafe { *(page.data.as_ptr().offset(HDR as isize + c.cur * 8) as *const u64) };

        _txn: &'a T,

        txn: &T,

            match lookup(page, c, k0, v0).await {

            match lookup(txn, page, c, k0, v0).await {

                        let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize));
                        Ok((K::from_raw_ptr(k), V::from_raw_ptr(v), 0))

                        let (k, v) = {
                            let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize));
                            let k = Const(k as *const u8);
                            let v = Const(v as *const u8);
                            (k, v)
                        };
                        Ok((
                            K::from_raw_ptr(txn, k).await,
                            V::from_raw_ptr(txn, v).await,
                            0,
                        ))

                        let (k, v) =
                            read::<K, V>(page.data.as_ptr().add(off as usize & 0xfff));

                        let (k, v) = {
                            let (k, v) = read::<K, V>(page.data.as_ptr().add(off as usize & 0xfff));
                            (Const(k as *const u8), Const(v as *const u8))
                        };

                            K::from_raw_ptr(k),
                            V::from_raw_ptr(v),

                            K::from_raw_ptr(txn, k).await,
                            V::from_raw_ptr(txn, v).await,

async unsafe fn lookup<'a, K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(

async unsafe fn lookup<
    'a,
    K: UnsizedStorable + ?Sized,
    V: UnsizedStorable + ?Sized,
    T: LoadPage,
>(
    txn: &T,

        lookup_leaf(page, k0, v0, hdr)

        lookup_leaf(txn, page, k0, v0, hdr).await

        lookup_internal(page, k0, v0, hdr)

        lookup_internal(txn, page, k0, v0, hdr).await
    }
}
async fn binary_search_by<'a, T, F, O>(s: &'a [T], mut f: F) -> Result<usize, usize>
where
    F: FnMut(&'a T) -> O,
    O: core::future::Future<Output = Ordering>,
{
    use Ordering::*;
    let mut size = s.len();
    let mut left = 0;
    let mut right = size;
    while left < right {
        let mid = left + size / 2;
        // SAFETY: the while condition means `size` is strictly positive, so
        // `size/2 < size`. Thus `left + size/2 < left + size`, which
        // coupled with the `left + size <= self.len()` invariant means
        // we have `left + size/2 < self.len()`, and this is in-bounds.
        let cmp = f(unsafe { s.get_unchecked(mid) }).await;
        // The reason why we use if/else control flow rather than match
        // is because match reorders comparison operations, which is perf sensitive.
        // This is x86 asm for u8: https://rust.godbolt.org/z/8Y8Pra.
        if cmp == Less {
            left = mid + 1;
        } else if cmp == Greater {
            right = mid;
        } else {
            return Ok(mid);
        }
        size = right - left;

    Err(left)

unsafe fn lookup_internal<'a, K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(

async unsafe fn lookup_internal<
    'a,
    K: UnsizedStorable + ?Sized,
    V: UnsizedStorable + ?Sized,
    T: LoadPage,
>(
    txn: &T,

    let s = core::slice::from_raw_parts(
        page.data.as_ptr().add(HDR) as *const u64,
        hdr.n() as usize,
    );

    let s =
        core::slice::from_raw_parts(page.data.as_ptr().add(HDR) as *const u64, hdr.n() as usize);

        s.binary_search_by(|&off| {

        binary_search_by(s, |&off| async move {

            let (k, v) = read::<K, V>(page.data.as_ptr().offset(off as isize & 0xfff));
            let k = K::from_raw_ptr(k);
            match k.compare(k0) {

            let (k, v) = {
                let (k, v) = read::<K, V>(page.data.as_ptr().offset(off as isize & 0xfff));
                (Const(k), Const(v))
            };
            let k = K::from_raw_ptr(txn, k).await;
            match k.compare(txn, k0).await {

                    let v = V::from_raw_ptr(v);
                    v.compare(v0)

                    let v = V::from_raw_ptr(txn, v).await;
                    v.compare(txn, v0).await

        .await

        match s.binary_search_by(|&off| {

        match binary_search_by(s, |&off| async move {

            let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize & 0xfff));
            let k = K::from_raw_ptr(k);
            k.compare(k0)
        }) {

            let k = Const(read::<K, V>(page.data.as_ptr().offset(off as isize & 0xfff)).0);
            let k = K::from_raw_ptr(txn, k).await;
            k.compare(txn, k0).await
        })
        .await
        {

                    let off = u64::from_le(s[i-1]) & 0xfff;
                    let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize));
                    let k = K::from_raw_ptr(k);
                    if let Ordering::Equal = k.compare(k0) {

                    let off = u64::from_le(s[i - 1]) & 0xfff;
                    let k = Const(read::<K, V>(page.data.as_ptr().offset(off as isize)).0);
                    let k = K::from_raw_ptr(txn, k).await;
                    if let Ordering::Equal = k.compare(txn, k0).await {

                        break

                        break;

unsafe fn lookup_leaf<'a, K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized>(

async unsafe fn lookup_leaf<
    'a,
    K: UnsizedStorable + ?Sized,
    V: UnsizedStorable + ?Sized,
    T: LoadPage,
>(
    txn: &T,

    let s = core::slice::from_raw_parts(
        page.data.as_ptr().add(HDR) as *const u16,
        hdr.n() as usize,
    );

    let s =
        core::slice::from_raw_parts(page.data.as_ptr().add(HDR) as *const u16, hdr.n() as usize);

        s.binary_search_by(|&off| {

        binary_search_by(s, |&off| async move {

            let (k, v) = read::<K, V>(page.data.as_ptr().offset(off as isize));
            let k = K::from_raw_ptr(k as *const u8);
            match k.compare(k0) {

            let (k, v) = {
                let (k, v) = read::<K, V>(page.data.as_ptr().offset(off as isize));
                (Const(k as *const u8), Const(v as *const u8))
            };
            let k = K::from_raw_ptr(txn, k).await;
            match k.compare(txn, k0).await {

                    let v = V::from_raw_ptr(v as *const u8);
                    v.compare(v0)

                    let v = V::from_raw_ptr(txn, v).await;
                    v.compare(txn, v0).await

        .await

        match s.binary_search_by(|&off| {

        match binary_search_by(s, |&off| async move {

            let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize));
            let k = K::from_raw_ptr(k);
            k.compare(k0)
        }) {

            let k = Const(read::<K, V>(page.data.as_ptr().offset(off as isize)).0);
            let k = K::from_raw_ptr(txn, k).await;
            k.compare(txn, k0).await
        })
        .await
        {

                    let off = u16::from_le(s[i-1]);
                    let (k, _) = read::<K, V>(page.data.as_ptr().offset(off as isize));
                    let k = K::from_raw_ptr(k);
                    if let Ordering::Equal = k.compare(k0){

                    let off = u16::from_le(s[i - 1]);
                    let k = Const(read::<K, V>(page.data.as_ptr().offset(off as isize)).0);
                    let k = K::from_raw_ptr(txn, k).await;
                    if let Ordering::Equal = k.compare(txn, k0).await {

                        break

                        break;

#[async_trait(?Send)]

#[async_trait]

            ).await

            )
            .await

            ).await

            )
            .await

    unsafe fn put_mut(

    async unsafe fn put_mut<T: AllocPage>(
        txn: &mut T,

            Leaf::alloc_write(page, k0, v0, 0, r, &mut n);

            Leaf::alloc_write(txn, page, k0, v0, 0, r, &mut n).await;

            Internal::alloc_write(page, k0, v0, 0, r, &mut n);

            Internal::alloc_write(txn, page, k0, v0, 0, r, &mut n).await;

    unsafe fn set_left_child(
        page: &mut MutPage,
        c: &Self::Cursor,
        l: u64
    ) {

    unsafe fn set_left_child(page: &mut MutPage, c: &Self::Cursor, l: u64) {

            let mut new = txn.alloc_page().await?;

            let mut new = unsafe { txn.alloc_page().await? };

            let mut new = txn.alloc_page().await?;

            let mut new = unsafe { txn.alloc_page().await? };

                return Ok(Op::Put(Self::put(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    m.modified.skip_first,
                    &m.modified.c1,
                    k,
                    v,
                    m.modified.ins2,
                    m.modified.l,
                    m.modified.r,
                ).await?));

                return Ok(Op::Put(
                    Self::put(
                        txn,
                        m.modified.page,
                        m.modified.mutable,
                        m.modified.skip_first,
                        &m.modified.c1,
                        k,
                        v,
                        m.modified.ins2,
                        m.modified.l,
                        m.modified.r,
                    )
                    .await?,
                ));

                ).await?;

                )
                .await?;

                return Ok(Op::Put(Put::Ok(Self::update_left_child(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    &c1,
                    l,
                ).await?)));

                return Ok(Op::Put(Put::Ok(
                    Self::update_left_child(txn, m.modified.page, m.modified.mutable, &c1, l)
                        .await?,
                )));

#[async_trait]

    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize);

    async fn alloc_write<T: AllocPage>(
        txn: &mut T,
        new: &mut MutPage,
        k0: &K,
        v0: &V,
        l: u64,
        r: u64,
        n: &mut isize,
    );

fn modify<K: core::fmt::Debug + ?Sized, V: core::fmt::Debug + ?Sized, P: BTreeMutPage<K, V>, L: AllocWrite<K, V>>(

async fn modify<
    'a,
    K: core::fmt::Debug + ?Sized,
    V: core::fmt::Debug + ?Sized,
    P: BTreeMutPage<K, V>,
    L: AllocWrite<K, V>,
    T: AllocPage,
>(
    txn: &mut T,

    m: &mut ModifiedPage<K, V, P>,

    m: &mut ModifiedPage<'a, K, V, P>,

    while let Some((k, v, r)) = P::next(m.page.as_page(), &mut m.c0) {
        L::alloc_write(new, k, v, l, r, n);

    while let Some((k, v, r)) = P::next(txn, m.page.as_page(), &mut m.c0).await {
        L::alloc_write(txn, new, k, v, l, r, n).await;

            L::alloc_write(new, k, v, l, m.l, n);
            L::alloc_write(new, k2, v2, 0, m.r, n);

            L::alloc_write(txn, new, k, v, l, m.l, n).await;
            L::alloc_write(txn, new, k2, v2, 0, m.r, n).await;

            L::alloc_write(new, k, v, m.l, m.r, n);

            L::alloc_write(txn, new, k, v, m.l, m.r, n).await;

            L::alloc_write(new, k, v, l, m.r, n);

            L::alloc_write(txn, new, k, v, l, m.r, n).await;

    while let Some((k, v, r)) = P::next(m.page.as_page(), &mut c1) {

    while let Some((k, v, r)) = P::next(txn, m.page.as_page(), &mut c1).await {

            L::alloc_write(new, k, v, l, rr, n);

            L::alloc_write(txn, new, k, v, l, rr, n).await;

            L::alloc_write(new, k, v, l, r, n);

            L::alloc_write(txn, new, k, v, l, r, n).await;

    let mut new = txn.alloc_page().await?;

    let mut new = unsafe { txn.alloc_page().await? };

        modify::<_, _, _, L>(&mut new, &mut m.modified, &mut n);

        modify::<_, _, _, L, _>(txn, &mut new, &mut m.modified, &mut n).await;

        L::alloc_write(&mut new, m.mid.0, m.mid.1, 0, l, &mut n);
        while let Some((k, v, r)) = P::next(m.other.as_page(), &mut rc) {
            L::alloc_write(&mut new, k, v, 0, r, &mut n);

        L::alloc_write(txn, &mut new, m.mid.0, m.mid.1, 0, l, &mut n).await;
        while let Some((k, v, r)) = P::next(txn, m.other.as_page(), &mut rc).await {
            L::alloc_write(txn, &mut new, k, v, 0, r, &mut n).await;

        while let Some((k, v, r)) = P::next(m.other.as_page(), &mut rc) {
            L::alloc_write(&mut new, k, v, l, r, &mut n);

        while let Some((k, v, r)) = P::next(txn, m.other.as_page(), &mut rc).await {
            L::alloc_write(txn, &mut new, k, v, l, r, &mut n).await;

        L::alloc_write(&mut new, m.mid.0, m.mid.1, 0, 0, &mut n);
        modify::<_, _, _, L>(&mut new, &mut m.modified, &mut n);

        L::alloc_write(txn, &mut new, m.mid.0, m.mid.1, 0, 0, &mut n).await;
        modify::<_, _, _, L, _>(txn, &mut new, &mut m.modified, &mut n).await;

    let (k, v, r) = <Page<K, V>>::current(m.other.as_page(), &rc).unwrap();

    let (k, v, r) = <Page<K, V>>::current(txn, m.other.as_page(), &rc)
        .await
        .unwrap();

        ).await? {

        )
        .await?
        {

        ).await? {

        )
        .await?
        {

        ).await?;

        )
        .await?;

        ).await? {

        )
        .await?
        {

    let (k0, v0, r0) = P::current(m.other.as_page(), &lc).unwrap();

    let (k0, v0, r0) = P::current(txn, m.other.as_page(), &lc).await.unwrap();

        ).await? {

        )
        .await?
        {

        ).await? {

        )
        .await?
        {

        ).await?;

        )
        .await?;

        ).await? {

        )
        .await?
        {

        ).await? {

        )
        .await?
        {

            L::alloc_write(&mut page, k0, v0, 0, 0, &mut n);
            L::alloc_write(&mut page, k1, v1, l, r, &mut n);

            L::alloc_write(txn, &mut page, k0, v0, 0, 0, &mut n).await;
            L::alloc_write(txn, &mut page, k1, v1, l, r, &mut n).await;

            L::alloc_write(&mut page, k0, v0, l, r, &mut n);

            L::alloc_write(txn, &mut page, k0, v0, l, r, &mut n).await;

        let mut new = txn.alloc_page().await?;

        let mut new = unsafe { txn.alloc_page().await? };

            L::alloc_write(&mut new, k0, v0, 0, l, &mut n);
            L::alloc_write(&mut new, k1, v1, 0, r, &mut n);

            L::alloc_write(txn, &mut new, k0, v0, 0, l, &mut n).await;
            L::alloc_write(txn, &mut new, k1, v1, 0, r, &mut n).await;

            L::alloc_write(&mut new, k0, v0, l, r, &mut n);

            L::alloc_write(txn, &mut new, k0, v0, l, r, &mut n).await;

    let mut left = txn.alloc_page().await?;

    let mut left = unsafe { txn.alloc_page().await? };

    let mut right = txn.alloc_page().await?;

    let mut right = unsafe { txn.alloc_page().await? };

                L::alloc_write(current_page, k0, v0, 0, l0, &mut n);

                L::alloc_write(txn, current_page, k0, v0, 0, l0, &mut n).await;

                L::alloc_write(current_page, k1, v1, 0, r0, &mut n);

                L::alloc_write(txn, current_page, k1, v1, 0, r0, &mut n).await;

                L::alloc_write(current_page, k0, v0, l0, r0, &mut n);

                L::alloc_write(txn, current_page, k0, v0, l0, r0, &mut n).await;

            let ptr = page.data.add(off);
            let size = entry_size::<K, V>(ptr);

            let ptr = Const(page.data.add(off));
            let size = entry_size::<K, V>(ptr.0);

                let (k, v) = read::<K, V>(ptr);
                split = Some((K::from_raw_ptr(k), V::from_raw_ptr(v)));

                let (k, v) = {
                    let (k, v) = read::<K, V>(ptr.0);
                    (Const(k), Const(v))
                };
                split = Some((K::from_raw_ptr(txn, k).await, V::from_raw_ptr(txn, v).await));

                    ptr,

                    ptr.0,

            L::alloc_write(current_page, k0, v0, 0, l0, &mut n);
            L::alloc_write(current_page, k1, v1, 0, r0, &mut n);

            L::alloc_write(txn, current_page, k0, v0, 0, l0, &mut n).await;
            L::alloc_write(txn, current_page, k1, v1, 0, r0, &mut n).await;

            L::alloc_write(current_page, k0, v0, l0, r0, &mut n);

            L::alloc_write(txn, current_page, k0, v0, l0, r0, &mut n).await;

#[async_trait]

    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {
        alloc_write::<K, V, Self>(new, k0, v0, l, r, n)

    async fn alloc_write<T: AllocPage>(
        txn: &mut T,
        new: &mut MutPage,
        k0: &K,
        v0: &V,
        l: u64,
        r: u64,
        n: &mut isize,
    ) {
        alloc_write::<K, V, Self, T>(txn, new, k0, v0, l, r, n).await

#[async_trait]

    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {
        alloc_write::<K, V, Self>(new, k0, v0, l, r, n)

    async fn alloc_write<T: AllocPage>(
        txn: &mut T,
        new: &mut MutPage,
        k0: &K,
        v0: &V,
        l: u64,
        r: u64,
        n: &mut isize,
    ) {
        alloc_write::<K, V, Self, T>(txn, new, k0, v0, l, r, n).await

fn alloc_write<K: UnsizedStorable + ?Sized, V: UnsizedStorable + ?Sized, L: Alloc>(

async fn alloc_write<
    K: UnsizedStorable + ?Sized,
    V: UnsizedStorable + ?Sized,
    L: Alloc,
    T: AllocPage,
>(
    txn: &mut T,

        let new_ptr = new.0.data.add(off_new);
        k0.write_to_page(new_ptr);

        let new_ptr = Mut(new.0.data.add(off_new));
        k0.write_to_page(txn, new_ptr).await;

        let v_ptr = new_ptr.add((ks + V::ALIGN - 1) & !(V::ALIGN - 1));
        v0.write_to_page(v_ptr);

        let v_ptr = Mut(new_ptr.0.add((ks + V::ALIGN - 1) & !(V::ALIGN - 1)));
        v0.write_to_page(txn, v_ptr).await;

#[async_trait(?Send)]

#[async_trait]

    fn current<'a>(

    async fn current<'a, T: LoadPage>(
        _txn: &T,

            let off =
                unsafe { *(page.data.as_ptr().offset((HDR as isize + c.cur * 8) - 8) as *const u64) };

            let off = unsafe {
                *(page.data.as_ptr().offset((HDR as isize + c.cur * 8) - 8) as *const u64)
            };

        _txn: &'a T,

        txn: &T,

            match lookup(page, c, k0, v0).await {

            match lookup(txn, page, c, k0, v0).await {

#[async_trait(?Send)]

#[async_trait]

            put::put::<_, _, _, Leaf>(txn, page, mutable, replace, c.cur as usize, k0, v0, 0, 0).await

            put::put::<_, _, _, Leaf>(txn, page, mutable, replace, c.cur as usize, k0, v0, 0, 0)
                .await

            put::put::<_, _, _, Internal>(txn, page, mutable, replace, c.cur as usize, k0, v0, l, r).await

            put::put::<_, _, _, Internal>(txn, page, mutable, replace, c.cur as usize, k0, v0, l, r)
                .await

    unsafe fn put_mut(page: &mut MutPage, c: &mut Self::Cursor, k0: &K, v0: &V, r: u64) {

    async unsafe fn put_mut<T: AllocPage>(
        txn: &mut T,
        page: &mut MutPage,
        c: &mut Self::Cursor,
        k0: &K,
        v0: &V,
        r: u64,
    ) {

            Leaf::alloc_write(page, k0, v0, 0, r, &mut n);

            Leaf::alloc_write(txn, page, k0, v0, 0, r, &mut n).await;

            Internal::alloc_write(page, k0, v0, 0, r, &mut n);

            Internal::alloc_write(txn, page, k0, v0, 0, r, &mut n).await;

            let mut new = txn.alloc_page().await?;

            let mut new = unsafe { txn.alloc_page().await? };

            let mut new = txn.alloc_page().await?;

            let mut new = unsafe { txn.alloc_page().await? };

                return Ok(Op::Put(Self::put(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    m.modified.skip_first,
                    &m.modified.c1,
                    k,
                    v,
                    m.modified.ins2,
                    m.modified.l,
                    m.modified.r,
                ).await?));

                return Ok(Op::Put(
                    Self::put(
                        txn,
                        m.modified.page,
                        m.modified.mutable,
                        m.modified.skip_first,
                        &m.modified.c1,
                        k,
                        v,
                        m.modified.ins2,
                        m.modified.l,
                        m.modified.r,
                    )
                    .await?,
                ));

                ).await?;

                )
                .await?;

                return Ok(Op::Put(Put::Ok(Self::update_left_child(
                    txn,
                    m.modified.page,
                    m.modified.mutable,
                    &c1,
                    l,
                ).await?)));

                return Ok(Op::Put(Put::Ok(
                    Self::update_left_child(txn, m.modified.page, m.modified.mutable, &c1, l)
                        .await?,
                )));

fn leaf_linear_search<K: Storable, V: Storable>(

async fn leaf_linear_search<K: Storable, V: Storable, T: LoadPage>(
    txn: &T,

        match sm.k.compare(k0) {

        match sm.k.compare(txn, k0).await {

                    match sm.v.compare(v0) {

                    match sm.v.compare(txn, v0).await {

fn cmp<K: Storable, V: Storable>(

async fn cmp<K: Storable, V: Storable, T: LoadPage>(
    txn: &T,

    let tup = unsafe { &*(p.as_ptr().offset(off as isize & 0xfff) as *const Tuple<K, V>) };
    match tup.k.compare(k0) {

    let (k, v) = unsafe {
        let ptr = &*(p.as_ptr().offset(off as isize & 0xfff) as *const Tuple<K, V>);
        (&ptr.k, &ptr.v)
    };
    match k.compare(txn, k0).await {

                tup.v.compare(v0)

                v.compare(txn, v0).await

unsafe fn internal_search<K: Storable, V: Storable>(

async unsafe fn internal_search<K: Storable, V: Storable, T: LoadPage>(
    txn: &T,

        match cmp(k0, v0, p, *off) {

        match cmp(txn, k0, v0, p, *off).await {

async unsafe fn lookup<'a, K: Storable, V: Storable>(

async unsafe fn lookup<'a, K: Storable, V: Storable, T: LoadPage>(
    txn: &T,

        leaf_linear_search(k0, v0, s)

        leaf_linear_search(txn, k0, v0, s).await

        internal_search(k0, v0, s, page.data)

        internal_search(txn, k0, v0, s, page.data).await

    let (k, v, r) = <Page<K, V>>::current(m.other.as_page(), &rc).unwrap();

    let (k, v, r) = <Page<K, V>>::current(txn, m.other.as_page(), &rc)
        .await
        .unwrap();

        ).await? {

        )
        .await?
        {

        if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {

        if let Put::Ok(Ok { page, freed }) =
            <Page<K, V>>::put(txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl).await?
        {

        ).await?;

        )
        .await?;

        if let Put::Ok(Ok { page, freed }) = <Page<K, V>>::put(
            txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {

        if let Put::Ok(Ok { page, freed }) =
            <Page<K, V>>::put(txn, page.0, true, false, &lc, m.mid.0, m.mid.1, None, 0, rl).await?
        {

            txn, m.modified.page, m.modified.mutable, false, &lc, m.mid.0, m.mid.1, None, 0, rl,
        ).await? {

            txn,
            m.modified.page,
            m.modified.mutable,
            false,
            &lc,
            m.mid.0,
            m.mid.1,
            None,
            0,
            rl,
        )
        .await?
        {

        let p = page.0.data;

        let p = Mut(page.0.data);

                let ptr = p.add(HDR + n as usize * 8) as *mut u64;

                let ptr = p.0.add(HDR + n as usize * 8) as *mut u64;

        L::alloc_write(&mut page, k0, v0, l, r, &mut n);

        L::alloc_write(txn, &mut page, k0, v0, l, r, &mut n).await;

        let mut new = txn.alloc_page().await?;

        let mut new = unsafe { txn.alloc_page().await? };

        L::alloc_write(&mut new, k0, v0, l, r, &mut n);

        L::alloc_write(txn, &mut new, k0, v0, l, r, &mut n).await;

        let mut right = txn.alloc_page().await?;

        let mut right = unsafe { txn.alloc_page().await? };

            L::alloc_write(&mut right, k0, v0, l, r, &mut n);

            L::alloc_write(txn, &mut right, k0, v0, l, r, &mut n).await;

            left = txn.alloc_page().await?;

            left = unsafe { txn.alloc_page().await? };

        left = txn.alloc_page().await?;

        left = unsafe { txn.alloc_page().await? };

        L::alloc_write(&mut left, k0, v0, l, r, &mut n);

        L::alloc_write(txn, &mut left, k0, v0, l, r, &mut n).await;

            right = txn.alloc_page().await?;

            right = unsafe { txn.alloc_page().await? };

#[async_trait]

    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {

    async fn alloc_write<T: AllocPage>(
        _txn: &mut T,
        new: &mut MutPage,
        k0: &K,
        v0: &V,
        l: u64,
        r: u64,
        n: &mut isize,
    ) {

#[async_trait]

    fn alloc_write(new: &mut MutPage, k0: &K, v0: &V, l: u64, r: u64, n: &mut isize) {

    async fn alloc_write<T: AllocPage>(
        _txn: &mut T,
        new: &mut MutPage,
        k0: &K,
        v0: &V,
        l: u64,
        r: u64,
        n: &mut isize,
    ) {

        // We can't do the following, since K and V don't implement UnsizedStorable.
        // k0.write_to_page(txn, Mut(&mut new_ptr.k as *mut u8)).await?;
        // v0.write_to_page(txn, Mut(&mut new_ptr.v as *mut u8)).await?;

#[async_trait(?Send)]
pub trait BTreePage<K: ?Sized, V: ?Sized>: Sized {

#[async_trait]
pub trait BTreePage<K: ?Sized, V: ?Sized>: Sized + Send {

    fn next<'b>(

    async fn next<'b, T: LoadPage>(
        txn: &T,

        if let Some((k, v, r)) = Self::current(p, c) {

        if let Some((k, v, r)) = Self::current(txn, p, c).await {

    fn prev<'b>(

    async fn prev<'b, T: LoadPage>(
        txn: &T,

            Self::current(p, c)

            Self::current(txn, p, c).await

    fn current<'a>(

    async fn current<'a, T: LoadPage>(
        txn: &T,

        txn: &'a T,

        txn: &T,

    _txn: &'a T,

    txn: &'a T,

impl<'a, T: LoadPage, K: ?Sized + 'a, V: ?Sized + 'a, P: BTreePage<K, V>> Iterator
    for PageIterator<'a, T, K, V, P>

impl<'a, T: LoadPage, K: ?Sized + 'a, V: ?Sized + 'a, P: BTreePage<K, V>>
    PageIterator<'a, T, K, V, P>

    type Item = (&'a K, &'a V, u64);
    fn next(&mut self) -> Option<Self::Item> {
        P::next(self.page, &mut self.cursor)

    pub async fn next(&mut self) -> Option<(&'a K, &'a V, u64)> {
        P::next(self.txn, self.page, &mut self.cursor).await

#[async_trait(?Send)]

#[async_trait]

    unsafe fn put_mut(

    async unsafe fn put_mut<T: AllocPage>(
        txn: &mut T,

    unsafe fn set_left_child(
        page: &mut MutPage,
        c: &Self::Cursor,
        l: u64
    );

    unsafe fn set_left_child(page: &mut MutPage, c: &Self::Cursor, l: u64);

unsafe impl<K, V, P: BTreePage<K, V>> Sync for Db_<K, V, P>{}

unsafe impl<K, V, P: BTreePage<K, V>> Sync for Db_<K, V, P> {}

#[async_trait(?Send)]
impl<K:Storable, V:Storable, P: BTreePage<K, V> + Send + core::fmt::Debug> Storable for Db_<K, V, P> {

#[async_trait]
impl<K: Storable, V: Storable, P: BTreePage<K, V> + core::fmt::Debug> Storable for Db_<K, V, P> {

    async fn drop<T: AllocPage>(&self, t: &mut T) -> Result<(), T::Error> {

    async unsafe fn drop<T: AllocPage>(&self, t: &mut T) -> Result<(), T::Error> {

pub async fn  create_db_<T: AllocPage, K: ?Sized, V: ?Sized, P: BTreeMutPage<K, V>>(

pub async unsafe fn create_db_<T: AllocPage, K: ?Sized, V: ?Sized, P: BTreeMutPage<K, V>>(

pub async fn create_db<T: AllocPage, K: Storable, V: Storable>(

pub async unsafe fn create_db<T: AllocPage, K: Storable, V: Storable>(

pub async fn fork_db<T: AllocPage, K: Storable + ?Sized, V: Storable + ?Sized, P: BTreeMutPage<K, V>>(

pub async fn fork_db<
    T: AllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreeMutPage<K, V>,
>(

pub async fn get_shared<'a, T: LoadPage, K: 'a + Storable + ?Sized, V: 'a + Storable + ?Sized, P: BTreePage<K, V>>(

pub async fn get_shared<
    'a,
    T: LoadPage,
    K: 'a + Storable + ?Sized,
    V: 'a + Storable + ?Sized,
    P: BTreePage<K, V>,
>(

    let mut page = txn.load_page(db.db).await?;

    let mut page = unsafe { txn.load_page(db.db).await? };

        } else if let Some((k, v, _)) = P::current(page.as_page(), &cursor) {

        } else if let Some((k, v, _)) = P::current(txn, page.as_page(), &cursor).await {

            unsafe { last_match = Some((core::mem::transmute(k), core::mem::transmute(v), is_shared)) }

            unsafe {
                last_match = Some((core::mem::transmute(k), core::mem::transmute(v), is_shared))
            }

            page = txn.load_page(next_page).await?;

            page = unsafe { txn.load_page(next_page).await? };

pub async fn get<'a, T: LoadPage, K: 'a + Storable + ?Sized, V: 'a + Storable + ?Sized, P: BTreePage<K, V>>(

pub async fn get<
    'a,
    T: LoadPage,
    K: 'a + Storable + ?Sized,
    V: 'a + Storable + ?Sized,
    P: BTreePage<K, V>,
>(

    let mut page = txn.load_page(db.db).await?;

    let mut page = unsafe { txn.load_page(db.db).await? };

        } else if let Some((k, v, _)) = P::current(page.as_page(), &cursor) {

        } else if let Some((k, v, _)) = P::current(txn, page.as_page(), &cursor).await {

            page = txn.load_page(next_page).await?;

            page = unsafe { txn.load_page(next_page).await? };

pub async fn drop<T: AllocPage, K: Storable + ?Sized, V: Storable + ?Sized, P: BTreePage<K, V>>(

pub async unsafe fn drop<
    T: AllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreePage<K, V>,
>(

struct PageCursor_<K:?Sized, V:?Sized, P: BTreePage<K, V>>(MaybeUninit<cursor::PageCursor<K, V, P>>);

struct PageCursor_<K: ?Sized, V: ?Sized, P: BTreePage<K, V>>(
    MaybeUninit<cursor::PageCursor<K, V, P>>,
);

unsafe impl<K: ?Sized, V:?Sized, P: BTreePage<K, V>> Send for PageCursor_<K, V, P> {}

unsafe impl<K: ?Sized, V: ?Sized, P: BTreePage<K, V>> Send for PageCursor_<K, V, P> {}

async fn drop_<T: AllocPage, K: Storable + ?Sized, V: Storable + ?Sized, P: BTreePage<K, V>>(

async unsafe fn drop_<
    T: AllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreePage<K, V>,
>(

                let cur = unsafe { &mut *stack[ptr].0.as_mut_ptr() };
                P::current(cur.page.as_page(), &cur.cursor)

                let (page, cursor) = unsafe {
                    let cur = &mut *stack[ptr].0.as_mut_ptr();
                    (cur.page.as_page(), &cur.cursor)
                };
                P::current(txn, page, cursor).await

                let cur = unsafe { &mut *stack[ptr].0.as_mut_ptr() };
                if P::move_next(&mut cur.cursor) {
                    Some(P::left_child(cur.page.as_page(), &cur.cursor))

                let (page, cursor) = unsafe {
                    let cur = &mut *stack[ptr].0.as_mut_ptr();
                    (cur.page.as_page(), &mut cur.cursor)
                };
                if P::move_next(cursor) {
                    Some(P::left_child(page, cursor))

use core::cmp::Ordering;

    K: Storable + ?Sized + PartialEq + Sync,
    V: Storable + ?Sized + PartialEq + Sync,

    K: Storable + ?Sized + Sync,
    V: Storable + ?Sized + Sync,

    let mut cursor = Cursor::new(txn, db).await?;

    let mut cursor = unsafe { Cursor::new(txn, db).await? };

        (Some((k, v)), Some(value)) if k == key && v == value => {}
        (Some((k, _)), None) if k == key => {}

        (Some((k, v)), Some(value))
            if k.compare(txn, key).await == Ordering::Equal
                && v.compare(txn, value).await == Ordering::Equal => {}
        (Some((k, _)), None) if k.compare(txn, key).await == Ordering::Equal => {}

    K: Storable + ?Sized + PartialEq + Sync,
    V: Storable + ?Sized + PartialEq + Sync,

    K: Storable + ?Sized + Sync,
    V: Storable + ?Sized + Sync,

    let mut cursor = Cursor::new(txn, db).await?;

    let mut cursor = unsafe { Cursor::new(txn, db).await? };

        (Some((k, v)), Some(value)) if k == key && v == value => {}
        (Some((k, _)), None) if k == key => {}

        (Some((k, v)), Some(value))
            if k.compare(txn, key).await == Ordering::Equal
                && v.compare(txn, value).await == Ordering::Equal => {}
        (Some((k, _)), None) if k.compare(txn, key).await == Ordering::Equal => {}

    K: Storable + ?Sized + PartialEq + Sync,

    K: Storable + ?Sized + Sync,

            P::current(cur.page.as_page(), &cur.cursor).unwrap()

            P::current(txn, cur.page.as_page(), &cur.cursor)
                .await
                .unwrap()

        delk.drop(txn).await?;
        delv.drop(txn).await?;

        unsafe {
            delk.drop(txn).await?;
            delv.drop(txn).await?;
        }

    cursor.set_first_leaf(txn, right_subtree).await?;

    unsafe { cursor.set_first_leaf(txn, right_subtree).await? };

        last_op = handle_merge(txn, cursor, p0, &k0v0, incr_page, incr_mid, mil, merge, &mut free).await?;

        last_op = handle_merge(
            txn, cursor, p0, &k0v0, incr_page, incr_mid, mil, merge, &mut free,
        )
        .await?;

    for p in free.iter() {
        if p[0] & 1 == 1 {
            txn.decr_rc_owned(p[0] ^ 1).await?;
        } else if p[0] > 0 {
            txn.decr_rc(p[0]).await?;
        }
        if p[1] & 1 == 1 {
            txn.decr_rc_owned(p[1] ^ 1).await?;
        } else if p[1] > 0 {
            txn.decr_rc(p[1]).await?;

    unsafe {
        for p in free.iter() {
            if p[0] & 1 == 1 {
                txn.decr_rc_owned(p[0] ^ 1).await?;
            } else if p[0] > 0 {
                txn.decr_rc(p[0]).await?;
            }
            if p[1] & 1 == 1 {
                txn.decr_rc_owned(p[1] ^ 1).await?;
            } else if p[1] > 0 {
                txn.decr_rc(p[1]).await?;
            }

        let (k, v, _) = P::current(curs0.page.as_page(), &c1).unwrap();

        let (k, v, _) = P::current(txn, curs0.page.as_page(), &c1).await.unwrap();

        let other = txn.load_page(P::left_child(curs.page.as_page(), &curs.cursor)).await?;

        let other = unsafe {
            txn.load_page(P::left_child(curs.page.as_page(), &curs.cursor))
                .await?
        };

            if let Some(x) = P::current(curs.page.as_page(), &curs.cursor) {

            if let Some(x) = P::current(txn, curs.page.as_page(), &curs.cursor).await {

                let (k, v, _) = P::prev(curs.page.as_page(), &mut curs.cursor).unwrap();

                let (k, v, _) = P::prev(txn, curs.page.as_page(), &mut curs.cursor)
                    .await
                    .unwrap();

        let other = txn.load_page(r).await?;

        let other = unsafe { txn.load_page(r).await? };

    K: Storable + ?Sized + PartialEq,

    K: Storable + ?Sized,

    mod_is_left: bool,    // The modified page in the `merge` is the left one.

    mod_is_left: bool, // The modified page in the `merge` is the left one.

        Op::Merged {.. } | Op::Rebalanced { .. } => {

        Op::Merged { .. } | Op::Rebalanced { .. } => {

                while let Some((k0, v0, r)) = P::next(other.as_page(), &mut curs) {

                while let Some((k0, v0, r)) = P::next(txn, other.as_page(), &mut curs).await {

        Op::Rebalanced {
            k,
            v,
            l,
            r,
            freed,
        } => {

        Op::Rebalanced { k, v, l, r, freed } => {

        'a,

    'a,

    while let Some((k, v, r)) = P::next(m.page.as_page(), &mut c0) {

    while let Some((k, v, r)) = P::next(txn, m.page.as_page(), &mut c0).await {

    if let Some((k, v, r)) = P::next(m.page.as_page(), &mut c1) {

    if let Some((k, v, r)) = P::next(txn, m.page.as_page(), &mut c1).await {

    while let Some((k, v, r)) = P::next(m.page.as_page(), &mut c1) {

    while let Some((k, v, r)) = P::next(txn, m.page.as_page(), &mut c1).await {

        'a,

    'a,

        if d > 0 {
            if last_op.page.is_dirty() {
                txn.decr_rc_owned(last_op.page.offset).await?;

        unsafe {
            if d > 0 {
                if last_op.page.is_dirty() {
                    txn.decr_rc_owned(last_op.page.offset).await?;
                } else {
                    txn.decr_rc(last_op.page.offset).await?;
                }
                db.db = d;

                txn.decr_rc(last_op.page.offset).await?;

                // The page becomes empty.
                let (page, freed) =
                    P::del(txn, last_op.page, last_op.mutable, &last_op.c1, d).await?;
                free[0][0] = freed;
                db.db = page.0.offset

            db.db = d;
        } else {
            // The page becomes empty.
            let (page, freed) = P::del(txn, last_op.page, last_op.mutable, &last_op.c1, d).await?;
            free[0][0] = freed;
            db.db = page.0.offset

            let mut page = txn.alloc_page().await?;

            let mut page = unsafe { txn.alloc_page().await? };

            let page = if let Put::Ok(p) = P::put(txn, page.0, true, false, &c, k, v, None, l, r).await? {
                p.page
            } else {
                unreachable!()
            };

            let page =
                if let Put::Ok(p) = P::put(txn, page.0, true, false, &c, k, v, None, l, r).await? {
                    p.page
                } else {
                    unreachable!()
                };

async fn modify<'a, T: AllocPage, K: Storable + ?Sized, V: Storable + ?Sized, P: BTreeMutPage<K, V>>(

async fn modify<
    'a,
    T: AllocPage,
    K: Storable + ?Sized,
    V: Storable + ?Sized,
    P: BTreeMutPage<K, V>,
>(

        ).await

        )
        .await

            Ok(Put::Ok(P::update_left_child(
                txn, m.page, m.mutable, &m.c1, m.l,
            ).await?))

            Ok(Put::Ok(
                P::update_left_child(txn, m.page, m.mutable, &m.c1, m.l).await?,
            ))

            Ok(Put::Ok(P::update_left_child(
                txn, m.page, m.mutable, &c1, m.r,
            ).await?))

            Ok(Put::Ok(
                P::update_left_child(txn, m.page, m.mutable, &c1, m.r).await?,
            ))

unsafe impl<K, V, P: BTreePage<K, V>> Send for PageCursor<K, V, P>{}
unsafe impl<K, V, P: BTreePage<K, V>> Sync for PageCursor<K, V, P>{}

unsafe impl<K, V, P: BTreePage<K, V>> Send for PageCursor<K, V, P> {}
unsafe impl<K, V, P: BTreePage<K, V>> Sync for PageCursor<K, V, P> {}

unsafe impl<K, V, P: BTreePage<K, V>> Send for Cursor<K, V, P>{}
unsafe impl<K, V, P: BTreePage<K, V>> Sync for Cursor<K, V, P>{}

struct StackElt<K: ?Sized, V: ?Sized, P: BTreePage<K, V>>(*mut PageCursor<K, V, P>);
unsafe impl<K, V, P: BTreePage<K, V>> Send for StackElt<K, V, P> {}
unsafe impl<K, V, P: BTreePage<K, V>> Sync for StackElt<K, V, P> {}
unsafe impl<K, V, P: BTreePage<K, V>> Send for Cursor<K, V, P> {}
unsafe impl<K, V, P: BTreePage<K, V>> Sync for Cursor<K, V, P> {}
impl<K: ?Sized, V: ?Sized, P: BTreePage<K, V>> StackElt<K, V, P> {
    unsafe fn page<'a>(&self) -> Page<'a> {
        (&*self.0).page.as_page()
    }
    unsafe fn cursor<'a>(&self) -> &'a P::Cursor {
        &(&*self.0).cursor
    }
}

    pub async fn new<T: LoadPage>(txn: &T, db: &Db_<K, V, P>) -> Result<Self, T::Error> {

    pub async unsafe fn new<T: LoadPage>(txn: &T, db: &Db_<K, V, P>) -> Result<Self, T::Error> {

        let mut stack = [

        let mut stack: [core::mem::MaybeUninit<PageCursor<K, V, P>>; N_CURSORS] = [

    }
    unsafe fn stack_elt(&mut self, i: usize) -> StackElt<K, V, P> {
        StackElt(self.stack[i].assume_init_mut())

    pub(super) async fn set_first_leaf<T: LoadPage>(

    pub(super) async unsafe fn set_first_leaf<T: LoadPage>(

        let init = unsafe { &mut *self.stack[0].as_mut_ptr() };

        let init = unsafe { self.stack[0].assume_init_mut() };

            let current = unsafe { &mut *self.stack.get_unchecked_mut(self.len - 1).as_mut_ptr() };

            let current = unsafe { self.stack.get_unchecked_mut(self.len - 1).assume_init_mut() };

            if let Ok((kk, vv, _)) = P::set_cursor(txn, current.page.as_page(), cursor, k, v).await {

            if let Ok((kk, vv, _)) = P::set_cursor(txn, current.page.as_page(), cursor, k, v).await
            {

                let page = txn.load_page(next_page).await?;
                self.push(PageCursor {

                let page = unsafe { txn.load_page(next_page).await? };
                // Pushing
                self.stack[self.len] = MaybeUninit::new(PageCursor {

                });
                self.len += 1;
                /*
                    self.push(PageCursor {
                        cursor: P::cursor_before(&page),
                        page,

                     */

    pub async fn set_last< T: LoadPage>(
        &mut self,
        txn: &'a T,
    ) -> Result<(), T::Error> {

    pub async fn set_last<T: LoadPage>(&mut self, txn: &'a T) -> Result<(), T::Error> {

        let current = unsafe { &mut *self.stack[self.len - 1].as_mut_ptr() };

        let current = unsafe { self.stack[self.len - 1].assume_init_mut() };

            let current = unsafe { &mut *self.stack[self.len - 1].as_mut_ptr() };

            let current = unsafe { self.stack[self.len - 1].assume_init_mut() };

                let page = txn.load_page(l).await?;

                let page = unsafe { txn.load_page(l).await? };

    pub fn current<T: LoadPage>(

    pub async fn current<T: LoadPage>(

        _txn: &'a T,

        txn: &'a T,

            let current = unsafe { &mut *self.stack[self.len - 1].as_mut_ptr() };
            if P::is_init(&current.cursor) {

            let current = unsafe { self.stack_elt(self.len - 1) };
            if P::is_init(unsafe { current.cursor() }) {

                return Ok(None)
            } else if let Some((k, v, _)) = P::current(current.page.as_page(), &current.cursor)

                return Ok(None);
            } else if let Some((k, v, _)) =
                unsafe { P::current(txn, current.page(), &current.cursor()).await }

    pub async fn next< T: LoadPage>(

    pub async fn next<T: LoadPage>(

            let current = unsafe { &mut *self.stack[self.len - 1].as_mut_ptr() };
            if P::is_empty(&current.cursor) {

            let current = unsafe { self.stack_elt(self.len - 1) };
            if P::is_empty(unsafe { current.cursor() }) {

                let (cur_entry, r) = if let Some((k, v, r)) = P::current(current.page.as_page(), &current.cursor) {
                    (Some((k, v)), r)
                } else {
                    (None, P::right_child(current.page.as_page(), &current.cursor))

                let (cur_entry, r) = {
                    let cur_page = unsafe { current.page() };
                    let cur_cursor = unsafe { current.cursor() };
                    if let Some((k, v, r)) = P::current(txn, cur_page, cur_cursor).await {
                        (Some((k, v)), r)
                    } else {
                        (None, P::right_child(cur_page, cur_cursor))
                    }

                P::move_next(&mut current.cursor);

                unsafe {
                    P::move_next(&mut (&mut *current.0).cursor);
                }

                    let page = txn.load_page(r).await?;
                    self.push(PageCursor {

                    let page = unsafe { txn.load_page(r).await? };
                    // Pushing
                    self.stack[self.len] = MaybeUninit::new(PageCursor {

                    self.len += 1;

    pub async fn prev< T: LoadPage>(

    pub async fn prev<T: LoadPage>(

            let current = unsafe { &mut *self.stack[self.len - 1].as_mut_ptr() };

            let current = unsafe { self.stack_elt(self.len - 1) };

            if P::is_init(&current.cursor) {

            if P::is_init(unsafe { current.cursor() }) {

                let cur_entry = P::current(current.page.as_page(), &current.cursor);
                let left = P::left_child(current.page.as_page(), &current.cursor);

                let cur_entry = unsafe { P::current(txn, current.page(), current.cursor()).await };
                let left = unsafe { P::left_child(current.page(), current.cursor()) };

                    let page = txn.load_page(left).await?;

                    let page = unsafe { txn.load_page(left).await? };

                    P::move_prev(&mut current.cursor);

                    unsafe { P::move_prev(&mut (&mut *current.0).cursor) };

    let mut cursor = Cursor::new(txn, db).await?;