let n = 10_000_000 as u64;

    let n = 10_000_000u64;

    let n = 1_000_000 as u64;

    let n = 1_000_000u64;

    let n = 100_000 as u64;

    let n = 100_000u64;

    let n = 2_000 as u64;

    let n = 2_000u64;

    let n = 20 as u64;
    let i0 = 10 as u64;

    let n = 20u64;
    let i0 = 10u64;

        btree::iter(&txn, &db, None)

        iter(&txn, &db, None)

    let n = 256 as u64;
    let i0 = 127 as u64;

    let n = 256u64;
    let i0 = 127u64;

        btree::iter(&txn, &db, None)

        iter(&txn, &db, None)

    let n = 19 as u64;

    let n = 19u64;

        let mut curs: btree::cursor::Cursor<_, _, B> =
            btree::cursor::Cursor::new(txn, &db_free).unwrap();

        let mut curs: Cursor<_, _, B> = btree::cursor::Cursor::new(txn, &db_free).unwrap();

    let mut child = unsafe { libc::fork() };

    let child = unsafe { libc::fork() };

        let env = Env::new("/tmp/sanakirja0", 4096 * 20, 2).unwrap();

        let env = Env::new("/tmp/sanakirja0/db", 4096 * 20, 2).unwrap();

        info!("started child mutable txn {:?}", txn.env.root.lock());

        info!("started child mutable txn {:?}", txn.root);
        assert_eq!(txn.root, 0);

        std::thread::sleep(std::time::Duration::from_secs(2));

        std::thread::sleep(std::time::Duration::from_millis(200));

        assert_eq!(txn.root, 1);

        assert!(t.elapsed().unwrap() >= std::time::Duration::from_millis(900));
        info!("started child mutable txn {:?}", txn.env.root.lock());

        assert!(t.elapsed().unwrap() >= std::time::Duration::from_millis(90));
        info!("started child mutable txn {:?}", txn.root);

        std::thread::sleep(std::time::Duration::from_secs(1));

        std::thread::sleep(std::time::Duration::from_millis(100));

        std::thread::sleep(std::time::Duration::from_secs(1));

        std::thread::sleep(std::time::Duration::from_millis(100));

        info!("starting parent txn {:?}", env.root);

        info!("started parent txn");
        std::thread::sleep(std::time::Duration::from_secs(3));

        info!("started parent txn {:?}", txn.root);
        // The child didn't commit yet.
        assert_eq!(txn.root, 1);
        std::thread::sleep(std::time::Duration::from_millis(300));

        unsafe { libc::wait(&mut child) };

        std::thread::sleep(std::time::Duration::from_millis(100));
        let txn = Env::txn_begin(&env).unwrap();
        info!("started parent txn {:?}", txn.root);
        // The parent committed, this is a new transaction.
        assert_eq!(txn.root, 0);
        let mut status = 1;
        unsafe { libc::wait(&mut status) };
        assert_eq!(status, 0);

}
#[test]
fn more_than_two_versions() {
    env_logger::try_init().unwrap_or(());
    let n = 5;
    let env = Env::new_anon(40960, n).unwrap();
    let mut txn = Env::mut_txn_begin(&env).unwrap();
    // Allocate two pages.
    for i in 0..n {
        let page = txn.alloc_page().unwrap();
        debug!("page = {:?}", page);
        txn.set_root(i, page.0.offset);
    }
    txn.commit().unwrap();
    for i in 0..n {
        let mut txn = Env::mut_txn_begin(&env).unwrap();
        // Free one of the pages.
        debug!("root(0) = {:?}", txn.root(i));
        txn.decr_rc(txn.root(i).unwrap()).unwrap();
        txn.remove_root(i);
        txn.commit().unwrap();
    }
    let mut txn = Env::mut_txn_begin(&env).unwrap();
    unsafe {
        let p = &*(env.mmaps.lock()[0].ptr.add(txn.root * PAGE_SIZE) as *const GlobalHeader);
        debug!("free page: 0x{:x}", u64::from_le(p.free_db));
        let db: Db<u64, ()> = Db {
            db: u64::from_le(p.free_db),
            k: std::marker::PhantomData,
            v: std::marker::PhantomData,
            p: std::marker::PhantomData,
        };
        for x in iter(&txn, &db, None).unwrap() {
            debug!("0x{:x}", x.unwrap().0);
        }
    }
    let page = txn.alloc_page().unwrap();
    debug!("page = {:?}", page);

        btree::get(&txn, &db, &i, None).unwrap();

        get(&txn, &db, &i, None).unwrap();

        btree::get(&txn, &db, &i, None).unwrap();

        get(&txn, &db, &i, None).unwrap();

            assert_eq!(btree::get(&txn, &db, &k, None).unwrap(), Some((k, v)))

            assert_eq!(get(&txn, &db, &k, None).unwrap(), Some((k, v)))

            assert_eq!(btree::get(&txn, &db, &k, None).unwrap(), Some((k, v)))

            assert_eq!(get(&txn, &db, &k, None).unwrap(), Some((k, v)))

    P: btree::BTreePage<K, V>,

    P: BTreePage<K, V>,

                btree::get(&txn, &db, k.as_bytes(), None).unwrap(),

                get(&txn, &db, k.as_bytes(), None).unwrap(),

                let k = lmdb_rs::MDB_val {

                let k = MDB_val {

            let k = lmdb_rs::MDB_val {

            let k = MDB_val {

    for (kv, n) in btree::rev_iter(&txn, &db, Some((&i0, None)))

    for (kv, n) in rev_iter(&txn, &db, Some((&i0, None)))

    for (kv, n) in btree::iter(&txn, &db, Some((&i0, None))).unwrap().zip(i0..) {

    for (kv, n) in iter(&txn, &db, Some((&i0, None))).unwrap().zip(i0..) {

    let mut it = btree::rev_iter(&txn, &db, Some((&100, None))).unwrap();

    let mut it = rev_iter(&txn, &db, Some((&100, None))).unwrap();

    let n = 6 as u64;

    let n = 6u64;

    /*

    // add_refs(&txn, &db3, &mut refs).unwrap();

    */

    let n = 1000 as u64;
    let i0 = 10 as u64;

    let n = 1000u64;
    let i0 = 10u64;

#![deny(
    missing_docs,
    trivial_casts,
    trivial_numeric_casts,
    unused_import_braces,
    unused_qualifications
)]

//!
//! The binary format of a Sanakirja database is the following:
//!
//! - There is a fixed number of "current versions", set at file
//! initialisation. If a file has n versions, then for all k between 0
//! and n-1 (included), the k^th page (i.e. the byte positions between
//! `k * 4096` and `(k+1) * 4096`, also written as `k << 12` and
//! `(k+1) << 12`) stores the data relative to that version, and is
//! called the "root page" of that version.
//!
//!   This is a way to handle concurrent access: indeed, mutable
//! transactions do not exclude readers, but readers that started
//! before the commit of a mutable transaction will keep reading the
//! database as it was before the commit. However, this means that
//! older versions of the database have to be kept "alive", and the
//! "number of current versions" here is the limit on the number of
//! versions that can be kept "alive" at the same time.
//!
//!   When a reader starts, it takes a shared file lock on the file
//! representing the youngest committed version. When a writer starts,
//! it takes an exclusive file lock on the file representing the
//! oldest committed version. This implies that if readers are still
//! reading that version, the writer will wait for the exclusive lock.
//!
//!   After taking a lock, the writer (also called "mutable
//! transaction" or [`MutTxn`]) copies the entire root page of the
//! youngest committed version onto the root page of the oldest
//! committed version, hence erasing the root page of the oldest
//! version.
//!
//! - Root pages have the following format: a 32-bytes header
//! (described below), followed by 4064 bytes, usable in a more or
//! less free format. The current implementation defines two methods
//! on [`MutTxn`], [`MutTxn::set_root`] and [`MutTxn::remove_root`],
//! treating that space as an array of type `[u64; 510]`. A reasonable
//! use for these is to point to different datastructures allocated in
//! the file, such as the offsets in the file to the root pages of B
//! trees.
//!
//!   Now, about the header, there's a version identifier on the first
//! 16 bytes, followed by two bytes: `root` is the version used by the
//! current mutable transaction (if there is current mutable
//! transaction), or by the next mutable transaction (else). The
//! `n_roots` field is the total number of versions.
//!
//!   ```
//!   #[repr(C)]
//!   pub struct GlobalHeader {
//!       /// Version of Sanakirja
//!       pub version: u16,
//!       /// Which page is currently the root page? (only valid for page 0).
//!       pub root: u8,
//!       /// Total number of versions (or "root pages")
//!       pub n_roots: u8,
//!       /// CRC of this page.
//!       pub crc: u32,
//!       /// First free page at the end of the file (only valid for page 0).
//!       pub length: u64,
//!       /// Offset of the free list.
//!       pub free_db: u64,
//!       /// Offset of the RC database.
//!       pub rc_db: u64,
//!   }
//!   ```

pub use environment::{Env, MutTxn, Txn};
pub use sanakirja_core::{btree, direct_repr, CowPage, LoadPage, MutPage, Page, Storable};

pub use environment::{Commit, Env, MutTxn, RootDb, Txn};
pub use sanakirja_core::{btree, direct_repr, LoadPage, Storable, UnsizedStorable};

    /// The root page of this transaction, which is 1 + the root
    /// written on page 0. The root written on page 0 changes at
    /// commit time.
    pub(crate) root: usize,

    /// Offset to the root of the B tree of free pages.

    /// Reference counts use a strange encoding, meant to avoid code
    /// bloat: indeed, the list of free pages uses `Db<u64, ()>`, so
    /// we're just reusing the same code here, encoding the reference
    /// counts in the 12 least significant bits of the keys, and the
    /// actual pages in the 52 most significant bits.

    ///
    /// Since we can't reuse them in the same transaction, another
    /// option would be to put them directly into the table of free
    /// pages. However, since calls to `put` may allocate and free
    /// pages, this could recurse infinitely, which is why we store
    /// them outside of the file.

/// When dropping a transaction, we need to unlock the read-write
/// locks internal to this process, and possibly the file locks.

            let root = env.root.lock();
            debug!("unlock exclusive {:?}", *root);
            env.roots[*root].rw.unlock_exclusive();
            if let Some(ref f) = env.roots[*root].lock_file {
                f.unlock().unwrap_or(())
            }

            env.mut_txn_unlock().unwrap_or(());
            env.roots[self.root].rw.unlock_exclusive();
            env.unlock(self.root).unwrap_or(())

/// Transactions that can be committed.

/// Transactions that can be committed. This trait is an abstraction
/// over mutable transactions and their subtransactions.

/// The following is very easy, we're just extending all values of the
/// current transaction with values of the subtransaction.

    #[cfg(feature = "mmap")]
    fn mut_txn_lock(&self) -> Result<(), Error> {
        self.mut_txn_lock.lock();
        if let Some(ref f) = self.file {
            f.lock_exclusive()?;
        }
        Ok(())
    }
    #[cfg(not(feature = "mmap"))]
    fn mut_txn_lock(&self) -> Result<(), Error> {
        self.mut_txn_lock.lock();
        Ok(())
    }
    #[cfg(feature = "mmap")]
    fn mut_txn_unlock(&self) -> Result<(), Error> {
        unsafe {
            self.mut_txn_lock.unlock();
        }
        if let Some(ref f) = self.file {
            f.unlock()?
        }
        Ok(())
    }
    #[cfg(not(feature = "mmap"))]
    fn mut_txn_unlock(&self) -> Result<(), Error> {
        unsafe {
            self.mut_txn_lock.unlock();
        }
        Ok(())
    }

            let (header, free, rc) = {
                let env_ = env.borrow();
                let maps = env_.mmaps.lock()[0].ptr;
                let v = env_.root.lock();
                let n = env_.roots.len();
                env_.roots[*v].rw.lock_exclusive();
                if let Some(ref f) = env_.roots[*v].lock_file {
                    f.lock_exclusive()?
                }
                // Root of the last MutTxn.
                let v0 = (*v + n - 1) % env_.roots.len();
                debug!("v = {:?} v0 = {:?}", v, v0);
                // Copy the roots of the last transaction onto this one.
                let page_ptr = maps.offset((v0 * PAGE_SIZE) as isize);
                let next_page_ptr = maps.offset((*v * PAGE_SIZE) as isize);
                std::ptr::copy_nonoverlapping(page_ptr.add(8), next_page_ptr.add(8), PAGE_SIZE - 8);

            let env_ = env.borrow();

                let header = GlobalHeader::from_le(&*(page_ptr as *const GlobalHeader));
                env_.check_crc(v0)?;

            // First, take an exclusive file lock on the whole file to
            // make sure that no other process is starting a mutable
            // transaction at the same time. The worst that can happen
            // here is if the other process commits while we're still
            // waiting for a lock on the current page, because if that
            // happens, this new transaction will erase the
            // transaction in the other process.
            env_.mut_txn_lock()?;
            // Then, we can lock the root page of this transaction.
            let maps = env_.mmaps.lock()[0].ptr;
            let root = (&*(maps as *const GlobalHeader)).root as usize;
            debug!("BEGIN_TXN root = {:?}", root);
            env_.roots[root].rw.lock_exclusive();
            env_.lock_exclusive(root)?;
            // Root of the last MutTxn.
            let v0 = (root + env_.roots.len() - 1) % env_.roots.len();
            env_.check_crc(v0)?;
            // Copy the root page of the last transaction onto this
            // one.
            let page_ptr = maps.offset((v0 * PAGE_SIZE) as isize);
            let next_page_ptr = maps.offset((root * PAGE_SIZE) as isize);
            std::ptr::copy_nonoverlapping(page_ptr.add(8), next_page_ptr.add(8), PAGE_SIZE - 8);

                let free = header.free_db;
                let rc = if header.rc_db == 0 {
                    None
                } else {
                    Some(btree::Db {
                        db: header.rc_db,
                        k: std::marker::PhantomData,
                        v: std::marker::PhantomData,
                        p: std::marker::PhantomData,
                    })
                };
                (header, free, rc)
            };
            let length = if header.length == 0 {
                (PAGE_SIZE as u64) * (header.n_roots as u64)
            } else {
                header.length
            };

            // Finally, read the header and start the transaction.
            let header = GlobalHeader::from_le(&*(next_page_ptr as *const GlobalHeader));
            debug!("n_roots = {:?}", header.n_roots);

                root,

                rc,
                length,
                free,

                rc: if header.rc_db == 0 {
                    None
                } else {
                    Some(btree::Db::from_page(header.rc_db))
                },
                length: if header.length == 0 {
                    (PAGE_SIZE as u64) * (header.n_roots as u64)
                } else {
                    header.length
                },
                free: header.free_db,

        unsafe {
            let mut free_db: btree::Db<u64, ()> = btree::Db::from_page(self.free);
            while !self.free_owned_pages.is_empty() || !self.free_pages.is_empty() {
                let mut free_owned_pages =
                    std::mem::replace(&mut self.free_owned_pages, Vec::new());
                let mut free_pages = std::mem::replace(&mut self.free_pages, Vec::new());
                for p in free_owned_pages.drain(..).chain(free_pages.drain(..)) {
                    btree::put(&mut self, &mut free_db, &p, &())?;

        debug!("COMMIT");
        // If there's no tree of free pages, and no pages to free,
        // don't bother with free pages at all (don't even allocate a
        // tree).
        let free_db =
            if self.free == 0 && self.free_owned_pages.is_empty() && self.free_pages.is_empty() {
                None
            } else {
                // Else, allocate or load the tree of free pages.
                let mut free_db: btree::Db<u64, ()> = if self.free == 0 {
                    btree::create_db(&mut self)?
                } else {
                    btree::Db::from_page(self.free)
                };
                debug!("free_db = {:?}", free_db);
                // Adding all the pages freed during the transaction to the
                // tree of free pages. If this call to `btree::put` frees
                // pages, add them again. This converges in at most log n
                // iterations (where n is the total number of free pages).
                // First, set the free table to 0 in this transaction, to
                // avoid recursing in the calls to `put` below (indeed,
                // the table of free pages is used when allocating new
                // pages, which may happen in a call to `put`).
                self.free = 0;
                // Then, while there are pages to free, free them. Since
                // the calls to `put` below might free pages (and add
                // pages to these two vectors), the (outer) loop might run
                // for more than one iteration.
                while !self.free_pages.is_empty() || !self.free_owned_pages.is_empty() {
                    while let Some(p) = self.free_pages.pop() {
                        let p = p & !0xfff;
                        btree::put(&mut self, &mut free_db, &p.to_le(), &())?;
                    }
                    while let Some(p) = self.free_owned_pages.pop() {
                        let p = p & !0xfff;
                        btree::put(&mut self, &mut free_db, &p.to_le(), &())?;
                    }

            }
            for p in self.occupied_owned_pages.iter_mut() {
                clear_dirty(p);

                Some(free_db)
            };
        // Clear the dirty bit of all pages we've touched. If they've
        // been freed and have already been flushed by the kernel, we
        // don't want to resurrect them to the main memory, so we
        // check that.
        let mut occ = std::mem::replace(&mut self.occupied_owned_pages, Vec::new());
        for p in occ.iter_mut() {
            if let Some(ref free_db) = free_db {
                if let Some((pp, ())) = btree::get(&self, free_db, &p.0.offset, None)? {
                    if *pp == p.0.offset {
                        continue;
                    }
                }

            let env = self.env.borrow();
            let mut maps = env.mmaps.lock();
            let mut root = env.root.lock();
            let globptr = maps[0].ptr.add(*root * PAGE_SIZE) as *mut GlobalHeader;
            (&mut *globptr).length = self.length.to_le();
            (&mut *globptr).free_db = free_db.db.to_le();
            self.free = free_db.db;

            clear_dirty(p);
        }

            let root_dbs = std::slice::from_raw_parts_mut(
                maps[0].ptr.add(*root * PAGE_SIZE + GLOBAL_HEADER_SIZE) as *mut u64,

        let env = self.env.borrow();
        let mut maps = env.mmaps.lock();
        // Get this transaction's root page.
        let globptr =
            unsafe { &mut *(maps[0].ptr.add(self.root * PAGE_SIZE) as *mut GlobalHeader) };
        // Set the length and free database.
        globptr.length = self.length.to_le();
        if let Some(free_db) = free_db {
            debug!("COMMIT: free_db = 0x{:x}", free_db.db);
            globptr.free_db = free_db.db.to_le();
        }
        // Set the "root databases" modified by this transaction.
        let root_dbs = unsafe {
            std::slice::from_raw_parts_mut(
                maps[0].ptr.add(self.root * PAGE_SIZE + GLOBAL_HEADER_SIZE) as *mut u64,

            );
            for (&r, rr) in self.roots.iter().zip(root_dbs.iter_mut()) {
                debug!("root_db: {:?}", rr as *mut u64);
                debug!("committing root: {:?} {:?}", r, rr);
                if r > 0 {
                    *rr = r
                }

            )
        };
        for (&r, rr) in self.roots.iter().zip(root_dbs.iter_mut()) {
            debug!("root_db: {:?}", rr);
            debug!("committing root: {:?} {:?}", r, rr);
            if r > 0 {
                *rr = r

        }

            set_crc(maps[0].ptr.add(*root * PAGE_SIZE));

        // Set the root page's CRC.
        unsafe {
            set_crc(maps[0].ptr.add(self.root * PAGE_SIZE));
        }

            // Moving the root page, both on page 0, and on the environment.
            (&mut *(maps[0].ptr as *mut GlobalHeader)).root = *root as u8;

        // Move the current global root page by one page on page 0.
        unsafe {
            (&mut *(maps[0].ptr as *mut GlobalHeader)).root =
                (self.root as u8 + 1) % (env.roots.len() as u8);
        }

            for m in maps.iter_mut() {
                m.flush()?
            }
            debug!("commit: unlock {:?}", *root);
            env.roots[*root].rw.unlock_exclusive();
            if let Some(ref f) = env.roots[*root].lock_file {
                debug!("commit: unlock file");
                f.unlock().unwrap_or(())
            }
            *env.first_unused_page.lock() = self.length;
            *root = (*root + 1) % env.roots.len();
            Ok(())

        // Flush all the maps.
        for m in maps.iter_mut() {
            m.flush()?

        // And finally, unlock the root page in the environment.
        debug!("commit: unlock {:?}", self.root);
        unsafe { env.roots[self.root].rw.unlock_exclusive() };
        // Unlock the root page on the file lock (if relevant).
        env.unlock(self.root)?;
        // And unlock the global mutable transaction mutex.
        env.mut_txn_unlock()?;
        debug!("/COMMIT");
        Ok(())

    /// Setting the `num`th element of the initial page, treated as a
    /// `[u64; 510]`, to `value`. This doesn't actually write anything
    /// to that page, since that page is written during the commit.
    ///
    /// In the current implementation, `value` is probably going to be
    /// the offset in the file of the root page of a B tree.

    /// Setting the `num`th element of the initial page, treated as a
    /// [u64; 510].

        if self.roots.get(num).is_none() {
            self.roots.resize(num + 1, 0u64);
        }

    /// Add the page at offset `offset` to the list of free pages that
    /// were allocated by this `MutTxn` (and hence can be reallocated
    /// by the same transaction).

    /// Add the page at offset `offset` to the list of free pages
    /// allocated by a previous transaction, and hence may still be
    /// accessible by other transactions.

    /// Pop a free page from the list of free pages.
    fn free_pages_pop(&mut self) -> Result<Option<u64>, crate::Error> {

    /// Pop a free page from the B tree of free pages.
    fn free_pages_pop(&mut self) -> Result<Option<u64>, Error> {

            debug!("self.free = 0");

        let mut db: btree::Db<u64, ()> = btree::Db {
            db: self.free,
            k: std::marker::PhantomData,
            v: std::marker::PhantomData,
            p: std::marker::PhantomData,
        };

        let mut db: btree::Db<u64, ()> = btree::Db::from_page(self.free);

        let f = if let Some((f, ())) = curs.set_last(self)? {

        let mut f = if let Some((f, ())) = curs.set_last(self)? {

        // Get the last page that is also free in all other versions.
        loop {
            debug!("trying 0x{:x}", f);
            if self.free_for_all(f)? {
                break;
            } else if let Some((f_, ())) = curs.prev(self)? {
                f = *f_
            } else {
                debug!("no more candidate");
                return Ok(None);
            }
        }
        self.free = 0;

    // Check whether this page is also free for the other
    // versions.
    fn free_for_all(&self, f: u64) -> Result<bool, Error> {
        let env = self.env.borrow();
        // We already know it's free for the youngest previous
        // transaction and for the current one (because the tree of
        // free pages was copied from there), so we only have
        // `self.roots.len() - 2` root pages to check.
        for i in 1..env.roots.len() - 1 {
            let db: btree::Db<u64, ()> = unsafe {
                let p = &*(env.mmaps.lock()[0]
                    .ptr
                    .add(((self.root + i) % env.roots.len()) * PAGE_SIZE)
                    as *const GlobalHeader);
                if f >= u64::from_le(p.length) {
                    // Page `f` was allocated strictyl after
                    // transaction `i`.
                    continue;
                }
                if p.free_db == 0 {
                    // This version doesn't have any free page.
                    return Ok(false);
                }
                btree::Db::from_page(p.free_db)
            };
            if let Some((&f_, ())) = btree::get(self, &db, &f, None)? {
                if f_ != f {
                    return Ok(false);
                }
            }
        }
        Ok(true)
    }

        // If we have allocated and freed a page in this transaction, use it first.

        // If we have allocated and freed a page in this transaction,
        // use it first.

            debug!("free owned pop {:?}", offset);

            debug!("free owned pop 0x{:x}", offset);

                debug!("free pages pop {:?}", offset);

                debug!("free pages pop 0x{:x}", offset);

                let page = MutPage(CowPage { data, offset });

                    .push(MutPage(CowPage { offset, data }));
                Ok(page)

                    .push(MutPage(CowPage { data, offset }));
                Ok(MutPage(CowPage { data, offset }))

                debug!("allocate in the free space {:?}", self.length);

                debug!("allocate in the free space 0x{:x}", self.length);

                let page = MutPage(CowPage { data, offset });

                Ok(page)

                Ok(MutPage(CowPage { data, offset }))

    fn decr_rc(&mut self, off: u64) -> Result<usize, Error> {
        debug!("decr_rc {:?} {:?}", off, self.rc);

    /// Increment the reference count for page `off`.
    fn incr_rc(&mut self, off: u64) -> Result<usize, Error> {

            curs.set(self, &off, None)?;
            let rc = if let Some((rc, ())) = curs.next(self)? {

            let rc = if let Some((rc, _)) = curs.set(self, &off, None)? {

                    *rc

                    *rc & 0xfff

            debug!("decr_rc, rc = {:?}", rc);

                btree::del(self, &mut rc_, &rc, None)?;
                if rc & 0xfff > 2 {
                    btree::put(self, &mut rc_, &(rc - 1), &())?;
                    self.rc = Some(rc_);
                } else {
                    // Implicit "1".
                    self.rc = Some(rc_)
                }
                return Ok((rc & 0xfff) as usize - 1);
            } else {
                self.rc = Some(rc_)

                btree::del::del_at_cursor(self, &mut rc_, &mut curs)?;

            assert!(rc + 1 <= 0xfff);
            btree::put(self, &mut rc_, &(off | (rc + 1)), &())?;
            self.rc = Some(rc_);
            Ok(rc as usize + 1)
        } else {
            let mut rc = btree::create_db(self)?;
            btree::put(self, &mut rc, &(off | 2), &())?;
            self.rc = Some(rc);
            Ok(2)

        self.free_page(off);
        Ok(0)

    fn decr_rc(&mut self, off: u64) -> Result<usize, Error> {
        let rc = self.decr_rc_(off)?;
        if rc == 0 {
            self.free_page(off);
        }
        Ok(rc)
    }

        debug!("decr_rc_owned {:?} {:?}", off, self.rc);

        let rc = self.decr_rc_(off)?;
        if rc == 0 {
            self.free_owned_page(off);
        }
        Ok(rc)
    }
}
impl<E: Borrow<Env>, A> MutTxn<E, A> {
    /// Decrement the reference count of page `off`, freeing that page
    /// if the RC reaches 0 after decrementing it.
    fn decr_rc_(&mut self, off: u64) -> Result<usize, Error> {
        debug!("decr_rc 0x{:x} {:?}", off, self.rc);
        // If there's no RC table, free the page. Also, in order to
        // avoid infinite recursion (since `del` and `put` below might
        // free pages), we `take` the reference counter table.

            // The reference count is stored as the 12 LSBs of the
            // keys. If the page isn't in the RC table, the count is
            // 1.

            debug!("rc = {:?}", rc);

            debug!("decr_rc, rc = 0x{:x}", rc);

                debug!("del");

                // If the reference count is strictly more than 2,
                // replace the reference count with a decremented
                // value.

                debug!("/del");

                    // Implicit "1".
                    self.rc = Some(rc_);

                    // Else, we don't free the page, but don't add the
                    // page back, since this is an implicit value of
                    // "1" for the reference count.
                    self.rc = Some(rc_)

                self.rc = Some(rc_);

                self.rc = Some(rc_)

        self.free_owned_page(off);

    }
    fn incr_rc(&mut self, off: u64) -> Result<usize, Error> {
        debug!("incr_rc {:?}", off);
        if let Some(mut rc_) = self.rc.take() {
            let mut curs = btree::cursor::Cursor::new(self, &rc_)?;
            let rc = if let Some((rc, _)) = curs.set(self, &off, None)? {
                if *rc & !0xfff == off {
                    *rc & 0xfff
                } else {
                    1
                }
            } else {
                1
            };
            if rc > 1 {
                btree::del::del_at_cursor(self, &mut rc_, &mut curs)?;
            }
            assert!(rc + 1 <= 0xfff);
            btree::put(self, &mut rc_, &(off | (rc + 1)), &())?;
            self.rc = Some(rc_);
            Ok(rc as usize + 1)
        } else {
            let mut rc = btree::create_db(self)?;
            debug!("put in RC: {:?} {:?}", rc, off | 2);
            btree::put(self, &mut rc, &(off | 2), &())?;
            debug!("done");
            self.rc = Some(rc);
            Ok(2)
        }

    type Error = crate::Error;

    type Error = Error;

impl<E: Borrow<Env>, T> RootDb for MutTxn<E, T> {
    fn root_db<K: Storable + ?Sized, V: Storable + ?Sized, P: crate::btree::BTreePage<K, V>>(
        &self,
        n: usize,
    ) -> Option<sanakirja_core::btree::Db_<K, V, P>> {

impl<E: Borrow<Env>, T> MutTxn<E, T> {
    /// Low-level method to get the root page number `n`, if that page
    /// isn't a B tree (use the [`RootDb`] trait else).
    pub fn root(&self, n: usize) -> Option<u64> {

            Some(sanakirja_core::btree::Db_ {
                db: *db,
                k: std::marker::PhantomData,
                v: std::marker::PhantomData,
                p: std::marker::PhantomData,
            })

            Some(*db)

                    let root = env.root.lock();

                        .add(*root * PAGE_SIZE + GLOBAL_HEADER_SIZE + 8 * n)

                        .add(self.root * PAGE_SIZE + GLOBAL_HEADER_SIZE + 8 * n)

                    Some(sanakirja_core::btree::Db_ {
                        db,
                        k: std::marker::PhantomData,
                        v: std::marker::PhantomData,
                        p: std::marker::PhantomData,
                    })

                    Some(db)

        }
    }
}
impl<E: Borrow<Env>, T> RootDb for MutTxn<E, T> {
    // Just call method `root` and convert the result to a `Db`.
    fn root_db<K: Storable + ?Sized, V: Storable + ?Sized, P: crate::btree::BTreePage<K, V>>(
        &self,
        n: usize,
    ) -> Option<sanakirja_core::btree::Db_<K, V, P>> {
        if let Some(db) = self.root(n) {
            Some(sanakirja_core::btree::Db_::from_page(db))
        } else {
            None

use parking_lot::lock_api::RawRwLock;

use parking_lot::lock_api::{RawMutex, RawRwLock};

use global_header::*;

pub(crate) use global_header::*;

/// A chunk of memory, possibly of a memory-mapped file, or allocated
/// with `std::alloc`.

/// Environment, required to start any transactions. Thread-safe, but
/// opening the same database several times in the same process is not
/// cross-platform.

/// An environment, which may be either a memory-mapped file, or
/// memory allocated with [`std::alloc`].

    first_unused_page: Mutex<u64>,

    mut_txn_lock: parking_lot::RawMutex,

    roots: Vec<RootLock>,
    // Root number of the next MutTxn.
    pub(crate) root: Mutex<usize>,
}
struct RootLock {
    /// It is undefined behavior to have a file mmapped for than once.
    #[cfg(feature = "mmap")]
    lock_file: Option<std::fs::File>,
    rw: parking_lot::RawRwLock,
    n_txn: AtomicUsize,

    pub(crate) roots: Vec<RootLock>,

#[cfg(feature = "mmap")]
impl Drop for Env {
    fn drop(&mut self) {
        for map in self.mmaps.lock().drain(..) {
            drop(map.mmap);
        }
    }
}

/// An immutable transaction.
pub struct Txn<E: Borrow<Env>> {
    env: E,
    root: usize,
}
#[cfg(feature = "mmap")]
#[test]
#[should_panic]
fn nroots_test() {
    let path = tempfile::tempdir().unwrap();
    let path = path.path().join("db");
    let l0 = 1 << 15; // 8 pages
    Env::new(&path, l0, 19).unwrap();
}
#[cfg(feature = "mmap")]
#[test]
fn mmap_growth_test() {
    let path = tempfile::tempdir().unwrap();
    let path = path.path().join("db");
    let l0 = 1 << 15; // 8 pages
    {
        let env = Env::new(&path, l0, 2).unwrap();
        let map1 = env.open_mmap(0, l0).unwrap();
        println!("{:?}", map1);
        let map2 = env.open_mmap(1, l0).unwrap();
        println!("{:?}", map2);
        map1.flush().unwrap();
        map2.flush().unwrap();
    }
    let len = std::fs::metadata(&path).unwrap().len();
    assert_eq!(len, (l0 << 2) - l0);
}

/// A lock on a root page for this process only, because taking
/// multiple locks on the same file from a single process isn't
/// cross-platform (or even properly defined).
///
/// Usage is as follows:
///
/// - For read-only transactions, we first take a read lock on the `rw`
/// field, and increment `n_txn`, locking the file if the former value
/// is 0.
///
/// - For read-write transactions, we first take a write lock on the
/// `rw` field, and then take an exclusive lock on the file (this is
/// valid since only one read-write transaction can be active in a
/// process at the same time).
///
pub(crate) struct RootLock {
    /// It is undefined behavior to have a file mmapped for than once.
    #[cfg(feature = "mmap")]
    lock_file: Option<std::fs::File>,

#[cfg(not(feature = "crc32"))]
fn set_crc(_ptr: *mut u8) {}

    /// Read-write lock.
    rw: parking_lot::RawRwLock,

#[cfg(feature = "crc32")]
fn set_crc(ptr: *mut u8) {
    unsafe {
        let root_page = std::slice::from_raw_parts(ptr.add(8), PAGE_SIZE - 8);
        let mut h = HASHER.clone();
        h.update(root_page);
        let globptr = ptr as *mut GlobalHeader;
        (&mut *globptr).crc = h.finalize().to_le();
    }

    /// Count of read-only transactions.
    n_txn: AtomicUsize,

    /// Same as [`new`](#new), but does not take a lock on the file
    /// system.

    /// Same as [`new`](#new), but does not create any lock on the
    /// file system.

    /// This method is provided because waiting for a lock on the file
    /// system may block the whole process, whereas.

    /// The database is very likely to get corrupted if an environment
    /// is opened from multiple processes, or more than once by the
    /// same process, if at least one of these instances can start a
    /// mutable transaction.

    /// However, the database is very likely to get corrupted if open
    /// more than once at the same time, even within the same process.
    ///
    /// Therefore, do not use this method without another locking
    /// mechanism in place to avoid that situation.

    /// The `n_roots` argument is ignored if the database already
    /// exists, and is used to initialise the first `n_roots` pages of
    /// the file else.

        let db_exists = std::fs::metadata(&path).is_ok();
        let length = if let Ok(meta) = std::fs::metadata(&path) {

        let meta = std::fs::metadata(&path);
        let length = if let Ok(ref meta) = meta {

        // Find the next multiple of PAGE_SIZE greater than or equal
        // to `length`.

        Self::new_nolock_mmap(Some(file), length, mmap, !db_exists, n_roots)

        Self::new_nolock_(Some(file), length, mmap, meta.is_err(), n_roots)

    /// Create an environment from a file, without creating any lock
    /// on the filesystem.

    unsafe fn new_nolock_mmap(

    unsafe fn new_nolock_(

        assert!(n_roots < ((length >> 12) as usize));

        assert!(n_roots >= 1);
        assert!(n_roots <= ((length >> 12) as usize));
        assert!(n_roots < 256);

        if initialise {
            for i in 0..n_roots {
                *(map.add(i * PAGE_SIZE) as *mut GlobalHeader) = (GlobalHeader {
                    version: CURRENT_VERSION,
                    root: 0,
                    n_roots: n_roots as u8,
                    crc: 0,
                    length: (n_roots * PAGE_SIZE) as u64,
                    free_db: 0,
                    rc_db: 0,
                })
                .to_le();

        let n_roots = if initialise {
            // Initialise the first `n_roots` pages at the start of
            // the file.
            init(map, n_roots);
            // Since the first `n_roots` pages are occupied by roots,
            // the first unused page is found at offset `n_roots *
            // PAGE_SIZE`.
            n_roots
        } else {
            // Read the root and number of roots from the first page's
            // header.
            GlobalHeader::from_le(&*(map as *const GlobalHeader)).n_roots as usize
        };

                set_crc(map.add(i * PAGE_SIZE));
            }
        }
        let glob = GlobalHeader::from_le(&*(map as *const GlobalHeader));
        let mut roots = Vec::with_capacity(glob.n_roots as usize);
        for _ in 0..glob.n_roots {
            roots.push(RootLock {
                rw: <parking_lot::RawRwLock as parking_lot::lock_api::RawRwLock>::INIT,
                n_txn: AtomicUsize::new(0),
                lock_file: None,
            })
        }

        // Finally, create the environment.

            first_unused_page: Mutex::new(2),
            roots,
            root: Mutex::new(glob.root as usize),

            mut_txn_lock: RawMutex::INIT,
            // Initialise a different `RootLock` for each root page.
            roots: (0..n_roots)
                .map(|_| RootLock {
                    rw: RawRwLock::INIT,
                    n_txn: AtomicUsize::new(0),
                    lock_file: None,
                })
                .collect(),

    /// No-mmap version of the same thing.

    unsafe fn new_nolock_mmap(length: u64, initialise: bool) -> Result<Env, Error> {

    unsafe fn new_nolock_(length: u64, initialise: bool, n_roots: usize) -> Result<Env, Error> {
        assert!(n_roots >= 1);
        assert!(n_roots <= ((length >> 12) as usize));
        assert!(n_roots < 256);

        let n_roots = 2;
        for i in 0..n_roots {
            *(map.offset((i * PAGE_SIZE) as isize) as *mut GlobalHeader) = (GlobalHeader {
                version: CURRENT_VERSION,
                root: 0,
                n_roots: n_roots as u8,
                crc: 0,
                length: n_roots as u64 * PAGE_SIZE as u64,
                free_db: 0,
                rc_db: 0,
            })
            .to_le();
        }
        let glob = GlobalHeader::from_le(&*(map as *const GlobalHeader));
        let mut roots = Vec::with_capacity(glob.n_roots as usize);
        for _ in 0..glob.n_roots {
            roots.push(RootLock {
                rw: <parking_lot::RawRwLock as parking_lot::lock_api::RawRwLock>::INIT,
                n_txn: AtomicUsize::new(0),
            })
        }

        init(map, n_roots);

            first_unused_page: Mutex::new(2),
            roots,
            root: Mutex::new(glob.root as usize),

            mut_txn_lock: RawMutex::INIT,
            // Initialise a different `RootLock` for each root page.
            roots: (0..n_roots)
                .map(|_| RootLock {
                    rw: RawRwLock::INIT,
                    n_txn: AtomicUsize::new(0),
                })
                .collect(),

    }
}
unsafe fn init(map: *mut u8, n_roots: usize) {
    for i in 0..n_roots {
        *(map.offset((i * PAGE_SIZE) as isize) as *mut GlobalHeader) = (GlobalHeader {
            version: CURRENT_VERSION,
            root: 0,
            n_roots: n_roots as u8,
            crc: 0,
            length: n_roots as u64 * PAGE_SIZE as u64,
            free_db: 0,
            rc_db: 0,
        })
        .to_le();
        set_crc(map.add(i * PAGE_SIZE));

}

impl Env {

        unsafe { Self::new_nolock_mmap(None, length, mmap, true, n_roots) }

        unsafe { Self::new_nolock_(None, length, mmap, true, n_roots) }

    pub fn new_anon(length: u64) -> Result<Env, Error> {

    pub fn new_anon(length: u64, n_roots: usize) -> Result<Env, Error> {

        unsafe { Self::new_nolock_mmap(length, true) }

        unsafe { Self::new_nolock_(length, true, n_roots) }

    /// Now, here is how databases grow: when we run out of space, we
    /// allocate a new chunk of memory/disk space, of a size twice as
    /// large as the last chunk. The size of the first chunk is the
    /// size of the file when we first opened the environment.

    /// Allocate the memory of the appropriate size and offest for the
    /// `i`th chunk.

    /// The same, but for memory-mapped file. If we're doing that, it
    /// means we need to grow the file.

    /// Find an offset in a file, possibly mapping more of the file if
    /// necessary (for example if the file has been grown by another
    /// process since we openend this environment).

                return mmaps[i].ptr.offset(offset as isize);

                return mmaps[i].ptr.add(offset as usize);

        }
    }
    /// Close this repository, releasing the locks. It is undefined
    /// behaviour to use the environment afterwards. This method can
    /// be used for instance to release the locks before allocating a
    /// new environment (note that `std::mem::replace` followed by
    /// `Drop::drop` of the previous value would not release the locks
    /// in the correct order).
    ///
    /// The safe alternative to this method is to use an `Option<Env>`
    /// instead of an `Env`.
    #[cfg(feature = "mmap")]
    pub unsafe fn close(&mut self) {
        for m in self.mmaps.lock().drain(..) {
            drop(m.mmap);

        for l in self.roots.drain(..) {
            if let Some(ref f) = l.lock_file {
                f.unlock().unwrap_or(())
            }
        }

        }
    }
    /// If the CRC feature is disabled, we're not checking CRCs.
    #[cfg(not(feature = "crc32"))]
    fn check_crc(&self, _root: usize) -> Result<(), crate::CRCError> {
        Ok(())
    }
    /// Else, we are checking CRCs, so return a CRC error if the check
    /// fails (the CRC is a 32 bit integer encoded little-endian at
    /// bytes [8,12[ of the root pages).
    #[cfg(feature = "crc32")]
    fn check_crc(&self, root: usize) -> Result<(), crate::CRCError> {
        unsafe {
            let maps = self.mmaps.lock();
            let page_ptr = maps[0].ptr.add(root * PAGE_SIZE);
            let crc = (&*(page_ptr as *const GlobalHeader)).crc;
            let root_page = std::slice::from_raw_parts(page_ptr.add(8), PAGE_SIZE - 8);
            let mut h = HASHER.clone();
            h.update(root_page);
            if h.finalize() != crc {
                return Err(crate::CRCError {});
            }

        Ok(())

}
#[cfg(feature = "mmap")]
#[test]
#[should_panic]
fn nroots_test() {
    let path = tempfile::tempdir().unwrap();
    let path = path.path().join("db");
    let l0 = 1 << 15; // 8 pages
    Env::new(&path, l0, 19).unwrap();
}
#[cfg(feature = "mmap")]
#[test]
fn mmap_growth_test() {
    let path = tempfile::tempdir().unwrap();
    let path = path.path().join("db");
    let l0 = 1 << 15; // 8 pages
    {
        let env = Env::new(&path, l0, 2).unwrap();
        let map1 = env.open_mmap(0, l0).unwrap();
        println!("{:?}", map1);
        let map2 = env.open_mmap(1, l0).unwrap();
        println!("{:?}", map2);
        map1.flush().unwrap();
        map2.flush().unwrap();
    }
    let len = std::fs::metadata(&path).unwrap().len();
    assert_eq!(len, (l0 << 2) - l0);
}
#[cfg(not(feature = "crc32"))]
fn set_crc(_ptr: *mut u8) {}
#[cfg(feature = "crc32")]
fn set_crc(ptr: *mut u8) {
    unsafe {
        let root_page = std::slice::from_raw_parts(ptr.add(8), PAGE_SIZE - 8);
        let mut h = HASHER.clone();
        h.update(root_page);
        let globptr = ptr as *mut GlobalHeader;
        (&mut *globptr).crc = h.finalize().to_le();
    }
}
/// An immutable transaction.
pub struct Txn<E: Borrow<Env>> {
    env: E,
    pub(crate) root: usize,

        use log::*;

            debug!("root lock");
            let root = env_.root.lock();
            let root = (*root + env_.roots.len() - 1) % env_.roots.len();
            debug!("shared {:?}", root);

            // Find the youngest committed version and lock it. Note
            // that there may be processes incrementing the version
            // number in parallel to this process. If that happens,
            // then since we take a shared file lock on a root page
            // (at the end of this function), the only thing that may
            // happen is that we don't open the very last version.
            let cur_mut_root =
                unsafe { (&*(env_.mmaps.lock()[0].ptr as *const GlobalHeader)).root as usize };
            // Last committed root page.
            let root = (cur_mut_root + env_.roots.len() - 1) % env_.roots.len();

            // Increase the read-only transaction count for that root,
            // and if the previous value is 0, take a file lock.

            debug!("old_n_txn: {:?}", old_n_txn);

                if let Some(ref f) = env_.roots[root].lock_file {
                    debug!("file lock shared");
                    f.lock_shared()?;
                }

                env_.lock_shared(root)?

    #[cfg(not(feature = "crc32"))]
    fn check_crc(&self, _root: usize) -> Result<(), crate::CRCError> {

    #[cfg(feature = "mmap")]
    fn lock_shared(&self, root: usize) -> Result<(), Error> {
        if let Some(ref f) = self.roots[root].lock_file {
            f.lock_shared()?;
        }

    #[cfg(feature = "crc32")]
    fn check_crc(&self, root: usize) -> Result<(), crate::CRCError> {
        unsafe {
            let maps = self.mmaps.lock();
            let page_ptr = maps[0].ptr.add(root * PAGE_SIZE);
            let crc = (&*(page_ptr as *const GlobalHeader)).crc;
            let root_page = std::slice::from_raw_parts(page_ptr.add(8), PAGE_SIZE - 8);
            let mut h = HASHER.clone();
            h.update(root_page);
            if h.finalize() != crc {
                return Err(crate::CRCError {});
            }

    #[cfg(not(feature = "mmap"))]
    fn lock_shared(&self, _root: usize) -> Result<(), Error> {
        Ok(())
    }
    #[cfg(feature = "mmap")]
    fn lock_exclusive(&self, root: usize) -> Result<(), Error> {
        if let Some(ref f) = self.roots[root].lock_file {
            f.lock_exclusive()?;
        }
        Ok(())
    }
    #[cfg(not(feature = "mmap"))]
    fn lock_exclusive(&self, _root: usize) -> Result<(), Error> {
        Ok(())
    }
    #[cfg(feature = "mmap")]
    fn unlock(&self, root: usize) -> Result<(), Error> {
        if let Some(ref f) = self.roots[root].lock_file {
            f.unlock()?

        Ok(())
    }
    #[cfg(not(feature = "mmap"))]
    fn unlock(&self, _root: usize) -> Result<(), Error> {

            if let Some(ref f) = env.roots[self.root].lock_file {
                f.unlock().unwrap_or(())
            }

            env.unlock(self.root).unwrap_or(())

    type Error = crate::Error;

    type Error = Error;

/// The trait, implemented by [`Txn`] and [`MutTxn`], for treating the
/// 4064 bytes after the header of root pages as pointers to B trees
/// (well, actually `Option` of pointers to databases, where `None` is
/// encoded by 0).

    /// Return the database stored in the root page of the current
    /// transaction at index `n`, if any.

    /// This is a straightforward implementation of just accessing index `n`.

                Some(sanakirja_core::btree::Db_ {
                    db,
                    k: std::marker::PhantomData,
                    v: std::marker::PhantomData,
                    p: std::marker::PhantomData,
                })

                Some(sanakirja_core::btree::Db_::from_page(db))

use sanakirja_core::btree;

    P: btree::BTreePage<K, V>,

    P: BTreePage<K, V>,

    P: btree::BTreePage<K, V>,

    P: BTreePage<K, V>,

    buf: &mut dyn std::io::Write,

    buf: &mut dyn Write,

sanakirja-core = { path = "../sanakirja-core", version = "1.0" }

sanakirja-core = { path = "../sanakirja-core", version = "1.0" }