// Copyright 2015 Pierre-Étienne Meunier and Florent Becker.
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.

use fs2::FileExt;
use memmap;
use skiplist;
use std;
use std::collections::{HashMap, HashSet};
use std::fs::{File, OpenOptions};
use std::path::Path;
use std::ptr::copy_nonoverlapping;
use std::sync::{Condvar, Mutex, MutexGuard};
use {Db, Representable, RC_ROOT};

/// Current version of the database. This is written as the first byte
/// in the file, and is used to recognise version changes.
pub const CURRENT_VERSION: u64 = 0;

const OFF_MAP_LENGTH: isize = 1;
const OFF_CURRENT_FREE: isize = 2;
// We need a fixed page size for compatibility reasons.
pub const PAGE_SIZE: usize = 4096;
pub const PAGE_SIZE_64: u64 = 4096;

pub const ZERO_HEADER: isize = 24; // size of the header on page 0, in bytes.
const DEFAULT_FLUSH_LIMIT: u64 = 4096;

/// Errors that can occur while transacting.
#[derive(Debug)]
pub enum Error {
    /// IO errors, from the `std::io` module.
    IO(std::io::Error),

    /// The mmap was not large enough to allocate all required memory.
    NotEnoughSpace,

    /// Lock poisoning error.
    Poison,
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match *self {
            Error::IO(ref err) => write!(f, "IO error: {}", err),
            Error::NotEnoughSpace => write!(
                f,
                "Not enough space. Try opening the environment with a larger size."
            ),
            Error::Poison => write!(f, "Lock poisoning error"),
        }
    }
}

impl std::error::Error for Error {
    fn description(&self) -> &str {
        match *self {
            Error::IO(ref err) => err.description(),
            Error::NotEnoughSpace => {
                "Not enough space. Try opening the environment with a larger size."
            }
            Error::Poison => "Poison error",
        }
    }
    fn cause(&self) -> Option<&std::error::Error> {
        match *self {
            Error::IO(ref err) => Some(err),
            Error::NotEnoughSpace => None,
            Error::Poison => None,
        }
    }
}
impl From<std::io::Error> for Error {
    fn from(e: std::io::Error) -> Error {
        Error::IO(e)
    }
}

impl<T> From<std::sync::PoisonError<T>> for Error {
    fn from(_: std::sync::PoisonError<T>) -> Error {
        Error::Poison
    }
}

// Lock order: first take thread locks, then process locks.

// Why are there two synchronization mechanisms?
// Because we would need to upgrade the read lock into a write lock,
// and there is no real way to do this with standard mechanisms.
// So, we take a mutex to make sure no other mutable transaction can start,
// and then at the time of writing, we also take the RwLock.

/// Environment, required to start any transactions. Thread-safe, but
/// opening the same database several times in the same process is not
/// cross-platform.
pub struct Env {
    length: u64,
    backing_file: File,
    flush_limit: u64,
    /// It is undefined behavior to have a file mmapped for than once.
    lock_file: Option<File>,
    mmap: Option<memmap::MmapMut>,
    map: *mut u8,
    first_unused_page: Mutex<u64>,

    /// The clock is incremented every time a Txn starts, and every
    /// time a MutTxn ends. Every time we commit, we count the number
    /// of active Txns.
    clock: Mutex<u64>,
    txn_counter: Mutex<usize>,
    /// Last commit date (according to clock) + number of active Txns
    /// at the time of the last commit. At the end of a Txn started
    /// before the last commit date, decrement the counter.
    last_commit_date: Mutex<(u64, usize)>,
    concurrent_txns_are_finished: Condvar,

    /// Ensure only one mutable transaction can be started.
    mutable: Mutex<()>,
}
unsafe impl Send for Env {}
unsafe impl Sync for Env {}

impl Drop for Env {
    fn drop(&mut self) {
        if let Some(mmap) = self.mmap.take() {
            drop(mmap);
            if let Ok(f) = self.first_unused_page.lock() {
                if *f > 0 {
                    self.backing_file.set_len(*f).unwrap_or(());
                }
            }
        }
        if let Some(ref mut lock_file) = self.lock_file {
            lock_file.unlock().unwrap_or(());
        }
    }
}

/// An immutable transaction.
pub struct Txn<'env> {
    env: &'env Env,
    start_date: u64,
}

/// A mutable transaction.
pub struct MutTxn<'env, T> {
    env: &'env Env,
    mutable: Option<MutexGuard<'env, ()>>,
    parent: T,

    /// The offset from the beginning of the file, of the first free
    /// page at the end of the file. Note that there might be other
    /// free pages in the file.
    last_page: u64,

    /// current page storing the list of free pages.
    current_list_page: Page,

    /// length of the current page of free pages.
    current_list_length: u64,

    /// position in the current page of free pages.
    current_list_position: u64,

    /// Offsets of pages that were allocated by this transaction, and
    /// have not been freed since.
    occupied_clean_pages: HashSet<u64>,

    /// Offsets of pages that were allocated by this transaction, and
    /// then freed. */
    free_clean_pages: Vec<u64>,

    /// Offsets of old pages freed by this transaction. These were
    /// *not* allocated by this transaction.
    free_pages: Vec<u64>,

    roots: HashMap<usize, u64>,

    mut_page_count: u64,
}

impl<'env> Drop for Txn<'env> {
    fn drop(&mut self) {
        let mut m = self.env.txn_counter.lock().unwrap();
        *m -= 1;
        let mut m = self.env.last_commit_date.lock().unwrap();
        if self.start_date <= m.0 {
            m.1 -= 1
        }
        if m.1 == 0 {
            self.env.concurrent_txns_are_finished.notify_one()
        }
    }
}

impl<'env, T> Drop for MutTxn<'env, T> {
    fn drop(&mut self) {
        debug!("dropping transaction");
        if let Some(ref mut guard) = self.mutable {
            debug!("dropping guard");
            **guard
        }
    }
}

#[derive(Debug)]
/// Statistics about the database, useful for debugging and
/// benchmarking purposes.
pub struct Statistics {
    /// Set of free pages inside the file (not including the free
    /// space at the end of the file)
    pub free_pages: HashSet<u64>,
    /// "Bookkeeping" pages, i.e. pages that contain just a list of free pages.
    pub bookkeeping_pages: Vec<u64>,
    /// Total number of pages in the file that are either used or
    /// free. Does not include the free space at the end of the file.
    pub total_pages: usize,
    /// A clone of the reference-counting structure, with an RC value
    /// for each page. Pages with an RC strictly less than 2 are not
    /// included.
    pub reference_counts: HashMap<u64, u64>,
}

impl Env {
    /// File size of the database path, if it exists.
    pub fn file_size<P: AsRef<Path>>(path: P) -> Result<u64, Error> {
        let db_path = path.as_ref().join("db");
        debug!(
            "db_path = {:?}, len = {:?}",
            db_path,
            std::fs::metadata(&db_path)?.len()
        );
        Ok(std::fs::metadata(&db_path)?.len())
    }

    /// Size of the underlying file. This could be different from
    /// `self.file_size()`, for instance if the environment was opened
    /// with a different size than the underlying file size.
    pub fn size(&self) -> u64 {
        self.length
    }

    /// Set the maximum number of dirty pages before triggering a
    /// flush. The default is 4096 pages. Some operating systems might
    /// have other acceptable values. For example, Linux defines the
    /// "dirty ratio" as the number of memory pages that have been
    /// written to (so-called "dirty pages") divided by number of
    /// pages in available memory. This can be checked for instance
    /// using the command `systctl -a | grep dirty`.
    pub fn set_flush_limit(&mut self, flush: u64) {
        self.flush_limit = flush
    }

    /// Get the maximum number of dirty pages before triggering a flush.
    pub fn flush_limit(&self) -> u64 {
        self.flush_limit
    }

    /// Initialize an environment. `length` must be a strictly
    /// positive multiple of 4096. The same file can only be open in
    /// one process or thread at the same time, and this is enforced
    /// by a locked file.
    pub fn new<P: AsRef<Path>>(path: P, length: u64) -> Result<Env, Error> {
        let lock_file = OpenOptions::new()
            .read(true)
            .write(true)
            .truncate(false)
            .create(true)
            .open(path.as_ref().join("db").with_extension("lock"))?;
        lock_file.lock_exclusive()?;
        let mut env = unsafe { Self::new_nolock(path, length)? };
        env.lock_file = Some(lock_file);
        Ok(env)
    }

    /// Same as [`new`](#new), but does not take a lock on the file
    /// system.
    ///
    /// This method is provided because waiting for a lock on the file
    /// system may block the whole process, whereas.
    ///
    /// 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.
    pub unsafe fn new_nolock<P: AsRef<Path>>(path: P, length: u64) -> Result<Env, Error> {
        // let length = (1 as u64).shl(log_length);
        debug!("length {:?}", length);
        let db_path = path.as_ref().join("db");
        let db_exists = std::fs::metadata(&db_path).is_ok();
        let length = if let Ok(meta) = std::fs::metadata(&db_path) {
            std::cmp::max(meta.len(), length)
        } else {
            length
        };
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .truncate(false)
            .create(true)
            .open(&db_path)?;
        debug!("allocate: {:?}", length);
        file.set_len(length)?;
        let mut mmap = memmap::MmapMut::map_mut(&file)?;
        let map = mmap.as_mut_ptr();
        debug!("mmap: {:?} {:?}", map, mmap.len());
        if !db_exists {
            std::ptr::write_bytes(map, 0, PAGE_SIZE);
            *(map as *mut u64) = CURRENT_VERSION.to_le();
        } else {
            assert!(u64::from_le(*(map as *const u64)) == CURRENT_VERSION)
        }
        debug!(
            "metadata.len() = {:?}",
            std::fs::metadata(&db_path).map(|x| x.len())
        );
        let env = Env {
            length: length,
            mmap: Some(mmap),
            map: map,
            backing_file: file,
            lock_file: None,
            first_unused_page: Mutex::new(0),
            mutable: Mutex::new(()),
            flush_limit: DEFAULT_FLUSH_LIMIT,
            txn_counter: Mutex::new(0),
            clock: Mutex::new(0),
            last_commit_date: Mutex::new((0, 0)),
            concurrent_txns_are_finished: Condvar::new(),
        };
        Ok(env)
    }
    /// Start a read-only transaction.
    pub fn txn_begin<'env>(&'env self) -> Result<Txn<'env>, Error> {
        let mut read = self.clock.lock()?;
        *read += 1;
        let mut counter = self.txn_counter.lock()?;
        *counter += 1;
        Ok(Txn {
            env: self,
            start_date: *read,
        })
    }

    /// Start a mutable transaction. Mutable transactions that go out
    /// of scope are automatically aborted.
    pub fn mut_txn_begin<'env>(&'env self) -> Result<MutTxn<'env, ()>, Error> {
        unsafe {
            debug!("taking mutable lock");
            let guard = self.mutable.lock()?;

            // Wait until all transactions that were started before
            // the start of the last mutable transaction are finished.
            debug!("taking last commit lock");
            let mut last_commit = self.last_commit_date.lock()?;
            while last_commit.1 > 0 {
                last_commit = self.concurrent_txns_are_finished.wait(last_commit)?;
            }

            debug!("lock ok");

            let last_page = u64::from_le(*((self.map as *const u64).offset(OFF_MAP_LENGTH)));
            let current_list_page =
                u64::from_le(*((self.map as *const u64).offset(OFF_CURRENT_FREE)));

            debug!("map header = {:?}, {:?}", last_page, current_list_page);
            assert!(current_list_page < self.length);
            let current_list_page = Page {
                data: self.map.offset(current_list_page as isize),
                offset: current_list_page,
            };
            let current_list_length = if current_list_page.offset == 0 {
                0
            } else {
                u64::from_le(*((current_list_page.data as *const u64).offset(1)))
            };
            Ok(MutTxn {
                env: self,
                mutable: Some(guard),
                parent: (),
                last_page: if last_page == 0 {
                    PAGE_SIZE_64
                } else {
                    last_page
                },
                current_list_page: current_list_page,
                current_list_length: current_list_length,
                current_list_position: current_list_length,
                occupied_clean_pages: HashSet::new(),
                free_clean_pages: Vec::new(),
                free_pages: Vec::new(),
                roots: HashMap::new(),
                mut_page_count: 0,
            })
        }
    }

    /// 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`.
    pub unsafe fn close(&mut self) {
        if let Some(mmap) = self.mmap.take() {
            drop(mmap);
            if let Ok(f) = self.first_unused_page.lock() {
                if *f > 0 {
                    self.backing_file.set_len(*f).unwrap_or(());
                }
            }
        }
        if let Some(ref mut lock_file) = self.lock_file {
            lock_file.unlock().unwrap_or(());
        }
    }
}

impl<'a> Txn<'a> {
    /// Compute statistics about pages. This is a potentially costlty
    /// operation, as we need to go through all bookkeeping pages.
    pub fn statistics(&self) -> Statistics {
        unsafe {
            let total_pages =
                u64::from_le(*((self.env.map as *const u64).offset(OFF_MAP_LENGTH))) as usize;
            let mut free_pages = HashSet::new();
            let mut bookkeeping_pages = Vec::new();
            let mut cur = u64::from_le(*((self.env.map as *const u64).offset(OFF_CURRENT_FREE)));
            while cur != 0 {
                bookkeeping_pages.push(cur);
                let p = self.env.map.offset(cur as isize) as *const u64;
                let prev = u64::from_le(*p);
                let len = u64::from_le(*(p.offset(1))); // size (number of u64).
                debug!("bookkeeping page: {:?}, {} {}", cur, prev, len);
                {
                    let mut p: *const u64 = (p).offset(2);
                    let mut i = 0;
                    while i < len {
                        let free_page = u64::from_le(*p);
                        if !free_pages.insert(free_page) {
                            panic!("free page counted twice: {:?}", free_page)
                        }
                        p = p.offset(1);
                        i += 1
                    }
                }
                cur = prev
            }

            let mut reference_counts = HashMap::new();
            let rc_root = self.root_(RC_ROOT);
            if rc_root != 0 {
                use Transaction;
                let db: Db<u64, u64> = Db(rc_root, std::marker::PhantomData);
                reference_counts.extend(self.iter(&db, None))
            }
            Statistics {
                total_pages: total_pages / PAGE_SIZE,
                free_pages,
                bookkeeping_pages,
                reference_counts,
            }
        }
    }

    /// Fills the `h` argument with the set of all pages referenced at
    /// least once in the tree whose root is `root`.
    pub fn references<K: Representable, V: Representable>(
        &self,
        h: &mut HashSet<u64>,
        root: Db<K, V>,
    ) {
        self.references_::<K, V>(h, root.0)
    }
    fn references_<K: Representable, V: Representable>(&self, h: &mut HashSet<u64>, page: u64) {
        // debug!("is_tree, page = {:?}", page);
        if page != 0 {
            if h.contains(&page) {
                panic!("h.contains({:?})", page);
            }
            h.insert(page);
            let page = Page {
                data: unsafe { self.env.map.offset(page as isize) },
                offset: page,
            };
            for (_, v, r) in skiplist::iter_all::<_, K, V>(&page) {
                if let Some((k, v)) = v {
                    for offset in k.page_offsets() {
                        h.insert(offset);
                    }
                    for offset in v.page_offsets() {
                        h.insert(offset);
                    }
                }
                // debug!("{:?} -> {:?}", page, r);
                self.references_::<K, V>(h, r)
            }
        }
    }
}

/// This is a semi-owned page: just as we can mutate several indices
/// of an array in the same scope, we must be able to get several
/// pages from a single environment in the same scope. However, pages
/// don't outlive their environment. Pages longer than one PAGE_SIZE
/// might trigger calls to munmap when they go out of scope.
#[derive(Debug, Clone, Copy)]
pub struct Page {
    pub data: *const u8,
    pub offset: u64,
}

#[derive(Debug)]
pub(crate) struct MutPage {
    pub data: *mut u8,
    pub offset: u64,
}

pub trait LoadPage {
    fn load_page(&self, off: u64) -> Page;
    fn root_(&self, num: usize) -> u64;
}

impl<'env> LoadPage for Txn<'env> {
    /// Find the appropriate map segment
    fn load_page(&self, off: u64) -> Page {
        trace!("load_page: off={:?}, length = {:?}", off, self.env.length);
        assert!(off < self.env.length);
        unsafe {
            Page {
                data: self.env.map.offset(off as isize),
                offset: off,
            }
        }
    }
    fn root_(&self, num: usize) -> u64 {
        assert!(ZERO_HEADER as usize + ((num + 1) << 3) < PAGE_SIZE);
        unsafe {
            u64::from_le(*((self.env.map.offset(ZERO_HEADER) as *const u64).offset(num as isize)))
        }
    }
}

impl<'env, A> LoadPage for MutTxn<'env, A> {
    fn load_page(&self, off: u64) -> Page {
        if off >= self.env.length {
            panic!("{:?} >= {:?}", off, self.env.length)
        }
        unsafe {
            Page {
                data: self.env.map.offset(off as isize),
                offset: off,
            }
        }
    }
    fn root_(&self, num: usize) -> u64 {
        if let Some(root) = self.roots.get(&num) {
            *root
        } else {
            assert!(ZERO_HEADER + ((num as isize + 1) << 3) < (PAGE_SIZE as isize));
            unsafe {
                u64::from_le(
                    *((self.env.map.offset(ZERO_HEADER) as *const u64).offset(num as isize)),
                )
            }
        }
    }
}

#[derive(Debug)]
pub(crate) enum Cow {
    Page(Page),
    MutPage(MutPage),
}

impl<'env, T> MutTxn<'env, T> {
    /// Start a mutable transaction.
    pub fn mut_txn_begin<'txn>(
        &'txn mut self,
    ) -> Result<MutTxn<'env, &'txn mut MutTxn<'env, T>>, Error> {
        unsafe {
            let mut txn = MutTxn {
                env: self.env,
                mutable: None,
                parent: std::mem::uninitialized(),
                last_page: self.last_page,
                current_list_page: Page {
                    data: self.current_list_page.data,
                    offset: self.current_list_page.offset,
                },
                current_list_length: self.current_list_length,
                current_list_position: self.current_list_position,
                occupied_clean_pages: HashSet::new(),
                free_clean_pages: Vec::new(),
                free_pages: Vec::new(),
                roots: self.roots.clone(), // reference_counts:self.reference_counts
                mut_page_count: 0,
            };
            txn.parent = self;
            Ok(txn)
        }
    }

    pub(crate) fn set_root_(&mut self, num: usize, value: u64) {
        self.roots.insert(num, value);
    }

    pub(crate) fn load_cow_page(&mut self, off: u64) -> Cow {
        trace!(
            "transaction::load_mut_page: {:?} {:?}",
            off,
            self.occupied_clean_pages
        );
        assert!(off < self.env.length);
        if off != 0 && self.occupied_clean_pages.contains(&off) {
            unsafe {
                Cow::MutPage(MutPage {
                    data: self.env.map.offset(off as isize),
                    offset: off,
                })
            }
        } else {
            unsafe {
                let d = self.env.map.offset(off as isize);
                Cow::Page(Page {
                    data: d,
                    offset: off,
                })
            }
        }
    }

    pub(crate) unsafe fn free_page(&mut self, offset: u64) {
        debug!("transaction::free page: {:?}", offset);
        if self.occupied_clean_pages.remove(&offset) {
            self.free_clean_pages.push(offset);
        } else {
            // Else, register it for freeing (we cannot reuse it in this transaction).
            self.free_pages.push(offset)
        }
    }

    /// Pop a free page from the list of free pages.
    fn free_pages_pop(&mut self) -> Option<u64> {
        debug!(
            "free_pages_pop, current_list_position:{}",
            self.current_list_position
        );
        if self.current_list_page.offset == 0 {
            None
        } else {
            if self.current_list_position == 0 {
                let previous_page =
                    unsafe { u64::from_le(*(self.current_list_page.data as *const u64)) };
                debug!("free_pages_pop, previous page:{}", previous_page);
                if previous_page == 0 {
                    None
                } else {
                    // free page (i.e. push to the list of old
                    // free pages), move to previous bookkeeping
                    // pages, and call recursively.
                    self.free_pages.push(self.current_list_page.offset);
                    unsafe {
                        self.current_list_page = Page {
                            data: self.env.map.offset(previous_page as isize),
                            offset: previous_page,
                        };
                        self.current_list_length =
                            u64::from_le(*((self.current_list_page.data as *const u64).offset(1)))
                    }
                    self.current_list_position = self.current_list_length;
                    self.free_pages_pop()
                }
            } else {
                let pos = self.current_list_position;
                // find the page at the top.
                self.current_list_position -= 1;
                debug!(
                    "free_pages_pop, new position:{}",
                    self.current_list_position
                );
                unsafe {
                    Some(u64::from_le(
                        *((self.current_list_page.data as *mut u64).offset(1 + pos as isize)),
                    ))
                }
            }
        }
    }

    /// Allocate a single page.
    pub(crate) fn alloc_page(&mut self) -> Result<MutPage, Error> {
        debug!("alloc page");

        self.mut_page_count += 1;
        if self.mut_page_count >= self.env.flush_limit {
            if let Some(ref map) = self.env.mmap {
                map.flush()?;
            }
            self.mut_page_count = 0
        }

        // If we have allocated and freed a page in this transaction, use it first.
        if let Some(page) = self.free_clean_pages.pop() {
            debug!("clean page reuse:{}", page);
            self.occupied_clean_pages.insert(page);
            Ok(MutPage {
                data: unsafe { self.env.map.offset(page as isize) },
                offset: page,
            })
        } else {
            // Else, if there are free pages, take one.
            if let Some(page) = self.free_pages_pop() {
                debug!("using an old free page: {}", page);
                self.occupied_clean_pages.insert(page);
                Ok(MutPage {
                    data: unsafe { self.env.map.offset(page as isize) },
                    offset: page,
                })
            } else {
                // Else, allocate in the free space.
                let last = self.last_page;
                debug!("eating the free space: {}", last);
                if self.last_page + PAGE_SIZE_64 < self.env.length {
                    self.last_page += PAGE_SIZE_64;
                    self.occupied_clean_pages.insert(last);
                    Ok(MutPage {
                        data: unsafe { self.env.map.offset(last as isize) },
                        offset: last,
                    })
                } else {
                    Err(Error::NotEnoughSpace)
                }
            }
        }
    }
}

/// Transactions that can be committed.
pub trait Commit {
    /// Commit the transaction.
    fn commit(self) -> Result<(), Error>;
}

impl<'a, 'env, T> Commit for MutTxn<'env, &'a mut MutTxn<'env, T>> {
    fn commit(self) -> Result<(), Error> {
        self.parent.last_page = self.last_page;
        self.parent.current_list_page = Page {
            offset: self.current_list_page.offset,
            data: self.current_list_page.data,
        };
        self.parent.current_list_length = self.current_list_length;
        self.parent.current_list_position = self.current_list_position;
        self.parent
            .occupied_clean_pages
            .extend(self.occupied_clean_pages.iter());
        self.parent
            .free_clean_pages
            .extend(self.free_clean_pages.iter());
        self.parent.free_pages.extend(self.free_pages.iter());
        for (u, v) in self.roots.iter() {
            self.parent.roots.insert(*u, *v);
        }
        Ok(())
    }
}

impl<'env> Commit for MutTxn<'env, ()> {
    /// Commit a transaction. This is guaranteed to be atomic: either
    /// the commit succeeds, and all the changes made during the
    /// transaction are written to disk. Or the commit doesn't
    /// succeed, and we're back to the state just before starting the
    /// transaction.
    fn commit(mut self) -> Result<(), Error> {
        // Tasks:
        //
        // - allocate new pages (copy-on-write) to write the new list
        // of free pages, including edited "stack pages".
        //
        // - write top of the stack
        // - write user data
        //
        // everything can be sync'ed at any time, except that the
        // first page needs to be sync'ed last.
        unsafe {
            // Copy the current bookkeeping page to a newly allocated page.
            let mut current_page = try!(self.alloc_page());
            if self.current_list_page.offset != 0 {
                // If there was at least one bookkeeping page before.
                debug!("commit: realloc BK, copy {:?}", self.current_list_position);
                copy_nonoverlapping(
                    self.current_list_page.data as *const u64,
                    current_page.data as *mut u64,
                    2 + self.current_list_position as usize,
                );
                *((current_page.data as *mut u64).offset(1)) = self.current_list_position.to_le();

                // and free the previous current bookkeeping page.
                debug!("freeing BK page {:?}", self.current_list_page.offset);
                self.free_pages.push(self.current_list_page.offset);
            } else {
                // Else, init the page.
                *(current_page.data as *mut u64) = 0; // previous page: none
                *((current_page.data as *mut u64).offset(1)) = 0; // len: 0
            }

            while !(self.free_pages.is_empty() && self.free_clean_pages.is_empty()) {
                debug!("commit: pushing");
                // If page is full, or this is the first page, allocate new page.
                let len = u64::from_le(*((current_page.data as *const u64).offset(1)));
                debug!("len={:?}", len);
                if 16 + len * 8 + 8 >= PAGE_SIZE_64 {
                    debug!("commit: current is full, len={}", len);
                    // 8 more bytes wouldn't fit in this page, time to allocate a new one

                    let p = self
                        .free_pages
                        .pop()
                        .unwrap_or_else(|| self.free_clean_pages.pop().unwrap());

                    let new_page = MutPage {
                        data: self.env.map.offset(p as isize),
                        offset: p,
                    };

                    debug!("commit {} allocated {:?}", line!(), new_page.offset);
                    // Write a reference to the current page (which cannot be null).
                    *(new_page.data as *mut u64) = current_page.offset.to_le();
                    // Write the length of the new page (0).
                    *((new_page.data as *mut u64).offset(1)) = 0;

                    current_page = new_page;
                } else {
                    // push
                    let p = self
                        .free_pages
                        .pop()
                        .unwrap_or_else(|| self.free_clean_pages.pop().unwrap());
                    debug!("commit: push {}", p);

                    // increase length.
                    *((current_page.data as *mut u64).offset(1)) = (len + 1).to_le();
                    // write pointer.
                    *((current_page.data as *mut u64).offset(2 + len as isize)) = p.to_le();
                }
            }
            // Take lock
            {
                debug!("commit: taking local lock");
                let mut last_commit = self.env.last_commit_date.lock()?;
                debug!("commit: taking txn_counter lock");
                let n_txns = self.env.txn_counter.lock()?;
                debug!("commit: taking clock lock");
                let mut clock = self.env.clock.lock()?;
                debug!("ok");
                *clock += 1;
                last_commit.0 = *clock;
                last_commit.1 = *n_txns;

                // In the future, we'd like to take an exclusive
                // inter-process lock only for committing, and not for
                // the entire transaction.
                //
                // Right now this only works between threads.

                debug!("commit: lock ok");
                for (u, v) in self.roots.iter() {
                    *((self.env.map.offset(ZERO_HEADER) as *mut u64).offset(*u as isize)) =
                        (*v).to_le();
                }
                // synchronize all maps. Since PAGE_SIZE is not always
                // an actual page size, we flush the first two pages
                // last, instead of just the last one.
                debug!("env.length = {:?}", self.env.length);
                if let Some(ref mmap) = self.env.mmap {
                    mmap.flush_range(2 * PAGE_SIZE, (self.env.length - 2 * PAGE_SIZE_64) as usize)?;
                    *((self.env.map as *mut u64).offset(OFF_MAP_LENGTH)) = self.last_page.to_le();
                    *((self.env.map as *mut u64).offset(OFF_CURRENT_FREE)) =
                        current_page.offset.to_le();
                    mmap.flush_range(0, 2 * PAGE_SIZE)?;
                }
                // This lock should not be released here.
                // debug!("commit: releasing lock");
                // self.env.lock_file.unlock().unwrap();
                {
                    let mut last = self.env.first_unused_page.lock()?;
                    debug!("last_page = {:?}", self.last_page);
                    *last = self.last_page;
                }
                if let Some(guard) = self.mutable.take() {
                    debug!("dropping guard");
                    *guard
                }

                Ok(())
            }
        }
    }
}