mmtkvdb 0.16.0

//! Key-value database using LMDB
//!
//! # Caveats / Safety
//!
//! Because of how memory-mapped I/O is being used and also because of certain
//! assumptions of the underlying LMDB API, opening environments and databases
//! requires `unsafe` Rust when using this library.
//!
//! Do not open the same environment twice in the same process at the same time
//! (see documentation of [`EnvBuilder`]). The file format is platform
//! specific, so endianness and word size must not change across use.
//!
//! When opening an already existing database, all options must be compatible
//! with the previously used options (see documentation of [`DbBuilder`]).
//!
//! Stale readers might need to be cleared manually, see
//! [`Env::clear_stale_readers`].
//!
//! # Example
//!
//! ```
//! # fn doit() -> std::io::Result<()> {
//! // question mark operator may return with `std::io::Error`
//! use mmtkvdb::{self as kv, traits::*};
//! use tempfile::tempdir;
//! let location = tempdir()?;
//! let db1_opts = kv::DbBuilder::new().name("db1");
//! let db2_opts = kv::DbBuilder::new().key_type::<str>().value_type::<i32>().name("db2");
//! let env_builder = kv::EnvBuilder::new().dir(location.path()).max_dbs(2);
//! {
//!     // SAFETY:
//!     // * database isn't opened twice in this process at the same time
//!     // * it is assumed no other processes access the environment
//!     // * since environment is newly created, it matches the platform
//!     let mut env = unsafe { env_builder.open_rw()? };
//!     // SAFETY: database doesn't exist yet (or would need compatible options)
//!     let db1 = unsafe { env.create_db(&db1_opts)? };
//!     // SAFETY: database doesn't exist yet (or would need compatible options)
//!     let db2 = unsafe { env.create_db(&db2_opts)? };
//!     let mut txn = env.txn_rw()?;
//!     txn.put(&db1, "key1".as_bytes(), "value1".as_bytes())?;
//!     txn.put(&db2, "Prime", &17)?;
//!     txn.commit()?;
//! }
//! {
//!     // SAFETY:
//!     // * database isn't opened twice in this process at the same time
//!     // * it is assumed no other processes access the environment
//!     // * the environment has been created above, thus it matches the platform
//!     let env = unsafe { env_builder.open_ro()? };
//!     // SAFETY: database options are compatible with the ones used when
//!     // creating the database
//!     let db1 = unsafe { env.open_db(&db1_opts)? };
//!     // SAFETY: database options are compatible with the ones used when
//!     // creating the database
//!     let db2 = unsafe { env.open_db(&db2_opts)? };
//!     let txn = env.txn_ro()?;
//!     assert_eq!(txn.get(&db1, "key1".as_bytes())?, Some("value1".as_bytes()));
//!     assert_eq!(txn.get_owned(&db2, "Prime")?, Some(17));
//! }
//! location.close()?;
//! # Ok(())
//! # }
//! # doit().expect("error during database access")
//! ```

#![deny(unsafe_op_in_unsafe_fn)]
#![warn(missing_docs)]
#![allow(unused_macro_rules)]

pub mod cow;
mod helpers;
pub mod lmdb;
pub mod storable;
mod syncutils;

#[cfg(test)]
mod tests;

pub use cow::{GenericCow, Owned};
use helpers::*;
pub use storable::{BorrowStorable, NoKey, NoValue, Storable, StorableConstBytesLen, StorableRef};
use syncutils::*;

use std::cell::UnsafeCell;
use std::collections::{HashMap, HashSet};
use std::ffi::{c_int, c_uint, CStr, CString};
use std::fmt;
use std::hash::{Hash, Hasher};
use std::io;
use std::marker::PhantomData;
use std::mem::{forget, take, MaybeUninit};
use std::ops::{Deref, DerefMut};
use std::path::Path;
use std::ptr::{null, null_mut};
use std::slice;
use std::str;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering::Relaxed};
use std::sync::{Arc, Mutex, Weak};

// TODO: Ensure that `usize` is really `c_size_t` without depending on unstable
// features.
#[allow(non_camel_case_types)]
type c_size_t = usize;
/*
// Compile-time check enforcing `usize` is same size as `c_size_t`
const _: fn() = || {
    let _ = core::mem::transmute::<usize, std::ffi::c_size_t>;
};
*/

/// Important traits re-exported without name
pub mod traits {
    pub use super::Env as _;
    pub use super::EnvRef as _;
    pub use super::Txn as _;
}

/// Converts an error code from LMDB into a `Result`
///
/// # Safety
///
/// * If `code` is negative, it must be a valid LMDB error code.
/// * `lmdb::mdb_strerror` is assumed to be thread-safe for valid negative LMDB
///   error codes. This is currently not guaranteed by LMDB's API
///   documentation. See also: `https://bugs.openldap.org/show_bug.cgi?id=8361`
unsafe fn check_err_code(code: c_int) -> io::Result<()> {
    if code > 0 {
        Err(io::Error::from_raw_os_error(code))
    } else if code < 0 {
        Err(io::Error::new(
            io::ErrorKind::Other,
            // SAFETY: because `code` is negative and an existent LMDB error
            // code, `lmdb::mdb_strerror` returns a pointer to a static
            // C string (not guaranteed by the API docs, but believed to not
            // change in future versions of LMDB)
            unsafe { CStr::from_ptr(lmdb::mdb_strerror(code)) }
                .to_str()
                .unwrap_or("unknown error (LMDB error string is not valid UTF-8)"),
        ))
    } else {
        Ok(())
    }
}

/// Type alias for [`c_uint`] as used by LMDB to specify flags
type LmdbFlags = c_uint;

/// Macro checking if an `lmdb_flags` field has a certain flag set
macro_rules! flag_get {
    ($self:ident, $flag:ident) => {
        ($self.lmdb_flags & (lmdb::$flag as LmdbFlags) != 0)
    };
}

/// Macro setting or clearing a flag in the `lmdb_flags` field
macro_rules! flag_set {
    ($self:ident, $flag:ident, $value:expr) => {
        if $value {
            $self.lmdb_flags |= lmdb::$flag as LmdbFlags;
        } else {
            $self.lmdb_flags &= !(lmdb::$flag as LmdbFlags);
        }
    };
}

static ENV_COUNTER: AtomicU64 = AtomicU64::new(0);

/// Internal struct with LMDB environment handle as raw pointer
///
/// On drop, the environment is closed. Contains an additional mutex to
/// synchronize opening/creating/closing databases.
#[derive(Debug)]
struct EnvBackend {
    inner: *mut lmdb::MDB_env,
    id: u64,
    max_keysize: usize,
    nestable_txns: bool,
    /// `Mutex` ensures synchronization of `lmdb::mdb_dbi_open` and contained
    /// `HashMap` avoids re-creating a new `DbBackend` for a database that is
    /// already open (because databases must not be closed more than once)
    dbis: Mutex<HashMap<lmdb::MDB_dbi, Weak<DbBackend>>>,
}

// SAFETY: `lmdb::MDB_env` is generally thread-safe, but particular function
// must be synchronized
unsafe impl Send for EnvBackend {}
// SAFETY: `lmdb::MDB_env` is generally thread-safe, but particular function
// must be synchronized
unsafe impl Sync for EnvBackend {}

impl PartialEq for EnvBackend {
    fn eq(&self, other: &Self) -> bool {
        self.inner == other.inner
    }
}

impl Eq for EnvBackend {}

impl Hash for EnvBackend {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.inner.hash(state);
    }
}

impl Drop for EnvBackend {
    fn drop(&mut self) {
        // LMDB's API specification is contradictory and claims both that
        // databases must be closed before calling `mdb_env_close` and that
        // "closing a database handle is not necessary". We close them anyway,
        // just to be sure:
        for dbi in take(&mut *self.dbis.lock().unwrap()).into_keys() {
            // SAFETY:
            //  *  due to `&mut self`, it is not possible to call this function
            //     (for a given environment) from different threads at the same
            //     time
            //  *  the database can't have been closed already, because the
            //     `lmdb::MDB_dbi` is always removed from `EnvBackend::dbis` in
            //     that case
            //  *  all transactions have been closed at this point because
            //     `TxnRo<'a>` and `TxnRw<'a>` have a lifetime `'a` which has
            //     ended by now
            //  *  all cursors have been closed at this point because closing
            //     all transactions also closes all cursors
            unsafe {
                lmdb::mdb_dbi_close(self.inner, dbi);
            }
        }
        // SAFETY:
        //  *  due to `&mut self`, it is not possible to call this function
        //     (for a given environment) from different threads at the same
        //     time
        //  *  all transactions, cursors, and databases have been closed (see
        //     above)
        unsafe { lmdb::mdb_env_close(self.inner) }
    }
}

/// Read-only handle for accessing environment that stores key-value databases
///
/// An environment can be opened using [`EnvBuilder`].
///
/// The methods for read-only access are accessible through the [`Env`] trait
/// (which is implemented by `EnvRo` and [`EnvRw`]).
///
/// Use [`Env::open_db`] to retrieve database handles and [`Env::txn_ro`] to
/// start a read-only transaction.
///
/// It's possible to clone the handle, in which case the environment will be
/// closed when the last handle is dropped. It's also possible to create a
/// read-only handle from a read-write handle ([`EnvRw`]) by invoking
/// [`Env::clone_ro`].
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct EnvRo {
    backend: Arc<EnvBackend>,
}

/// Read-write handle for accessing environment that stores key-value databases
///
/// An environment can be opened using [`EnvBuilder`].
///
/// The methods for read-only access are accessible through the [`Env`] trait
/// (which is implemented by [`EnvRo`] and `EnvRw`).
/// Methods for write access, however, are available directly on the `EnvRw`
/// struct.
///
/// Use [`Env::open_db`] or [`EnvRw::create_db`] to retrieve database handles
/// and [`EnvRw::txn_rw`] to start a read-write transaction.
///
/// It's also possible to create a read-only handle from a read-write handle by
/// invoking [`Env::clone_ro`].
#[derive(PartialEq, Eq, Hash, Debug)]
pub struct EnvRw {
    env_ro: EnvRo,
}

impl AsRef<EnvRo> for EnvRo {
    fn as_ref(&self) -> &EnvRo {
        &self
    }
}

impl AsRef<EnvRo> for EnvRw {
    fn as_ref(&self) -> &EnvRo {
        &self.env_ro
    }
}

/// Abstraction over [`&EnvRo`](EnvRo) and [`&mut EnvRw`](EnvRw), which allows
/// creating databases in the latter case if non-existing
///
/// This trait is only needed if you want to create generic code which works on
/// either `&EnvRo` or `&mut EnvRw` and creates databases in the latter case if
/// non-existing.
/// In most cases, `EnvRo` or `EnvRw` can be used directly. (Note that you can
/// call `EnvRo`'s methods on `EnvRw` automatically due to [`Deref`] coercion.)
pub trait EnvRef {
    /// Open database in environment and, if possible, create if non-existing
    ///
    /// Note that the implementation of `EnvRef` for [`&EnvRo`]() does *not*
    /// create any database if non-existing.
    ///
    /// # Safety
    ///
    /// * If a database exists already, it must have been created with
    ///   compatible options.
    /// * Implementation of [`Storable`] for the used
    ///   [keys](DbBuilder::key_type) and [values](DbBuilder::value_type) must
    ///   behave the same as when the database was created.
    unsafe fn open_or_create_db<K, V, C>(
        &mut self,
        options: &DbSpec<K, V, C>,
    ) -> io::Result<Db<K, V, C>>
    where
        K: Storable + ?Sized,
        V: Storable + ?Sized,
        C: Constraint;
}

impl<T: Env> EnvRef for &T {
    unsafe fn open_or_create_db<K, V, C>(
        &mut self,
        options: &DbSpec<K, V, C>,
    ) -> io::Result<Db<K, V, C>>
    where
        K: Storable + ?Sized,
        V: Storable + ?Sized,
        C: Constraint,
    {
        // SAFETY: all requirements of `EnvRo::open_db` are listed in
        // documentation of `EnvRef:open_or_create_db` as well
        unsafe { self.open_db(options) }
    }
}

impl EnvRef for &mut EnvRw {
    unsafe fn open_or_create_db<K, V, C>(
        &mut self,
        options: &DbSpec<K, V, C>,
    ) -> io::Result<Db<K, V, C>>
    where
        K: Storable + ?Sized,
        V: Storable + ?Sized,
        C: Constraint,
    {
        // SAFETY: all requirements of `EnvRw::create_db` are listed in
        // documentation of `EnvRef:open_or_create_db` as well
        unsafe { self.create_db(options) }
    }
}

/// Internal struct with LMDB transaction handle as raw pointer
///
/// On drop, all associated cursors (which haven't been dropped and thus closed
/// yet) will be closed.
#[derive(Debug)]
struct TxnBackend<'a> {
    env_ro: &'a EnvRo,
    inner: *mut lmdb::MDB_txn,
    cursors: WeakVec<CursorBackend>,
}

impl<'a> TxnBackend<'a> {
    fn close_cursors(&mut self) {
        for cursor in take(&mut self.cursors).iter() {
            cursor.closed.store(true, Relaxed);
            // SAFETY:
            //  *  Holding a shared reference to `cursor` means that its drop
            //     handler did not run yet. Thus the cursor has not been closed
            //     by the drop handler. Storing `true` in `closed` ensures that
            //     when the drop handler later obtains a mutable reference, it
            //     will read `true` and doesn't close the cursor twice.
            //  *  The cursor also won't be closed by this method twice,
            //     because `self.cursors` is emptied above.
            //  *  There is no other place where the cursor could have been
            //     closed.
            //  *  All callers of `TxnBackend::close_cursors` will drop or
            //     forget the `TxnBackend` afterwards. Thus the cursor can't be
            //     used after calling `lmdb::mdb_cursor_close` below, because
            //     using the cursor requires a `TxnRo` or `TxnRw` handle. In
            //     this context, `CursorBackend::assert_txn_backend` ensures
            //     that a cursor handle really belongs to the transaction for
            //     which it has been created. Even a `Relaxed` store of `true`
            //     to `cursor.closed` will happen-before subsequent `Relaxed`
            //     loads in `CursorBackend::assert_txn_backend` if the address
            //     `TxnBackend.inner` has been reused. That is because
            //     releasing the allocated memory in LMDB synchronizes-with an
            //     allocation from within the same memory region.
            unsafe {
                lmdb::mdb_cursor_close(cursor.inner);
            }
        }
    }
    /// Drop without aborting the transaction (closes all cursors)
    fn dont_abort(mut self) {
        self.close_cursors();
        // NOTE: doesn't leak because `self.cursors` isn't holding any
        // allocation
        forget(self);
    }
}

impl<'a> Drop for TxnBackend<'a> {
    fn drop(&mut self) {
        self.close_cursors();
        // SAFETY: transaction cannot be used further and all cursors have been
        // closed
        unsafe {
            lmdb::mdb_txn_abort(self.inner);
        }
    }
}

/// Read-only transaction
///
/// A read-only transaction can be started with [`Env::txn_ro`].
///
/// The methods for read-only access to the database are accessible through the
/// [`Txn`] trait (which is implemented by `TxnRo` and [`TxnRw`]).
///
/// Read-only transactions do not need to be committed or aborted; their
/// handles can be simply dropped when not needed anymore.
///
/// Opposed to read-write transactions ([`TxnRw`]), read-only transactions
/// (`TxnRo`) are [`Send`]. However, they are not [`Sync`]. Thus, if you want
/// to use a read-only transaction from several threads concurrently, you have
/// to synchronize the use, e.g. by wrapping the `TxnRo` in a [`Mutex`].
#[derive(Debug)]
pub struct TxnRo<'a> {
    backend: UnsafeCell<TxnBackend<'a>>,
}

// SAFETY: Option `lmdb::MDB_NOTLS` is used, thus `TxnRo` can be used from
// different threads. However, the `UnsafeCell` demands `!Sync`.
unsafe impl<'a> Send for TxnRo<'a> {}

/// Helper type to store `HashSet<ArcByAddr<DbBackend>>` as owned value or
/// mutable reference
#[derive(Debug)]
enum UsedDbs<'a> {
    Toplevel(HashSet<ArcByAddr<DbBackend>>),
    Nested(&'a mut HashSet<ArcByAddr<DbBackend>>),
}

impl<'a> Deref for UsedDbs<'a> {
    type Target = HashSet<ArcByAddr<DbBackend>>;
    fn deref(&self) -> &Self::Target {
        match self {
            UsedDbs::Toplevel(x) => x,
            UsedDbs::Nested(x) => x,
        }
    }
}

impl<'a> DerefMut for UsedDbs<'a> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        match self {
            UsedDbs::Toplevel(x) => x,
            UsedDbs::Nested(x) => x,
        }
    }
}

/// Read-write transaction
///
/// A read-write transaction can be started with [`EnvRw::txn_rw`].
///
/// The methods for read-only access to the database are accessible through the
/// [`Txn`] trait (which is implemented by [`TxnRo`] and `TxnRw`).
/// Methods for write access, however, are available directly on the `TxnRw`
/// struct.
///
/// Read-write transactions are [aborted](TxnRw::abort) when dropped. To make
/// changes permanent, [`TxnRw::commit`] must be called.
///
/// Read-write transactions are neither [`Send`] nor [`Sync`].
pub struct TxnRw<'a> {
    backend: UnsafeCell<TxnBackend<'a>>,
    used_dbs: UsedDbs<'a>,
    commit_handlers: Vec<Box<dyn FnOnce(&mut Self) -> io::Result<()> + 'a>>,
}

impl<'a> fmt::Debug for TxnRw<'a> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Point")
            .field("backend", &self.backend)
            .field("used_dbs", &self.used_dbs)
            .finish_non_exhaustive()
    }
}

/// Internal struct with LMDB database handle
///
/// `DbBackend` can be wrapped in an [`ArcByAddr`] to allow for equality checks
/// and hashing.
///
/// There is no strong reference to [`EnvBackend`] to avoid a lifetime argument
/// (particularly such that databases can persist beyond mutation of
/// [`EnvRw`]).
///
/// It is an error if a database handle is used on a different environment.
/// Method [`DbBackend::assert_env_backend`] can be used to panic in that case.
///
/// Note that `DbBackend`s must not be dropped while there is an open
/// transaction which modified the respective database. Hence clones of
/// `ArcByAddr<DbBackend>` must be stored in [`TxnRw::used_dbs`] before
/// modifying any data in the database. This will defer invocation of
/// [`DbBackend::drop`] and [`lmdb::mdb_dbi_close`].
#[derive(Debug)]
struct DbBackend {
    env_backend: Weak<EnvBackend>,
    env_id: u64,
    inner: lmdb::MDB_dbi,
}

impl DbBackend {
    /// Drop without running normal destructor
    fn dont_close(mut self) {
        self.env_backend = Default::default();
        // NOTE: doesn't leak because `self.env_backend` doesn't hold any
        // allocation
        forget(self);
    }
    /// Panic if database handle does not belong to environment
    fn assert_env_backend(&self, env_backend: &EnvBackend) {
        if self.env_id != env_backend.id {
            panic!("database does not belong to environment");
        }
    }
}

impl Drop for DbBackend {
    fn drop(&mut self) {
        if let Some(env_backend) = Weak::upgrade(&self.env_backend) {
            let mut dbis = env_backend.dbis.lock().unwrap();
            // If `env_backend.dbis` does not contain the `lmdb::MDB_dbi`, then
            // `lmdb::mdb_dbi_close` has already been called.
            if let Some(weak) = dbis.get(&self.inner) {
                // If `weak.strong_count` is not zero, then the database has
                // been reopened after the last `Arc<DbBackend> was dropped.
                if weak.strong_count() == 0 {
                    // Removing the `lmdb::MDB_dbi` from `dbis` ensures that
                    // the database doesn't get closed twice.
                    dbis.remove(&self.inner);
                    // SAFETY:
                    //  *  due to `env_backend.dbis.lock()`, it is not possible
                    //     to call this function (for a given environment) from
                    //     different threads at the same time
                    //  *  because `weak.strong_count()` is zero, no new
                    //     `DbBackend` (which is still in use) with the same
                    //     `DbBackend::inner` may have been created by
                    //     `EnvRo::open_or_opt_create_db`
                    //  *  removing the `lmdb::MDB_dbi` from `dbis` ensures that
                    //     `mdb_dbi_close` isn't called twice, even if a
                    //     recreated `DbBackend` gets dropped later
                    //  *  `mdb_cursor_close` isn't called later because all
                    //     `CursorBackend`s must have been dropped before the last
                    //     `Db` handle and before thus the `DbBackend` is dropped
                    //     (see SAFETY comment on `CursorBackend`)
                    unsafe {
                        lmdb::mdb_dbi_close(env_backend.inner, self.inner);
                    }
                }
            }
        }
    }
}

/// Database handle
///
/// These handles are created using [`DbBuilder`] builder, which can be passed
/// by (shared) reference to [`Env::open_db`] or [`EnvRw::create_db`].
///
/// The type arguments `K`, `V`, and `C` reflect the type used for keys and
/// values, and whether duplicate keys are allowed (`C` being [`KeysUnique`] or
/// [`KeysDuplicate`]), respectively.
///
/// Operations on databases is done using the methods of the [`Txn`] trait or
/// [`TxnRw`] struct. The database handle must be passed to these methods by
/// (shared) reference.
///
/// Using a [`Db`] handle in a different environment (or in an environment that
/// has been closed and re-opened) will cause a panic.
#[derive(Debug)]
pub struct Db<K: ?Sized, V: ?Sized, C> {
    key: PhantomData<fn(&K) -> &K>,
    value: PhantomData<fn(&V) -> &V>,
    constraint: C,
    backend: ArcByAddr<DbBackend>,
}

impl<K: ?Sized, V: ?Sized, C> Clone for Db<K, V, C>
where
    C: Constraint,
{
    fn clone(&self) -> Self {
        Db {
            key: PhantomData,
            value: PhantomData,
            constraint: self.constraint,
            backend: self.backend.clone(),
        }
    }
}

impl<K: ?Sized, V: ?Sized, C> PartialEq for Db<K, V, C> {
    fn eq(&self, other: &Self) -> bool {
        self.backend == other.backend
    }
}

impl<K: ?Sized, V: ?Sized, C> Eq for Db<K, V, C> {}

/// Internal struct with LMDB cursor handle
///
/// There is no lifetime argument, but cursors will be closed when
/// [`TxnBackend`] gets dropped. In that case, this struct will be marked as
/// closed by setting [`CursorBackend::closed`] to true.
///
/// `CursorBackend` can be wrapped in an [`ArcByAddr`] to allow for equality
/// checks and hashing. (Equality can't be checked by comparing the inner raw
/// pointer because the address behind the raw pointer could be reused after
/// the cursor has been closed.)
///
/// Like database handles, cursor handles have no lifetime argument.
///
/// It is an error if a cursor handle is used on a different transaction.
/// Method [`CursorBackend::assert_txn_backend`] can be used to panic in that
/// case.
///
/// This struct implements `Send` and `Sync`, thus proper synchronization must
/// be ensured when using it (which is done by making operations work on
/// transaction handles).
//
// SAFETY: Creators must ensure that a corresponding `Db` handle exists which
// gets dropped after `CursorBackend` gets dropped. See `CursorBackend::drop`.
#[derive(Debug)]
struct CursorBackend {
    inner: *mut lmdb::MDB_cursor,
    closed: AtomicBool,
    inner_txn: *mut lmdb::MDB_txn,
}

// SAFETY: `Cursor`s can only be used with a `TxnRo` or `TxnRw`. Those
// transaction handles implement `Send` or `!Send` as necessary.
unsafe impl Send for CursorBackend {}
// SAFETY: `Cursor`s can only be used with a `TxnRo` or `TxnRw`. Those
// transaction handles implement `Sync` or `!Sync` as necessary.
unsafe impl Sync for CursorBackend {}

impl CursorBackend {
    /// Panic if cursor handle does not belong to transaction
    fn assert_txn_backend(&self, txn_backend: &TxnBackend) {
        if self.closed.load(Relaxed) || self.inner_txn != txn_backend.inner {
            panic!("cursor does not belong to transaction");
        }
    }
}

impl Drop for CursorBackend {
    fn drop(&mut self) {
        // NOTE: mutex should never be poisoned
        if !*self.closed.get_mut() {
            // SAFETY:
            //  *  If the cursor was closed before, then
            //     `TxnBackend::close_cursors` would read as `true` because
            //     the only other place where the cursor might have been closed
            //     is `TxnBackend::close_cursors`.
            //  *  `CursorBackend` is always dropped before the last associated
            //     `Db` handle and thus before the `DbBackend` is dropped.  See
            //     SAFETY comment on `CursorBackend`. Therefore,
            //     `lmdb::mdb_cursor_close` gets called before
            //     `lmdb::mdb_dbi_close`. Note that the LMDB API specification
            //     is unclear about whether this is a requirement for read-only
            //     transactions. For read-write transactions where there has
            //     been an actual write to the database, this is ensured also
            //     additionally due to storing the `DbBackend` in
            //     `TxnRw:used_dbs`.
            unsafe {
                lmdb::mdb_cursor_close(self.inner);
            }
        }
    }
}

#[derive(Debug)]
/// Cursor for reading from or writing to a particular database
///
/// A new cursor can be created using [`Txn::new_cursor`]. It is used by
/// invoking one of the cursor methods in [`Txn`] or [`TxnRw`].
pub struct Cursor<K: ?Sized, V: ?Sized, C> {
    // SAFETY: field order matters, see `CursorBackend::drop`
    backend: ArcByAddr<CursorBackend>,
    db: Db<K, V, C>,
}

impl<K: ?Sized, V: ?Sized, C> Clone for Cursor<K, V, C>
where
    C: Constraint,
{
    fn clone(&self) -> Self {
        Cursor {
            db: self.db.clone(),
            backend: self.backend.clone(),
        }
    }
}

impl<K: ?Sized, V: ?Sized, C> PartialEq for Cursor<K, V, C> {
    fn eq(&self, other: &Self) -> bool {
        self.backend == other.backend
    }
}

impl<K: ?Sized, V: ?Sized, C> Eq for Cursor<K, V, C> {}

/// Options for opening an environment
///
/// This struct follows the builder pattern. Start with [`EnvBuilder::new`] and
/// open the environment with [`EnvBuilder::open_ro`] (for read-only access) or
/// [`EnvBuilder::open_rw`] (for read-write access).
///
/// # Caveats / Safety
///
/// Do not open the same environment twice in the same process at the same
/// time.
///
/// Since this cannot be easily ensured (and because the environment must be
/// free of corruption on the storage medium), [`open_ro`](EnvBuilder::open_ro)
/// and [`open_rw`](EnvBuilder::open_rw) are marked as `unsafe`.
/// Use [`Env::clone_ro`] to (re-)open an already open environment in read-only
/// mode.
///
/// The file format is platform specific, so endianness and word size must not
/// change across use.
///
/// # Example
///
/// ```
/// # fn doit() -> std::io::Result<()> {
/// // question mark operator may return with `std::io::Error`
/// use mmtkvdb::{self as kv, traits::*};
/// use tempfile::tempdir;
/// let location = tempdir()?;
/// let env_builder = kv::EnvBuilder::new().dir(location.path()).max_dbs(1);
/// // SAFETY:
/// // * database isn't opened twice in this process at the same time
/// // * it is assumed no other processes access the environment
/// // * since environment is newly created, it matches the platform
/// let mut env_rw = unsafe { env_builder.open_rw() }?;
/// let env_ro = env_rw.clone_ro();
/// # Ok(())
/// # }
/// # doit().expect("error during database access")
/// ```
#[derive(Clone, Debug)]
pub struct EnvBuilder<P> {
    path: P,
    lmdb_flags: LmdbFlags,
    max_size: usize,
    max_readers: usize,
    max_dbs: usize,
    unix_file_mode: u32,
}

/// Constraints on database (type argument `C` to [`DbBuilder`] and [`Db`])
pub trait Constraint: Copy + 'static {
    /// Duplicate keys allowed?
    const DUPLICATE_KEYS: bool;
}

/// Type argument to [`DbBuilder`] and [`Db`] indicating unique keys
#[derive(Clone, Copy, Debug)]
pub struct KeysUnique;

impl Constraint for KeysUnique {
    const DUPLICATE_KEYS: bool = false;
}

/// Type argument to [`DbBuilder`] and [`Db`] indicating non-unique keys
#[derive(Clone, Copy, Debug)]
pub struct KeysDuplicate;

impl Constraint for KeysDuplicate {
    const DUPLICATE_KEYS: bool = true;
}

/// Type alias for `DbBuilder` which has a [name](DbBuilder::name) set (or
/// unnamed database selected)
///
/// Values of type `DbSpec<K, V, C>` are created by creating a new
/// [`DbBuilder`] with [`DbBuilder::new`], setting the desired options, and
/// then invoking [`DbBuilder::name`] or [`DbBuilder::unnamed`].
///
/// Values of type `DbSpec<K, V, C>` can be passed to the following methods in
/// order to create and/or open databases:
///
/// * [`Env::open_db`]
/// * [`EnvRw::create_db`] (same as `Env::open_db` but creates the database if
///   not existent)
/// * [`EnvRef::open_or_create_db`] (never creates the database if called on
///   `&EnvRo`)
pub type DbSpec<K, V, C> = DbBuilder<K, V, C, Option<CString>>;

/// Options for opening a database
///
/// This struct follows the builder pattern. Start with [`DbBuilder::new`] and
/// pass completed builder by (shared) reference to [`Env::open_db`] or
/// [`EnvRw::create_db`] to open or create a database.
///
/// # Caveats / Safety
///
/// When opening an already existing database, all options must be compatible
/// with the previously used options. It has be be ensured by the caller that
/// this is true. Thus [`Env::open_db`] and [`EnvRw::create_db`] are marked as
/// `unsafe`.
///
/// [`DbBuilder::name`] has to be called last (after setting all other
/// options). This is due to some restrictions regarding `const fn`s.
/// A compiler error will be reported if the methods are invoked in the wrong
/// order.
///
/// # Type arguments
///
/// The type arguments `K`, `V`, and `C` reflect the type used for keys and
/// values, and whether duplicate keys are allowed (`C` being [`KeysUnique`] or
/// [`KeysDuplicate`]), respectively.
///
/// The fourth type parameter `N` determines if a [name](DbBuilder::name)
/// <small>(or the [unnamed database](DbBuilder::unnamed))</small>
/// has been set, or if a name must be selected yet. `N` is `()` if no database
/// name has been selected yet. Once a name has been selected, you can use the
/// type alias [`DbSpec<K, V, C>`] to refer to the completed builder.
///
/// # Example
///
/// ```
/// # fn doit() -> std::io::Result<()> {
/// // question mark operator may return with `std::io::Error`
/// use mmtkvdb::{self as kv, Txn as _};
/// use tempfile::tempdir;
/// let location = tempdir()?;
/// let env_builder = kv::EnvBuilder::new().dir(location.path()).max_dbs(1);
/// // SAFETY:
/// // * database isn't opened twice in this process at the same time
/// // * it is assumed no other processes access the environment
/// // * since environment is newly created, it matches the platform
/// let mut env_rw = unsafe { env_builder.open_rw() }?;
/// let db_opts = kv::DbBuilder::new()
///     .key_type::<str>()
///     .value_type::<i32>()
///     .name("account_balance");
/// // SAFETY: database doesn't exist yet (or would need compatible options)
/// let db = unsafe { env_rw.create_db(&db_opts) }?;
/// # Ok(())
/// # }
/// # doit().expect("error during database access")
/// ```
#[derive(Clone, Debug)]
pub struct DbBuilder<K: ?Sized, V: ?Sized, C, N> {
    key: PhantomData<fn(&K) -> &K>,
    value: PhantomData<fn(&V) -> &V>,
    constraint: C,
    lmdb_flags: LmdbFlags,
    name: N,
}

/// Some default values used for [`EnvBuilder`]
pub mod env_builder_defaults {
    /// Default value for `max_size`
    pub const MAX_SIZE: usize = 1024 * 1024 * 1024;
    /// Default value for `max_readers`
    pub const MAX_READERS: usize = 126;
    /// Default value for `max_dbs`
    pub const MAX_DBS: usize = 0;
    /// Default value for `unix_file_mode`
    pub const UNIX_FILE_MODE: u32 = 0o600;
}

impl EnvBuilder<()> {
    /// Default options for environment
    pub const fn new() -> Self {
        Self {
            path: (),
            // SAFETY: `lmdb::MDB_NOTLS` is required for `TxnRo` to be `Send`
            // and for `EnvRo::open_or_opt_create_db` to be called with its
            // `create` argument set to `false` while other transactions are
            // open
            lmdb_flags: lmdb::MDB_NORDAHEAD as LmdbFlags | lmdb::MDB_NOTLS as LmdbFlags,
            max_size: env_builder_defaults::MAX_SIZE,
            max_readers: env_builder_defaults::MAX_READERS,
            max_dbs: env_builder_defaults::MAX_DBS,
            unix_file_mode: env_builder_defaults::UNIX_FILE_MODE,
        }
    }
}

impl<P> EnvBuilder<P> {
    /// Set directory location
    ///
    /// Overwrites a previously set directory or file location.
    pub fn dir<'a>(mut self, path: &'a (impl AsRef<Path> + ?Sized)) -> EnvBuilder<&'a Path> {
        flag_set!(self, MDB_NOSUBDIR, false);
        EnvBuilder {
            path: path.as_ref(),
            lmdb_flags: self.lmdb_flags,
            max_size: self.max_size,
            max_readers: self.max_readers,
            max_dbs: self.max_dbs,
            unix_file_mode: self.unix_file_mode,
        }
    }
    /// Set file location
    ///
    /// Overwrites a previously set directory or file location.
    pub fn file<'a>(mut self, path: &'a (impl AsRef<Path> + ?Sized)) -> EnvBuilder<&'a Path> {
        flag_set!(self, MDB_NOSUBDIR, true);
        EnvBuilder {
            path: path.as_ref(),
            lmdb_flags: self.lmdb_flags,
            max_size: self.max_size,
            max_readers: self.max_readers,
            max_dbs: self.max_dbs,
            unix_file_mode: self.unix_file_mode,
        }
    }
    /// Get write-map option
    pub const fn get_writemap(&self) -> bool {
        flag_get!(self, MDB_WRITEMAP)
    }
    /// Set or clear write-map option
    ///
    /// Enabling this option is incompatible with nested transactions
    /// (see [`TxnRw::nested`]).
    pub const fn writemap(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_WRITEMAP, flag);
        self
    }
    /// Get no-meta-sync option
    pub const fn get_nometasync(&self) -> bool {
        flag_get!(self, MDB_NOMETASYNC)
    }
    /// Set or clear no-meta-sync option
    pub const fn nometasync(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_NOMETASYNC, flag);
        self
    }
    /// Get no-sync option
    pub const fn get_nosync(&self) -> bool {
        flag_get!(self, MDB_NOSYNC)
    }
    /// Set or clear no-sync option
    pub const fn nosync(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_NOSYNC, flag);
        self
    }
    /// Get map-async option
    pub const fn get_mapasync(&self) -> bool {
        flag_get!(self, MDB_MAPASYNC)
    }
    /// Set or clear map-async option
    pub const fn mapasync(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_MAPASYNC, flag);
        self
    }
    /// Get no-read-ahead option
    pub const fn get_nordahead(&self) -> bool {
        flag_get!(self, MDB_NORDAHEAD)
    }
    /// Set or clear no-read-ahead option
    pub const fn nordahead(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_NORDAHEAD, flag);
        self
    }
    /// Get maximum environment size
    pub const fn get_max_size(&self) -> usize {
        self.max_size
    }
    /// Set maximum environment size
    pub const fn max_size(mut self, max_size: usize) -> Self {
        self.max_size = max_size;
        self
    }
    /// Get maximum number of concurrent readers
    pub const fn get_max_readers(&self) -> usize {
        self.max_readers
    }
    /// Set maximum number of concurrent readers
    pub const fn max_readers(mut self, max_readers: usize) -> Self {
        self.max_readers = max_readers;
        self
    }
    /// Get maximum number of named databases
    pub const fn get_max_dbs(&self) -> usize {
        self.max_dbs
    }
    /// Set maximum number of named databases
    pub const fn max_dbs(mut self, max_dbs: usize) -> Self {
        self.max_dbs = max_dbs;
        self
    }
    /// Get file mode
    pub const fn get_unix_file_mode(&self) -> u32 {
        self.unix_file_mode
    }
    /// Set file mode
    pub const fn unix_file_mode(mut self, mode: u32) -> Self {
        self.unix_file_mode = mode;
        self
    }
    /// Check if transactions are nestable with current options
    ///
    /// See [`TxnRw::nested`].
    pub const fn nestable_txns(&self) -> bool {
        !self.get_writemap()
    }
}

impl<'a> EnvBuilder<&'a Path> {
    /// Open environment
    ///
    /// # Safety
    ///
    /// See [`EnvBuilder::open_ro`] and [`EnvBuilder::open_rw`]
    unsafe fn open_with_readonly_flag(&self, readonly: bool) -> io::Result<EnvRo> {
        let path = path_to_cstring(self.path)?;
        let mut inner = MaybeUninit::<*mut lmdb::MDB_env>::uninit();
        // SAFETY:
        //  *  `lmdb::mdb_env_create` doesn't have any requirements
        //  *  the status passed to `check_err_code` is valid
        unsafe {
            check_err_code(lmdb::mdb_env_create(inner.as_mut_ptr()))?;
        }
        // SAFETY: call to `lmdb::mdb_env_create` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        let inner = unsafe { inner.assume_init() };
        let backend = EnvBackend {
            inner: inner,
            id: ENV_COUNTER.fetch_add(1, Relaxed),
            max_keysize: unsafe { lmdb::mdb_env_get_maxkeysize(inner) }
                .try_into()
                .unwrap_or(usize::MAX),
            nestable_txns: self.nestable_txns(),
            dbis: Default::default(),
        };
        // SAFETY:
        //  *  `lmdb::mdb_env_set_mapsize` doesn't have any requirements
        //  *  the status passed to `check_err_code` is valid
        unsafe {
            check_err_code(lmdb::mdb_env_set_mapsize(
                backend.inner,
                self.max_size.try_into().map_err(|_| {
                    io::Error::new(io::ErrorKind::InvalidInput, "maximum size invalid")
                })?,
            ))?;
        }
        // SAFETY:
        //  *  `lmdb::mdb_env_set_maxreaders` doesn't have any requirements
        //     (other than being called before `mdb_env_open` which is the case
        //     and would otherwise only result in EINVAL being returned)
        //  *  the status passed to `check_err_code` is valid
        unsafe {
            check_err_code(lmdb::mdb_env_set_maxreaders(
                backend.inner,
                self.max_readers.try_into().map_err(|_| {
                    io::Error::new(io::ErrorKind::InvalidInput, "maximum reader count invalid")
                })?,
            ))?;
        }
        // SAFETY:
        //  *  `lmdb::mdb_env_set_maxdbs` doesn't have any requirements
        //     (other than being called before `mdb_env_open` which is the case
        //     and would otherwise only result in EINVAL being returned)
        //  *  the status passed to `check_err_code` is valid
        unsafe {
            check_err_code(lmdb::mdb_env_set_maxdbs(
                backend.inner,
                self.max_dbs.try_into().map_err(|_| {
                    io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "maximum number of named databases invalid",
                    )
                })?,
            ))?;
        }
        // SAFETY:
        //  *  safety requirements are documented in `EnvBuilder::env_ro` and
        //     `EnvBuilder::env_rw`
        //  *  the status passed to `check_err_code` is valid
        unsafe {
            check_err_code(lmdb::mdb_env_open(
                backend.inner,
                path.as_ptr(),
                if readonly {
                    self.lmdb_flags | lmdb::MDB_RDONLY as LmdbFlags
                } else {
                    self.lmdb_flags
                },
                self.unix_file_mode.try_into().map_err(|_| {
                    io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "file mode bits integer invalid",
                    )
                })?,
            ))?;
        }
        Ok(EnvRo {
            backend: Arc::new(backend),
        })
    }
    /// Open environment in read-only mode
    ///
    /// # Safety
    ///
    /// * Do not open the same environment twice in the same process at the
    ///   same time.
    /// * The environment must be free of corruption on the storage medium and
    ///   no other processes may corrupt the environment on the storage medium.
    /// * The file format is platform specific, so endianness and word size
    ///   must not change across use.
    pub unsafe fn open_ro(&self) -> io::Result<EnvRo> {
        // SAFETY: requirements of `EnvBuilder::open_with_readonly_flag` are
        // the same as documented by `EnvBuilder::open_ro`
        Ok(unsafe { self.open_with_readonly_flag(true) }?)
    }
    /// Open environment in read-write mode
    ///
    /// # Safety
    ///
    /// * Do not open the same environment twice in the same process at the
    ///   same time.
    /// * The environment must be free of corruption on the storage medium and
    ///   no other processes may corrupt the environment on the storage medium.
    /// * The file format is platform specific, so endianness and word size
    ///   must not change across use.
    pub unsafe fn open_rw(&self) -> io::Result<EnvRw> {
        // SAFETY: requirements of `EnvBuilder::open_with_readonly_flag` are
        // the same as documented by `EnvBuilder::open_rw`
        Ok(EnvRw {
            env_ro: unsafe { self.open_with_readonly_flag(false) }?,
        })
    }
}

impl DbBuilder<[u8], [u8], KeysUnique, ()> {
    /// Default options for database
    pub const fn new() -> Self {
        Self {
            key: PhantomData,
            value: PhantomData,
            constraint: KeysUnique,
            lmdb_flags: 0,
            name: (),
        }
    }
}

impl<K: ?Sized, V: ?Sized, C> DbBuilder<K, V, C, ()>
where
    C: Constraint,
{
    /// Clear `dupsort` option (also clears `reversedup` option)
    ///
    /// See [`DbBuilder::keys_duplicate`] for setting `dupsort` option and
    /// [`DbBuilder::has_duplicate_keys`] for getting current state of flag.
    pub const fn keys_unique(mut self) -> DbBuilder<K, V, KeysUnique, ()> {
        flag_set!(self, MDB_DUPSORT, false);
        flag_set!(self, MDB_REVERSEDUP, false);
        DbBuilder {
            constraint: KeysUnique,
            key: PhantomData,
            value: PhantomData,
            lmdb_flags: self.lmdb_flags,
            name: self.name,
        }
    }
    /// Set `dupsort` option
    ///
    /// Note: Using this option will put an additional constraint on the size
    /// of each value stored in the database (see [`Env::max_keysize`]).
    ///
    /// See [`DbBuilder::keys_unique`] for clearing `dupsort` option and
    /// [`DbBuilder::has_duplicate_keys`] for getting current state of flag.
    pub const fn keys_duplicate(mut self) -> DbBuilder<K, V, KeysDuplicate, ()> {
        flag_set!(self, MDB_DUPSORT, true);
        DbBuilder {
            constraint: KeysDuplicate,
            key: PhantomData,
            value: PhantomData,
            lmdb_flags: self.lmdb_flags,
            name: self.name,
        }
    }
    /// Set stored key type
    ///
    /// It's possible to use [`NoKey`] as type parameter for databases which
    /// contain only a single value.
    pub const fn key_type<T>(mut self) -> DbBuilder<T, V, C, ()>
    where
        T: ?Sized + Storable,
    {
        flag_set!(self, MDB_INTEGERKEY, T::OPTIMIZE_INT);
        DbBuilder {
            key: PhantomData,
            value: PhantomData,
            constraint: self.constraint,
            lmdb_flags: self.lmdb_flags,
            name: self.name,
        }
    }
    /// Set stored value type
    ///
    /// It's possible to use [`NoValue`] as type parameter for databases which
    /// only contain keys but no values.
    pub const fn value_type<T>(mut self) -> DbBuilder<K, T, C, ()>
    where
        T: ?Sized + Storable,
    {
        flag_set!(self, MDB_DUPFIXED, T::CONST_BYTES_LEN);
        flag_set!(self, MDB_INTEGERDUP, T::OPTIMIZE_INT);
        DbBuilder {
            value: PhantomData,
            key: PhantomData,
            constraint: self.constraint,
            lmdb_flags: self.lmdb_flags,
            name: self.name,
        }
    }
}

impl<K: ?Sized, V: ?Sized, C, N> DbBuilder<K, V, C, N>
where
    C: Constraint,
{
    /// Get `reversekey` option
    pub const fn get_reversekey(&self) -> bool {
        flag_get!(self, MDB_REVERSEKEY)
    }
    /// Set or clear `reversekey` option
    pub const fn reversekey(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_REVERSEKEY, flag);
        self
    }
    /// Return sanitized flags for LMDB
    /// (clear flags that are only used for for databases with duplicate keys
    /// if unique keys have been selected)
    const fn cleansed_lmdb_flags(&self) -> LmdbFlags {
        if flag_get!(self, MDB_DUPSORT) {
            self.lmdb_flags
        } else {
            self.lmdb_flags
                & !((lmdb::MDB_REVERSEDUP | lmdb::MDB_DUPFIXED | lmdb::MDB_INTEGERDUP) as LmdbFlags)
        }
    }
    /// Use unnamed database
    ///
    /// Should be used as last builder method because some other builder
    /// methods are not available anymore after calling this method.
    pub fn unnamed(self) -> DbSpec<K, V, C> {
        DbBuilder {
            value: PhantomData,
            key: PhantomData,
            constraint: self.constraint,
            lmdb_flags: self.lmdb_flags,
            name: None,
        }
    }
    /// Use named database (name must not contain a null byte)
    ///
    /// Should be used as last builder method because some other builder
    /// methods are not available anymore after calling this method.
    pub fn name<T: Into<Vec<u8>>>(self, name: T) -> DbSpec<K, V, C> {
        let name = CString::new(name).expect("database name must not contain NULL byte");
        DbBuilder {
            value: PhantomData,
            key: PhantomData,
            constraint: self.constraint,
            lmdb_flags: self.lmdb_flags,
            name: Some(name),
        }
    }
    /// Remove name information
    ///
    /// NOTE: This does not select the unnamed database. To select the unnamed
    /// database, use [`DbBuilder::unnamed`].
    pub fn strip_name<T: Into<Vec<u8>>>(self) -> DbBuilder<K, V, C, ()> {
        DbBuilder {
            value: PhantomData,
            key: PhantomData,
            constraint: self.constraint,
            lmdb_flags: self.lmdb_flags,
            name: (),
        }
    }
}

impl<K: ?Sized, V: ?Sized, C: Constraint, N> DbBuilder<K, V, C, N> {
    /// Returns true if duplicate keys are allowed (i.e. is `dupsort` option set?)
    ///
    /// See [`DbBuilder::keys_unique`] and [`DbBuilder::keys_duplicate`].
    pub fn has_duplicate_keys(&self) -> bool {
        C::DUPLICATE_KEYS
    }
}

impl<K: ?Sized, V: ?Sized, N> DbBuilder<K, V, KeysDuplicate, N> {
    /// Get `reversedup` option
    pub const fn get_reversedup(&self) -> bool {
        flag_get!(self, MDB_REVERSEDUP)
    }
    /// Set or clear `reversedup` option
    pub const fn reversedup(mut self, flag: bool) -> Self {
        flag_set!(self, MDB_REVERSEDUP, flag);
        self
    }
}

/// Function passed to [`lmdb::mdb_set_compare`] to compare two keys or values
/// in the database
///
/// # Safety
///
/// * Pointers `a` and `b` must be valid.
/// * For each `a` and `b`, [`lmdb::MDB_val::mv_data`] must be valid for reads
///   for the corresponding [`lmdb::MDB_val::mv_size`] bytes (no alignment
///   required and `mv_size` may differ for `a` and `b`) until the function
///   returns.
/// * The pointed to bytes must have been created with
///   [`<T as Storable>::to_bytes`](Storable::to_bytes) on the same platform
///   (same endianess / word size).
///
/// Note: Regardless of whether it is considered to be UB to unwind from an
/// `extern "C" fn`, this function can never unwind if implementors of
/// [`Storable::cmp_bytes_unchecked`] ensure that `cmp_bytes_unchecked` never
/// unwinds (which is demanded by the corresponding `unsafe` trait
/// documentation).
unsafe extern "C" fn compare_function<T>(a: *const lmdb::MDB_val, b: *const lmdb::MDB_val) -> c_int
where
    T: ?Sized + Storable,
{
    // SAFETY: `a` and `b` fulfill the documented requirements of
    // `compare_function`
    unsafe {
        let a: &[u8] = slice::from_raw_parts((*a).mv_data as *const u8, (*a).mv_size);
        let b: &[u8] = slice::from_raw_parts((*b).mv_data as *const u8, (*b).mv_size);
        T::cmp_bytes_unchecked(a, b) as c_int
    }
}

impl EnvRo {
    /// Return true if given size (in bytes) is valid for a key (or value in a
    /// database with duplicate keys)
    fn size_valid_keysize(&self, size: usize) -> bool {
        (1..=self.max_keysize()).contains(&size)
    }
    /// Return true if given slice has valid size/length as key (or value in a
    /// database with duplicate keys)
    fn slice_valid_keysize(&self, key: &[u8]) -> bool {
        self.size_valid_keysize(key.len())
    }
    /// Return error if slice doesn't have valid size/length as key
    fn assert_valid_keysize(&self, key: &[u8]) -> io::Result<()> {
        if self.slice_valid_keysize(key) {
            Ok(())
        } else {
            Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "invalid key size",
            ))
        }
    }
    /// Return error if slice doesn't have valid size/length as value in a
    /// database with duplicate keys
    fn assert_valid_valuesize(&self, value: &[u8]) -> io::Result<()> {
        if self.slice_valid_keysize(value) {
            Ok(())
        } else {
            Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "invalid value size",
            ))
        }
    }
    /// Opens several databases at once (which must be of the same key and
    /// value type and key constraint)
    ///
    /// # Safety
    ///
    /// * If a database exists already, it must have been created with
    /// compatible options.
    /// * Implementation of [`Storable`] for the used
    ///   [keys](DbBuilder::key_type) and [values](DbBuilder::value_type) must
    ///   behave the same as when the database was created.
    /// * If `create` is true, then this method call must be synchronized and
    ///   no write transaction must be currently open by the current thread.
    unsafe fn open_or_opt_create_db<'a, K, V, C>(
        &self,
        db_spec: &DbSpec<K, V, C>,
        create: bool,
    ) -> io::Result<Db<K, V, C>>
    where
        K: ?Sized + Storable + 'a,
        V: ?Sized + Storable + 'a,
        C: Constraint,
    {
        let mut dbis = self.backend.dbis.lock().unwrap();
        let mut txn_inner = MaybeUninit::<*mut lmdb::MDB_txn>::uninit();
        // SAFETY:
        //  *  It's allowed to open several read-only transactions (`create` is
        //     `false`).
        //  *  If `create` is true, then documentation requires that this
        //     method is synchronized and no write transaction must be
        //     currently open by the current thread.
        //  *  The transaction is only used by the current thread (only within
        //     this function as `TxnBackend` is dropped before this function
        //     returns or panics).
        //  *  There are no cursors created here which could be used by other
        //     threads or span transactions.
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_txn_begin(
                self.backend.inner,
                null_mut(),
                if create {
                    0 as LmdbFlags
                } else {
                    lmdb::MDB_RDONLY as LmdbFlags
                },
                txn_inner.as_mut_ptr(),
            ))?;
        }
        // SAFETY: call to `lmdb::mdb_txn_begin` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        let txn_inner = unsafe { txn_inner.assume_init() };
        let txn = TxnBackend {
            env_ro: self,
            inner: txn_inner,
            cursors: Default::default(),
        };
        let mut db_inner = MaybeUninit::<lmdb::MDB_dbi>::uninit();
        let mut flags = db_spec.cleansed_lmdb_flags();
        if create {
            flags |= lmdb::MDB_CREATE as LmdbFlags;
        }
        // SAFETY:
        //  *  Mutex `EnvBackend::dbis` is locked and ensures that
        //     `lmdb::mdb_dbi_close` is not called concurrently. This is
        //     important because we must determine below if the returned
        //     `lmdb::MDB_dbi` integer belongs to an already open database,
        //     in which case it must be ensured that it is only closed
        //     once.
        //  *  The database handle, if a new one, can only be used by other
        //     transactions after the corresponding transaction is
        //     successfully committed below.
        //  *  If the transaction is aborted, `lmdb::mdb_dbi_close` will not be
        //     called because the database handle will be closed automatically.
        //  *  If and only if the transaction is committed (and the
        //     `lmdb::MDB_dbi` handle is a new handle), a `DbBackend` is
        //     created which will take care of closing the database handle when
        //     dropped.
        //  *  `lmdb::mdb_dbi_open` is only called here and the locked mutex
        //     `EnvBackend::dbis` ensures that `lmdb::mdb_dbi_open` is not
        //     called from multiple concurrent transactions in the same process
        //     (for the same environment).
        //  *  Documentation of this function (`open_or_opt_create_db`) demands
        //     that if a database exists already, it must have been created
        //     with compatible options. Documentation of `EnvBuilder::open_ro`
        //     and `EnvBuilder::open_rw` makes additional demands in regard to
        //     the database being free of corruption and having been
        //     created on a compatible platform. In addition with the
        //     requirements for implementing the unsafe trait `Storable` for
        //     key and value types, as well as invoking
        //     `DbBuilder::cleansed_lmdb_flags` above, it is ensured that the
        //     `flags` are valid and that they match any previous creation of
        //     the database.
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_dbi_open(
                txn.inner,
                db_spec.name.as_ref().map_or(null(), |x| x.as_ptr()),
                flags,
                db_inner.as_mut_ptr(),
            ))?;
        }
        // SAFETY: call to `lmdb::mdb_dbi_open` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        let db_inner = unsafe { db_inner.assume_init() };
        let backend = match dbis.get(&db_inner).and_then(Weak::upgrade) {
            Some(arc) => {
                txn.commit()?;
                arc
            }
            None => {
                // SAFETY:
                //  *  Functions `mdb_set_compare` and `mdb_set_dupsort` are
                //     always called before any data access through the
                //     database handle can happen.
                //  *  Documentation of this function (`open_or_opt_create_db`)
                //     demands that if a database exists already, it must have
                //     been created with compatible options. Documentation of
                //     `EnvBuilder::open_ro` and `EnvBuilder::open_rw` makes
                //     additional demands in regard to the database being free
                //     of corruption and having been created on a compatible
                //     platform. This ensures that `mdb_set_compare` and
                //     `mdb_set_dupsort` are set consistently in regard to when
                //     the database was created.
                //  *  `mdb_set_compare` and `mdb_set_dupsort` are expected to
                //     ensure that only valid pointers are passed to the
                //     `compare_function`.
                //  *  `compare_function` does not unwind. (If it would unwind,
                //     it could leave the environment in an invalid state.
                //     TODO: Adjust this remark, the corresponding comment on
                //     `compare_function`, and the `Storable` trait when/if
                //     Rust implements an abort-on-panic shim for
                //     `extern "C" fn`.)
                //  *  The status passed to `check_err_code` is valid.
                unsafe {
                    if !K::TRIVIAL_CMP && !K::OPTIMIZE_INT {
                        check_err_code(lmdb::mdb_set_compare(
                            txn.inner,
                            db_inner,
                            Some(compare_function::<K>),
                        ))?;
                    }
                    if !V::TRIVIAL_CMP && !V::OPTIMIZE_INT && C::DUPLICATE_KEYS {
                        check_err_code(lmdb::mdb_set_dupsort(
                            txn.inner,
                            db_inner,
                            Some(compare_function::<V>),
                        ))?;
                    }
                }
                let env_backend = Arc::downgrade(&self.backend);
                let env_id = self.backend.id;
                txn.commit()?;
                let db_backend = DbBackend {
                    env_backend,
                    env_id,
                    inner: db_inner,
                };
                let arc = Arc::new(db_backend);
                dbis.insert(db_inner, Arc::downgrade(&arc));
                arc
            }
        };
        let db = Db {
            key: PhantomData,
            value: PhantomData,
            constraint: db_spec.constraint,
            backend: ArcByAddr::from(backend),
        };
        drop(dbis);
        Ok(db)
    }
}

/// Read-write or read-only handle for accessing environment that stores
/// key-value databases
///
/// This trait is implemented by [`EnvRo`] and [`EnvRw`] and provides the
/// methods for read-only access to the environment. Read-write access is
/// performed by methods directly implemented on `EnvRw`.
pub trait Env: AsRef<EnvRo> {
    /// Reference conversion to [`EnvRo`]
    ///
    /// To obtain an owned [`EnvRo`], you can use [`Env::clone_ro`] as a
    /// shorthand for `.as_env_ro().clone()`.
    fn as_env_ro(&self) -> &EnvRo {
        self.as_ref()
    }
    /// Clone as [`EnvRo`]
    fn clone_ro(&self) -> EnvRo {
        self.as_env_ro().clone()
    }
    /// Start read-only transaction
    fn txn_ro(&self) -> io::Result<TxnRo<'_>>;
    /// Get maximum size of keys and duplicate data
    fn max_keysize(&self) -> usize;
    /// Checks if key or duplicate data has valid size
    fn valid_keysize<'kr, K, KRef>(&self, key: KRef) -> bool
    where
        K: ?Sized + Storable + 'kr,
        KRef: StorableRef<'kr, K>;
    /// Open database in environment
    ///
    /// # Safety
    ///
    /// * If a database exists already, it must have been created with
    /// compatible options.
    /// * Implementation of [`Storable`] for the used
    ///   [keys](DbBuilder::key_type) and [values](DbBuilder::value_type) must
    ///   behave the same as when the database was created.
    unsafe fn open_db<K, V, C>(&self, options: &DbSpec<K, V, C>) -> io::Result<Db<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint;
    /// Clear stale readers
    ///
    /// Refer to LMDB's documentation when to clear stale readers
    fn clear_stale_readers(&self) -> io::Result<usize>;
}

impl Env for EnvRo {
    fn txn_ro(&self) -> io::Result<TxnRo<'_>> {
        unsafe {
            let mut inner_txn = MaybeUninit::<*mut lmdb::MDB_txn>::uninit();
            // SAFETY:
            //  *  `mdb_txn_begin` doesn't have any special requirements for
            //     read-only transactions because `lmdb::MDB_NOTLS` is set
            //     (and no child transactions are created for read-only
            //     transactions either).
            //  *  The status passed to `check_err_code` is valid.
            check_err_code(lmdb::mdb_txn_begin(
                self.backend.inner,
                null_mut(),
                lmdb::MDB_RDONLY as LmdbFlags,
                inner_txn.as_mut_ptr(),
            ))?;
            // SAFETY: call to `lmdb::mdb_txn_begin` above has been successful
            // as otherwise `check_err_code` would have returned an `Err`
            // resulting in early return from this function
            let inner_txn = inner_txn.assume_init();
            Ok(TxnRo {
                backend: UnsafeCell::new(TxnBackend {
                    env_ro: &self,
                    inner: inner_txn,
                    cursors: Default::default(),
                }),
            })
        }
    }
    fn max_keysize(&self) -> usize {
        self.backend.max_keysize
    }
    fn valid_keysize<'kr, K, KRef>(&self, key: KRef) -> bool
    where
        K: ?Sized + Storable + 'kr,
        KRef: StorableRef<'kr, K>,
    {
        self.size_valid_keysize(key.ref_bytes_len())
    }
    unsafe fn open_db<K, V, C>(&self, options: &DbSpec<K, V, C>) -> io::Result<Db<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        // SAFETY:
        //  *  Documentation of `Env::open_db` demands that if the database
        //     exists already, it must have been created with compatible
        //     options.
        //  *  Second argument of `open_or_opt_create_db` is set to `false`.
        unsafe { self.open_or_opt_create_db(options, false) }
    }
    fn clear_stale_readers(&self) -> io::Result<usize> {
        unsafe {
            let mut dead = MaybeUninit::<c_int>::uninit();
            // SAFETY:
            //  *  `mdb_reader_check` doesn't have any special requirements
            //  *  status passed to `check_err_code` is valid
            check_err_code(lmdb::mdb_reader_check(
                self.backend.inner,
                dead.as_mut_ptr(),
            ))?;
            // SAFETY: call to `lmdb::mdb_reader_check` above has been
            // successful as otherwise `check_err_code` would have returned an
            // `Err` resulting in early return from this function
            let dead = dead.assume_init();
            Ok(dead
                .try_into()
                .expect("reported number of stale reader transactions does not fit into usize"))
        }
    }
}

impl Env for EnvRw {
    fn txn_ro(&self) -> io::Result<TxnRo<'_>> {
        self.as_env_ro().txn_ro()
    }
    fn max_keysize(&self) -> usize {
        self.as_env_ro().max_keysize()
    }
    fn valid_keysize<'kr, K, KRef>(&self, key: KRef) -> bool
    where
        K: ?Sized + Storable + 'kr,
        KRef: StorableRef<'kr, K>,
    {
        self.as_env_ro().valid_keysize(key)
    }
    unsafe fn open_db<K, V, C>(&self, options: &DbSpec<K, V, C>) -> io::Result<Db<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        // SAFETY: calls same trait method which has the same requirements
        unsafe { self.as_env_ro().open_db(options) }
    }
    fn clear_stale_readers(&self) -> io::Result<usize> {
        self.as_env_ro().clear_stale_readers()
    }
}

impl EnvRw {
    /// Check if transactions are nestable
    ///
    /// See [`TxnRw::nested`].
    pub fn nestable_txns(&self) -> bool {
        self.env_ro.backend.nestable_txns
    }
    /// Open database in environment and create if non-existing
    ///
    /// # Safety
    ///
    /// * If a database exists already, it must have been created with
    /// compatible options.
    /// * Implementation of [`Storable`] for the used
    ///   [keys](DbBuilder::key_type) and [values](DbBuilder::value_type) must
    ///   behave the same as when the database was created.
    pub unsafe fn create_db<K, V, C>(
        &mut self,
        options: &DbSpec<K, V, C>,
    ) -> io::Result<Db<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        // SAFETY:
        //  *  Documentation of `EnvRw::create_db` demands that if the database
        //     exists already, it must have been created with compatible
        //     options.
        //  *  Calling `EnvRo::open_or_opt_create_db` is synchronized due to
        //     `&mut self` reference.
        //  *  No write transaction can be open because `EnvRw::txn_rw` also
        //     requires `&mut self`.
        unsafe { self.env_ro.open_or_opt_create_db(options, true) }
    }
    /// Delete database in environment
    ///
    /// This method panics if there are clones of the [`Db`] handle.
    /// Use [`TxnRw::delete_all`] to only delete the contents of the database.
    pub fn drop_db<K, V, C>(&mut self, db: Db<K, V, C>) -> io::Result<()>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        let db_backend = ArcByAddr::try_unwrap(db.backend).map_err(|_| {
            io::Error::new(io::ErrorKind::Other, "database in use by current process")
        })?;
        db_backend.assert_env_backend(&self.env_ro.backend);
        let dbis = self.env_ro.backend.dbis.lock().unwrap();
        unsafe {
            let mut txn_inner = MaybeUninit::<*mut lmdb::MDB_txn>::uninit();
            // SAFETY:
            //  *  `&mut self` ensures that no other write transaction (for the
            //     given environment) can be open by the current thread (and
            //     whole process).
            //  *  The transaction is only used by the current thread because
            //     it is only used within this function.
            //  *  There are no cursors created here which could be used by other
            //     threads or span transactions.
            //  *  The status passed to `check_err_code` is valid.
            check_err_code(lmdb::mdb_txn_begin(
                self.env_ro.backend.inner,
                null_mut(),
                0 as LmdbFlags,
                txn_inner.as_mut_ptr(),
            ))?;
            // SAFETY: call to `lmdb::mdb_txn_begin` above has been successful
            // as otherwise `check_err_code` would have returned an `Err`
            // resulting in early return from this function
            let txn_inner = txn_inner.assume_init();
            let txn = TxnBackend {
                env_ro: &self.env_ro,
                inner: txn_inner,
                cursors: Default::default(),
            };
            let db_inner = db_backend.inner;
            db_backend.dont_close();
            // SAFETY:
            //  *  `&mut self` ensures that `lmdb::mdb_drop` is only called by
            //     a single thread at the same time (for the given
            //     environment), because no write transaction can be open.
            //  *  No other threads can use the database or any of its cursors
            //     in a read-only transaction because the `DbBackend` was
            //     successfully unwrapped from the `Arc` above.
            //  *  No other write transaction which could have modified the
            //     database can be open because of holding a `&mut self` which
            //     ensures that no other write transaction is open.
            //  *  The above call of `DbBackend::dont_close` ensures that the
            //     database handle isn't closed twice.
            //  *  The status passed to `check_err_code` is valid.
            check_err_code(lmdb::mdb_drop(txn.inner, db_inner, 1))?;
            txn.commit()?;
        }
        drop(dbis);
        Ok(())
    }
    /// Start read-write transaction
    pub fn txn_rw(&mut self) -> io::Result<TxnRw<'_>> {
        let mut txn_inner = MaybeUninit::<*mut lmdb::MDB_txn>::uninit();
        // SAFETY:
        //  *  `&mut self` ensures that no other write transaction (for the
        //     given environment) can be open by the current thread (and
        //     whole process).
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_txn_begin(
                self.env_ro.backend.inner,
                null_mut(),
                0,
                txn_inner.as_mut_ptr(),
            ))?;
        }
        // SAFETY: call to `lmdb::mdb_txn_begin` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        let txn_inner = unsafe { txn_inner.assume_init() };
        Ok(TxnRw {
            backend: UnsafeCell::new(TxnBackend {
                env_ro: &self.env_ro,
                inner: txn_inner,
                cursors: Default::default(),
            }),
            used_dbs: UsedDbs::Toplevel(Default::default()),
            commit_handlers: Default::default(),
        })
    }
}

impl<'a> TxnBackend<'a> {
    /// Commit transaction
    fn commit(self) -> io::Result<()> {
        let inner = self.inner;
        self.dont_abort();
        // SAFETY:
        //  *  transaction cannot be used further
        //  *  all cursors have been closed by `TxnBackend::dont_abort(self)`
        //  *  the status passed to `check_err_code` is valid
        unsafe {
            check_err_code(lmdb::mdb_txn_commit(inner))?;
        }
        Ok(())
    }
    /// Get reference to value in database
    ///
    /// # Safety
    ///
    /// * The value is only valid until end of a read-only transaction or, in
    ///   case of a read-write transaction, until the transaction ends or
    ///   modifies data. The caller has to ensure that the lifetime `'b` is
    ///   short enough.
    unsafe fn get_unsafe<'b, 'kr, K, V, C, KRef>(
        &mut self,
        db: &Db<K, V, C>,
        key: KRef,
    ) -> io::Result<Option<V::AlignedRef<'b>>>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
    {
        db.backend.assert_env_backend(&self.env_ro.backend);
        let key: K::BytesRef<'_> = key.ref_to_bytes();
        let lmdb_key = lmdb::MDB_val {
            mv_size: key.len(),
            mv_data: key.as_ptr() as *mut _,
        };
        let mut lmdb_data = MaybeUninit::<lmdb::MDB_val>::uninit();
        Ok(
            // SAFETY:
            //  *  `lmdb_key.mv_data` is a valid pointer to `lmdb_key.mv_size`
            //     bytes of memory for reading.
            //  *  The memory pointed to by `lmdb_data.mv_data` is only read.
            //  *  Documentation requires that `'b` must not live longer than
            //     the end of the read-only transaction or until the read-write
            //     transaction ends or modifies data. The lifetime of the
            //     reference returned by `slice::from_raw_parts` below is
            //     inferred as being `'b`. Thus the memory won't be accessed
            //     afterwards.
            match unsafe {
                lmdb::mdb_get(
                    self.inner,
                    db.backend.inner,
                    &lmdb_key as *const _ as *mut lmdb::MDB_val,
                    lmdb_data.as_mut_ptr(),
                )
            } {
                lmdb::MDB_NOTFOUND => None,
                status => {
                    // SAFETY: `status` is a valid LMDB return code, and if
                    // negative, it is a valid LMDB error code
                    unsafe { check_err_code(status) }?;
                    // SAFETY: call to `lmdb::mdb_get` above has been
                    // successful as otherwise `check_err_code` would have
                    // returned an `Err` resulting in early return from this
                    // function
                    let lmdb_data = unsafe { lmdb_data.assume_init() };
                    // SAFETY:
                    //  *  The call to `lmdb::mdb_get` has been successful and
                    //     is expected to have written valid data to
                    //     `lmdb_data` (see SAFETY comment on `lmdb::mdb_get`
                    //     above regarding lifetimes).
                    //  *  Documentation of all methods opening databases
                    //     demand that if a database exists already, it must
                    //     have been created with compatible options.
                    //     Documentation of `EnvBuilder::open_ro` and
                    //     `EnvBuilder::open_rw` makes additional demands in
                    //     regard to the database being free of corruption and
                    //     having been created on a compatible platform. It is
                    //     thus safe to call `Storable::from_bytes_unchecked`.
                    Some(unsafe {
                        V::from_bytes_unchecked(slice::from_raw_parts(
                            lmdb_data.mv_data as *const u8,
                            lmdb_data.mv_size,
                        ))
                    })
                }
            },
        )
    }
    /// Create a new cursor
    fn new_cursor<K, V, C>(&mut self, db: &Db<K, V, C>) -> io::Result<Cursor<K, V, C>>
    where
        K: ?Sized,
        V: ?Sized,
        C: Constraint,
    {
        db.backend.assert_env_backend(&self.env_ro.backend);
        let mut inner_cursor = MaybeUninit::<*mut lmdb::MDB_cursor>::uninit();
        // SAFETY:
        //  *  Database handles are only closed when the `DbBackend` or
        //     `EnvBackend` is dropped. Because the `Cursor` struct contains a
        //     clone of the `Db` handle (after the `CursorBackend`, see SAFETY
        //     comment on `Cursor`), the database handle won't be closed while
        //     the cursor may be used.
        //  *  The cursor is not used after the transaction has ended because
        //     this modules API always requires a `TxnRo` or `TxnRw` for using
        //     the cursor. In regard to closing the cursor, see SAFETY comments
        //     where `lmdb::mdb_cursor_close` is invoked.
        unsafe {
            check_err_code(lmdb::mdb_cursor_open(
                self.inner,
                db.backend.inner,
                inner_cursor.as_mut_ptr(),
            ))?;
        }
        // SAFETY: call to `lmdb::mdb_cursor_open` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        let inner_cursor = unsafe { inner_cursor.assume_init() };
        // SAFETY: clone `db` first, see `CursorBackend::drop`
        let db = db.clone();
        let cursor_backend = Arc::new(CursorBackend {
            inner: inner_cursor,
            closed: Default::default(),
            inner_txn: self.inner,
        });
        self.cursors.push(Arc::downgrade(&cursor_backend));
        Ok(Cursor {
            backend: ArcByAddr::from(cursor_backend),
            db,
        })
    }
    /// Execute cursor operation
    ///
    /// The returned key or value may be retrieved by calling an `FnOnce`
    /// closure when needed.
    ///
    /// # Safety
    ///
    /// * Any returned keys or values are only valid until end of a
    ///  read-only transaction or until a read-write transaction ends or
    ///  modifies data. The caller has to ensure that the lifetime `'b` is
    ///  short enough.
    unsafe fn cursor_op_unsafe<'b, 'kr, 'vr, K, V, C, KRef, VRef>(
        &mut self,
        cursor: &Cursor<K, V, C>,
        key: Option<KRef>,
        value: Option<VRef>,
        op: lmdb::MDB_cursor_op,
    ) -> io::Result<
        Option<(
            impl FnOnce() -> K::AlignedRef<'b>,
            impl FnOnce() -> V::AlignedRef<'b>,
        )>,
    >
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + 'vr,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
        VRef: StorableRef<'vr, V>,
    {
        cursor.backend.assert_txn_backend(self);
        let mut lmdb_key = match key {
            Some(key) => {
                let key: K::BytesRef<'_> = key.ref_to_bytes();
                lmdb::MDB_val {
                    mv_size: key.len(),
                    mv_data: key.as_ptr() as *mut _,
                }
            }
            None => lmdb::MDB_val {
                mv_size: 0,
                mv_data: null_mut(),
            },
        };
        let mut lmdb_data = match value {
            Some(value) => {
                let value: V::BytesRef<'_> = value.ref_to_bytes();
                lmdb::MDB_val {
                    mv_size: value.len(),
                    mv_data: value.as_ptr() as *mut _,
                }
            }
            None => lmdb::MDB_val {
                mv_size: 0,
                mv_data: null_mut(),
            },
        };
        Ok(
            // SAFETY:
            //  *  `lmdb_key.mv_data` and `lmdb_data.mv_data` are valid
            //     pointers to `lmdb_key.mv_size` and `lmdb_data.mv_size`
            //     bytes of memory, respectively, for reading.
            //  *  The memory pointed to by `lmdb_key.mv_data` and
            //     `lmdb_data.mv_data` is only read subsequently.
            //  *  Documentation requires that `'b` must not live longer than
            //     the end of the read-only transaction or until the read-write
            //     transaction ends or modifies data. The lifetime of the
            //     reference returned by `slice::from_raw_parts` below is
            //     inferred as being `'b`. Thus the memory won't be accessed
            //     afterwards.
            match unsafe {
                lmdb::mdb_cursor_get(cursor.backend.inner, &mut lmdb_key, &mut lmdb_data, op)
            } {
                lmdb::MDB_NOTFOUND => None,
                status => {
                    // SAFETY: `status` is a valid LMDB return code, and if
                    // negative, it is a valid LMDB error code
                    unsafe { check_err_code(status) }?;
                    // SAFETY:
                    //  *  The call to `lmdb::mdb_cursor_get` has been
                    //     successful and is expected to either have written
                    //     valid data to `lmdb_key` and `lmdb_data` or to not
                    //     have written anything to one or both of them (see
                    //     SAFETY comment on `lmdb::mdb_cursor_get` above
                    //     regarding lifetimes).
                    //  *  Documentation of all methods opening databases
                    //     demand that if a database exists already, it must
                    //     have been created with compatible options.
                    //     Documentation of `EnvBuilder::open_ro` and
                    //     `EnvBuilder::open_rw` makes additional demands in
                    //     regard to the database being free of corruption and
                    //     having been created on a compatible platform. It is
                    //     thus safe to call `Storable::from_bytes_unchecked`.
                    Some(unsafe {
                        (
                            move || {
                                K::from_bytes_unchecked(slice::from_raw_parts(
                                    lmdb_key.mv_data as *const u8,
                                    lmdb_key.mv_size,
                                ))
                            },
                            move || {
                                V::from_bytes_unchecked(slice::from_raw_parts(
                                    lmdb_data.mv_data as *const u8,
                                    lmdb_data.mv_size,
                                ))
                            },
                        )
                    })
                }
            },
        )
    }
    /// Get number of values for current cursor position
    fn cursor_get_current_value_count<K, V>(
        &mut self,
        cursor: &Cursor<K, V, KeysDuplicate>,
    ) -> io::Result<usize>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
    {
        cursor.backend.assert_txn_backend(self);
        let mut count = MaybeUninit::<usize>::uninit();
        // SAFETY:
        //  *  `lmdb::mdb_cursor_count` is only called on databases using
        //     `lmdb::MDB_DUPSORT` (because the third type parameter of
        //     `cursor` is `KeysDuplicate`).
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_cursor_count(
                cursor.backend.inner,
                count.as_mut_ptr(),
            ))?;
        }
        // SAFETY: call to `lmdb::mdb_cursor_count` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        Ok(unsafe { count.assume_init() })
    }
}

/// Helper macro to create return type of cursor methods
macro_rules! cursor_return_type {
    (bool $(,)?) => (bool);
    (key $(,)?) => (Option<K::AlignedRef<'_>>);
    (value $(,)?) => (Option<V::AlignedRef<'_>>);
    ((key, value $(,)?)) => (Option<(K::AlignedRef<'_>, V::AlignedRef<'_>)>);
}

/// Create signature for cursor method
macro_rules! cursor_method_def {
    ($($meta:meta)*, $name:ident, (), $retval:tt $(,)?) => {
            $(#[$meta])*
            fn $name<K, V, C>(
                &self,
                cursor: &Cursor<K, V, C>,
            ) -> io::Result<cursor_return_type!($retval)>
            where
                K: ?Sized + Storable,
                V: ?Sized + Storable,
                C: Constraint;
    };
    ($($meta:meta)*, $name:ident, (key), $retval:tt $(,)?) => {
            $(#[$meta])*
            fn $name<'kr, K, V, C, KRef>(
                &self,
                cursor: &Cursor<K, V, C>,
                key: KRef,
            ) -> io::Result<cursor_return_type!($retval)>
            where
                K: ?Sized + Storable + 'kr,
                V: ?Sized + Storable,
                C: Constraint,
                KRef: StorableRef<'kr, K>;
    };
    ($($meta:meta)*, $name:ident, (key, value), $retval:tt $(,)?) => {
            $(#[$meta])*
            fn $name<'kr, 'vr, K, V, C, KRef, VRef>(
                &self,
                cursor: &Cursor<K, V, C>,
                key: KRef,
                value: VRef,
            ) -> io::Result<cursor_return_type!($retval)>
            where
                K: ?Sized + Storable + 'kr,
                V: ?Sized + Storable + 'vr,
                C: Constraint,
                KRef: StorableRef<'kr, K>,
                VRef: StorableRef<'vr, V>;
    };
}

/// Create signatures for several cursor methods
macro_rules! cursor_method_defs {
    {
        $(
            $(#[$meta:meta])*
            fn $name:ident $args:tt -> $retval:tt;
        )*
    } => {
        $(
            cursor_method_def!($($meta)*, $name, $args, $retval);
        )*
    };
}

/// Create signature for cursor method on databases with duplicate keys
macro_rules! cursor_method_dupkey_def {
    ($($meta:meta)*, $name:ident, (), $retval:tt $(,)?) => {
            $(#[$meta])*
            fn $name<K, V>(
                &self,
                cursor: &Cursor<K, V, KeysDuplicate>,
            ) -> io::Result<cursor_return_type!($retval)>
            where
                K: ?Sized + Storable,
                V: ?Sized + Storable;
    };
    ($($meta:meta)*, $name:ident, (key), $retval:tt $(,)?) => {
            $(#[$meta])*
            fn $name<'kr, K, V, KRef>(
                &self,
                cursor: &Cursor<K, V, KeysDuplicate>,
                key: KRef,
            ) -> io::Result<cursor_return_type!($retval)>
            where
                K: ?Sized + Storable + 'kr,
                V: ?Sized + Storable,
                KRef: StorableRef<'kr, K>;
    };
    ($($meta:meta)*, $name:ident, (key, value), $retval:tt $(,)?) => {
            $(#[$meta])*
            fn $name<'kr, 'vr, K, V, KRef, VRef>(
                &self,
                cursor: &Cursor<K, V, KeysDuplicate>,
                key: KRef,
                value: VRef,
            ) -> io::Result<cursor_return_type!($retval)>
            where
                K: ?Sized + Storable + 'kr,
                V: ?Sized + Storable + 'vr,
                KRef: StorableRef<'kr, K>,
                VRef: StorableRef<'vr, V>;
    };
}

/// Create signatures for several cursor methods on databases with duplicate
/// keys
macro_rules! cursor_method_dupkey_defs {
    {
        $(
            $(#[$meta:meta])*
            fn $name:ident $args:tt -> $retval:tt;
        )*
    } => {
        $(
            cursor_method_dupkey_def!($($meta)*, $name, $args, $retval);
        )*
    };
}

/// Read-write or read-only transaction
///
/// This trait provides the methods for read-access to databases.
/// The simplest interface is provided by [`Txn::get`], which returns a
/// [reference](Storable::AlignedRef) to the stored value for a given key
/// (or [`None`] if the key does not exist).
/// If an owned value is desired, use [`Txn::get_owned`] instead, which
/// automatically converts the reference (e.g. `&str`) into an owned value
/// (e.g. `String`) using [`GenericCow::into_owned`].
///
/// [`Txn::get`] and [`Txn::get_owned`] will only retrieve the first value in
/// case of [duplicate keys](DbBuilder::keys_duplicate). For more sophisticated
/// access, use [`Txn::new_cursor`] to create a new [`Cursor`] which then can
/// be used with one of the cursor methods in the `Txn` trait (or the [`TxnRw`]
/// struct in case of write access, respectively).
pub trait Txn {
    /// Get reference to value in database
    fn get<'kr, K, V, C, KRef>(
        &self,
        db: &Db<K, V, C>,
        key: KRef,
    ) -> io::Result<Option<V::AlignedRef<'_>>>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable,
        C: Constraint,
        KRef: StorableRef<'kr, K>;
    /// Get owned value from database
    fn get_owned<'a, 'kr, K, V, C, KRef>(
        &'a self,
        db: &Db<K, V, C>,
        key: KRef,
    ) -> io::Result<Option<<V as ToOwned>::Owned>>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + ToOwned,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
    {
        Ok(self.get(db, key)?.map(|x| x.into_owned()))
    }
    /// Create a new cursor
    fn new_cursor<K, V, C>(&self, db: &Db<K, V, C>) -> io::Result<Cursor<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint;
    /// Get number of values for current cursor position
    fn cursor_get_current_value_count<K, V>(
        &self,
        cursor: &Cursor<K, V, KeysDuplicate>,
    ) -> io::Result<usize>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable;
    cursor_method_defs! {
        /// Set cursor to first entry in database
        ///
        /// Returns `false` if database is empty.
        fn cursor_set_first() -> bool;
        /// Set cursor to first entry in database and get pair
        fn cursor_set_first_get_pair() -> (key, value);
        /// Set cursor to last entry in database
        ///
        /// Returns `false` if database is empty.
        fn cursor_set_last() -> bool;
        /// Set cursor to last entry in database and get pair
        fn cursor_set_last_get_pair() -> (key, value);
        /// Get key at current cursor position
        fn cursor_get_current_key() -> key;
        /// Get value at current cursor position
        fn cursor_get_current_value() -> value;
        /// Get key-value pair at current cursor position
        fn cursor_get_current_pair() -> (key, value);
        /// Set cursor to key
        ///
        /// Returns `false` if key does not exist.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_key(key) -> bool;
        /// Set cursor to key and get value
        fn cursor_set_key_get_value(key) -> value;
        /// Set cursor to key or next greater key if not existent
        ///
        /// Returns `false` if no matching key has been found.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_search_key(key) -> bool;
        /// Set cursor to key or next greater key if not existent and get pair
        fn cursor_search_key_get_pair(key) -> (key, value);
        /// Move cursor to next entry in database
        ///
        /// Returns `false` if there is no next entry.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_next() -> bool;
        /// Move cursor to next entry in database and get pair
        fn cursor_set_next_get_pair() -> (key, value);
        /// Move cursor to previous entry in database
        ///
        /// Returns `false` if there is no previous entry.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_prev() -> bool;
        /// Move cursor to previous entry in database and get pair
        fn cursor_set_prev_get_pair() -> (key, value);
        /// Move cursor to first value of next key
        ///
        /// Returns `false` if there is no next key.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_next_key() -> bool;
        /// Move cursor to first value of next key and get pair
        fn cursor_set_next_key_get_pair() -> (key, value);
        /// Move cursor to last value of previous key
        ///
        /// Returns `false` if there is no previous key.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_prev_key() -> bool;
        /// Move cursor to last value of previous key and get pair
        fn cursor_set_prev_key_get_pair() -> (key, value);
    }
    cursor_method_dupkey_defs! {
        /// Move cursor to first value of current key
        ///
        /// Returns `false` if no value has been found.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_first_value() -> bool;
        /// Move cursor to first value of current key and get value
        fn cursor_set_first_value_get_value() -> value;
        /// Move cursor to last value of current key
        ///
        /// Returns `false` if no value has been found.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_last_value() -> bool;
        /// Move cursor to last value of current key and get value
        fn cursor_set_last_value_get_value() -> value;
        /// Set cursor to key-value pair
        ///
        /// Returns `false` if not found.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_pair(key, value) -> bool;
        /// Set cursor to key and value or next greater value if not existent
        ///
        /// Returns `false` if no matching value has been found for the given key.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_key_search_value(key, value) -> bool;
        /// Set cursor to key and value or next greater value if not existent and get value
        fn cursor_set_key_search_value_get_value(key, value) -> value;
        /// Move cursor to next value of current key
        ///
        /// Returns `false` if there is no next value for the current key.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_next_value() -> bool;
        /// Move cursor to next value of current key and get value
        fn cursor_set_next_value_get_value() -> value;
        /// Move cursor to next value of current key
        ///
        /// Returns `false` if there is no previous value for the current key.
        /// Note that in this case it's *not* guaranteed that subsequent
        /// accesses to the cursor using `cursor_get_current_` methods return
        /// `None`.
        fn cursor_set_prev_value() -> bool;
        /// Move cursor to next value of current key and get value
        fn cursor_set_prev_value_get_value() -> value;
    }
}

/// Helper macro to extract return value for cursor methods
macro_rules! cursor_return_value {
    (bool, $x:expr $(,)?) => {
        match $x {
            Some(_) => true,
            None => false,
        }
    };
    (key, $x:expr $(,)?) => {
        $x.map(|(k, _)| k())
    };
    (value, $x:expr $(,)?) => {
        $x.map(|(_, v)| v())
    };
    ((key, value), $x:expr $(,)?) => {
        $x.map(|(k, v)| (k(), v()))
    };
}

/// Create implementation for cursor method
macro_rules! cursor_method_impl {
    ($name:ident, (), $retval:tt, $op:ident $(,)?) => {
        fn $name<K, V, C>(
            &self,
            cursor: &Cursor<K, V, C>,
        ) -> io::Result<cursor_return_type!($retval)>
        where
            K: ?Sized + Storable,
            V: ?Sized + Storable,
            C: Constraint,
        {
            // SAFETY:
            //  *  Neither `TxnRo` nor `TxnRw` is `Sync` and `self.backend` is
            //     not accessed by any method calling this method.
            //  *  The macro `cursor_return_type!` gives a return type with
            //     `'_` as lifetime, which corresponds to the elided lifetime
            //     of `&self`. This lifetime cannot live longer than the end of
            //     this transaction and ends before any data is modified by the
            //     transaction. It is passed to `TxnBackend::cursor_op_unsafe`
            //     by inference.
            Ok(cursor_return_value!($retval, unsafe {
                let backend = &mut *self.backend.get();
                backend.cursor_op_unsafe::<K, V, C, &K, &V>(cursor, None, None, lmdb::$op)?
            }))
        }
    };
    ($name:ident, (key), $retval:tt, $op:ident) => {
        fn $name<'kr, K, V, C, KRef>(
            &self,
            cursor: &Cursor<K, V, C>,
            key: KRef,
        ) -> io::Result<cursor_return_type!($retval)>
        where
            K: ?Sized + Storable + 'kr,
            V: ?Sized + Storable,
            C: Constraint,
            KRef: StorableRef<'kr, K>,
        {
            // SAFETY: see above
            Ok(cursor_return_value!($retval, unsafe {
                let backend = &mut *self.backend.get();
                backend.cursor_op_unsafe::<K, V, C, KRef, &V>(cursor, Some(key), None, lmdb::$op)?
            }))
        }
    };
    ($name:ident, (key, value), $retval:tt, $op:ident) => {
        fn $name<'kr, 'vr, K, V, C, KRef, VRef>(
            &self,
            cursor: &Cursor<K, V, C>,
            key: KRef,
            value: VRef,
        ) -> io::Result<cursor_return_type!($retval)>
        where
            K: ?Sized + Storable + 'kr,
            V: ?Sized + Storable + 'vr,
            C: Constraint,
            KRef: StorableRef<'kr, K>,
            VRef: StorableRef<'vr, V>,
        {
            // SAFETY: see above
            Ok(cursor_return_value!($retval, unsafe {
                let backend = &mut *self.backend.get();
                backend.cursor_op_unsafe::<K, V, C, KRef, VRef>(
                    cursor,
                    Some(key),
                    Some(value),
                    lmdb::$op,
                )?
            }))
        }
    };
}

/// Create implementation for cursor method for databases with duplicate keys
macro_rules! cursor_method_dupkey_impl {
    ($name:ident, (), $retval:tt, $op:ident $(,)?) => {
        fn $name<K, V>(
            &self,
            cursor: &Cursor<K, V, KeysDuplicate>,
        ) -> io::Result<cursor_return_type!($retval)>
        where
            K: ?Sized + Storable,
            V: ?Sized + Storable,
        {
            // SAFETY: see `cursor_method_impl!`
            Ok(cursor_return_value!($retval, unsafe {
                let backend = &mut *self.backend.get();
                backend.cursor_op_unsafe::<K, V, KeysDuplicate, &K, &V>(
                    cursor,
                    None,
                    None,
                    lmdb::$op,
                )?
            }))
        }
    };
    ($name:ident, (key), $retval:tt, $op:ident) => {
        fn $name<'kr, K, V, KRef>(
            &self,
            cursor: &Cursor<K, V, KeysDuplicate>,
            key: KRef,
        ) -> io::Result<cursor_return_type!($retval)>
        where
            K: ?Sized + Storable + 'kr,
            V: ?Sized + Storable,
            KRef: StorableRef<'kr, K>,
        {
            // SAFETY: see `cursor_method_impl!`
            Ok(cursor_return_value!($retval, unsafe {
                let backend = &mut *self.backend.get();
                backend.cursor_op_unsafe::<K, V, KeysDuplicate, KRef, &V>(
                    cursor,
                    Some(key),
                    None,
                    lmdb::$op,
                )?
            }))
        }
    };
    ($name:ident, (key, value), $retval:tt, $op:ident) => {
        fn $name<'kr, 'vr, K, V, KRef, VRef>(
            &self,
            cursor: &Cursor<K, V, KeysDuplicate>,
            key: KRef,
            value: VRef,
        ) -> io::Result<cursor_return_type!($retval)>
        where
            K: ?Sized + Storable + 'kr,
            V: ?Sized + Storable + 'vr,
            KRef: StorableRef<'kr, K>,
            VRef: StorableRef<'vr, V>,
        {
            // SAFETY: see `cursor_method_impl!`
            Ok(cursor_return_value!($retval, unsafe {
                let backend = &mut *self.backend.get();
                backend.cursor_op_unsafe::<K, V, KeysDuplicate, KRef, VRef>(
                    cursor,
                    Some(key),
                    Some(value),
                    lmdb::$op,
                )?
            }))
        }
    };
}

/// Create implementations for all cursor methods
macro_rules! cursor_method_impls {
    ($macro:ident, $dupkey_macro:ident) => {
        $macro!(cursor_set_first, (), bool, MDB_cursor_op_MDB_FIRST);
        $macro!(
            cursor_set_first_get_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_FIRST
        );
        $macro!(cursor_set_last, (), bool, MDB_cursor_op_MDB_LAST);
        $macro!(
            cursor_set_last_get_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_LAST
        );
        $macro!(
            cursor_get_current_key,
            (),
            key,
            MDB_cursor_op_MDB_GET_CURRENT
        );
        $macro!(
            cursor_get_current_value,
            (),
            value,
            MDB_cursor_op_MDB_GET_CURRENT
        );
        $macro!(
            cursor_get_current_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_GET_CURRENT
        );
        $macro!(cursor_set_key, (key), bool, MDB_cursor_op_MDB_SET);
        $macro!(
            cursor_set_key_get_value,
            (key),
            value,
            MDB_cursor_op_MDB_SET
        );
        $macro!(cursor_search_key, (key), bool, MDB_cursor_op_MDB_SET_RANGE);
        $macro!(
            cursor_search_key_get_pair,
            (key),
            (key, value),
            MDB_cursor_op_MDB_SET_RANGE
        );
        $macro!(cursor_set_next, (), bool, MDB_cursor_op_MDB_NEXT);
        $macro!(
            cursor_set_next_get_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_NEXT
        );
        $macro!(cursor_set_prev, (), bool, MDB_cursor_op_MDB_PREV);
        $macro!(
            cursor_set_prev_get_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_PREV
        );
        $macro!(cursor_set_next_key, (), bool, MDB_cursor_op_MDB_NEXT_NODUP);
        $macro!(
            cursor_set_next_key_get_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_NEXT_NODUP
        );
        $macro!(cursor_set_prev_key, (), bool, MDB_cursor_op_MDB_PREV_NODUP);
        $macro!(
            cursor_set_prev_key_get_pair,
            (),
            (key, value),
            MDB_cursor_op_MDB_PREV_NODUP
        );
        $dupkey_macro!(
            cursor_set_first_value,
            (),
            bool,
            MDB_cursor_op_MDB_FIRST_DUP
        );
        $dupkey_macro!(
            cursor_set_first_value_get_value,
            (),
            value,
            MDB_cursor_op_MDB_FIRST_DUP
        );
        $dupkey_macro!(cursor_set_last_value, (), bool, MDB_cursor_op_MDB_LAST_DUP);
        $dupkey_macro!(
            cursor_set_last_value_get_value,
            (),
            value,
            MDB_cursor_op_MDB_LAST_DUP
        );
        $dupkey_macro!(
            cursor_set_pair,
            (key, value),
            bool,
            MDB_cursor_op_MDB_GET_BOTH
        );
        $dupkey_macro!(
            cursor_set_key_search_value,
            (key, value),
            bool,
            MDB_cursor_op_MDB_GET_BOTH_RANGE
        );
        $dupkey_macro!(
            cursor_set_key_search_value_get_value,
            (key, value),
            value,
            MDB_cursor_op_MDB_GET_BOTH_RANGE
        );
        $dupkey_macro!(cursor_set_next_value, (), bool, MDB_cursor_op_MDB_NEXT_DUP);
        $dupkey_macro!(
            cursor_set_next_value_get_value,
            (),
            value,
            MDB_cursor_op_MDB_NEXT_DUP
        );
        $dupkey_macro!(cursor_set_prev_value, (), bool, MDB_cursor_op_MDB_PREV_DUP);
        $dupkey_macro!(
            cursor_set_prev_value_get_value,
            (),
            value,
            MDB_cursor_op_MDB_PREV_DUP
        );
    };
}

impl<'a> Txn for TxnRo<'a> {
    fn get<'kr, K, V, C, KRef>(
        &self,
        db: &Db<K, V, C>,
        key: KRef,
    ) -> io::Result<Option<V::AlignedRef<'_>>>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
    {
        // SAFETY:
        //  *  `TxnRo` is `!Sync` and `self.backend` is not accessed by any
        //     method calling this method.
        //  *  The return type of this function has `'_` as lifetime, which
        //     corresponds to the elided lifetime of `&self`. This lifetime
        //     cannot live longer than the end of this transaction and ends
        //     before any data is modified by the transaction. It is passed
        //     to `TxnBackend::get_unsafe` by inference.
        unsafe {
            let backend = &mut *self.backend.get();
            backend.get_unsafe::<K, V, C, KRef>(db, key)
        }
    }
    fn new_cursor<K, V, C>(&self, db: &Db<K, V, C>) -> io::Result<Cursor<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        // SAFETY: `TxnRo` is `!Sync` and `self.backend` is not accessed by any
        // method calling this method
        let backend = unsafe { &mut *self.backend.get() };
        backend.new_cursor(db)
    }
    fn cursor_get_current_value_count<K, V>(
        &self,
        cursor: &Cursor<K, V, KeysDuplicate>,
    ) -> io::Result<usize>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
    {
        // SAFETY: `TxnRo` is `!Sync` and `self.backend` is not accessed by any
        // method calling this method
        let backend = unsafe { &mut *self.backend.get() };
        backend.cursor_get_current_value_count(cursor)
    }
    cursor_method_impls!(cursor_method_impl, cursor_method_dupkey_impl);
}

impl<'a> Txn for TxnRw<'a> {
    fn get<'kr, K, V, C, KRef>(
        &self,
        db: &Db<K, V, C>,
        key: KRef,
    ) -> io::Result<Option<V::AlignedRef<'_>>>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
    {
        // SAFETY:
        //  *  `TxnRw` is `!Sync` and `self.backend` is not accessed by any
        //     method calling this method.
        //  *  The return type of this function has `'_` as lifetime, which
        //     corresponds to the elided lifetime of `&self`. This lifetime
        //     cannot live longer than the end of this transaction and ends
        //     before any data is modified by the transaction. It is passed
        //     to `TxnBackend::get_unsafe` by inference.
        unsafe {
            let backend = &mut *self.backend.get();
            backend.get_unsafe::<K, V, C, KRef>(db, key)
        }
    }
    fn new_cursor<K, V, C>(&self, db: &Db<K, V, C>) -> io::Result<Cursor<K, V, C>>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        // SAFETY: `TxnRw` is `!Sync` and `self.backend` is not accessed by any
        // method calling this method
        let backend = unsafe { &mut *self.backend.get() };
        backend.new_cursor(db)
    }
    fn cursor_get_current_value_count<K, V>(
        &self,
        cursor: &Cursor<K, V, KeysDuplicate>,
    ) -> io::Result<usize>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
    {
        // SAFETY: `TxnRw` is `!Sync` and `self.backend` is not accessed by any
        // method calling this method
        let backend = unsafe { &mut *self.backend.get() };
        backend.cursor_get_current_value_count(cursor)
    }
    cursor_method_impls!(cursor_method_impl, cursor_method_dupkey_impl);
}

impl<'a> TxnRw<'a> {
    /// Add commit handler to be executed on [`TxnRw::commit`] before the
    /// transaction is actually committed
    pub fn on_commit<F>(&mut self, commit_handler: F)
    where
        F: FnOnce(&mut Self) -> io::Result<()> + 'a,
    {
        self.commit_handlers.push(Box::new(commit_handler));
    }
    /// Commit transaction
    ///
    /// Previously added commit handlers (see [`TxnRw::on_commit`]) are
    /// executed in the same order as they have been added. If one commit
    /// handler reports an error, the transaction is aborted.
    pub fn commit(mut self) -> io::Result<()> {
        for handler in take(&mut self.commit_handlers) {
            handler(&mut self)?;
        }
        self.backend.into_inner().commit()
    }
    /// Abort transaction (same as [`drop`])
    pub fn abort(self) {}
    /// Start nested transaction
    ///
    /// Panics if environment does not support nested transactions (which is
    /// the case when the [`EnvBuilder::writemap`] flag has been set).
    pub fn nested(&mut self) -> io::Result<TxnRw<'_>> {
        let backend = self.backend.get_mut();
        assert!(
            backend.env_ro.backend.nestable_txns,
            "environment does not support nested transactions"
        );
        let mut txn_inner = MaybeUninit::<*mut lmdb::MDB_txn>::uninit();
        // SAFETY:
        //  *  A mutable borrow of the parent transaction (`&mut self`) ensures
        //     that the parent transaction doesn't issue any operations while
        //     the child is active and that as long as the child transaction is
        //     active, the same rules apply as if the parent transaction is
        //     still open (in particular: no calls to `lmdb::mdb_txn_begin`
        //     with parent set to NULL can happen).
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_txn_begin(
                backend.env_ro.backend.inner,
                backend.inner,
                0,
                txn_inner.as_mut_ptr(),
            ))?;
        }
        // SAFETY: call to `lmdb::mdb_txn_begin` above has been successful as
        // otherwise `check_err_code` would have returned an `Err` resulting in
        // early return from this function
        let txn_inner = unsafe { txn_inner.assume_init() };
        Ok(TxnRw {
            backend: UnsafeCell::new(TxnBackend {
                env_ro: &backend.env_ro,
                inner: txn_inner,
                cursors: Default::default(),
            }),
            used_dbs: UsedDbs::Nested(&mut self.used_dbs),
            commit_handlers: Default::default(),
        })
    }
    /// Implementation for various `put` related methods
    ///
    /// # Safety
    ///
    /// Passed `flags` must be valid and correctly chosen for the given `db`.
    unsafe fn put_with_flags<'kr, 'vr, K, V, C, KRef, VRef>(
        &mut self,
        db: &Db<K, V, C>,
        key: KRef,
        value: VRef,
        flags: LmdbFlags,
    ) -> io::Result<bool>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + 'vr,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
        VRef: StorableRef<'vr, V>,
    {
        let backend = self.backend.get_mut();
        db.backend.assert_env_backend(&backend.env_ro.backend);
        let key: K::BytesRef<'_> = key.ref_to_bytes();
        let value: V::BytesRef<'_> = value.ref_to_bytes();
        backend.env_ro.assert_valid_keysize(&key)?;
        if C::DUPLICATE_KEYS {
            backend.env_ro.assert_valid_valuesize(&value)?;
        }
        // TODO: use `get_or_insert_owned` if stabilized
        //self.used_dbs.get_or_insert_owned(&db.backend);
        if !self.used_dbs.contains(&db.backend) {
            self.used_dbs.insert(db.backend.to_owned());
        }
        let lmdb_key = lmdb::MDB_val {
            mv_size: key.len(),
            mv_data: key.as_ptr() as *mut _,
        };
        let mut lmdb_data = lmdb::MDB_val {
            mv_size: value.len(),
            mv_data: value.as_ptr() as *mut _,
        };
        Ok(
            // SAFETY:
            //  *  `lmdb_key.mv_data` is a valid pointer to `lmdb_key.mv_size`
            //     bytes of memory for reading.
            //  *  `lmdb_data.mv_data` is a valid pointer to
            //     `lmdb_data.mv_size` bytes of memory for reading and writing
            //     (writing is required when some `flags` are set).
            //  *  The memory pointed to by `lmdb_data.mv_data` is not read
            //     after the transaction ends or modifies data (because it is
            //     not read at all).
            //  *  Documentation demands that `flags` are valid and correctly
            //     chosen for the database.
            match unsafe {
                lmdb::mdb_put(
                    backend.inner,
                    db.backend.inner,
                    &lmdb_key as *const _ as *mut lmdb::MDB_val,
                    &mut lmdb_data,
                    flags,
                )
            } {
                lmdb::MDB_KEYEXIST => false,
                status => {
                    // SAFETY: `status` is a valid LMDB return code, and if
                    // negative, it is a valid LMDB error code
                    unsafe { check_err_code(status) }?;
                    true
                }
            },
        )
    }
    /// Put value into database
    ///
    /// This will overwrite an existing value if the database does not use the
    /// [`DbBuilder::keys_duplicate`] option.
    pub fn put<'kr, 'vr, K, V, C, KRef, VRef>(
        &mut self,
        db: &Db<K, V, C>,
        key: KRef,
        value: VRef,
    ) -> io::Result<()>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + 'vr,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
        VRef: StorableRef<'vr, V>,
    {
        // SAFETY: passing no flags (`0`) is valid for any database
        unsafe {
            self.put_with_flags(db, key, value, 0)?;
        }
        Ok(())
    }
    /// Put value into database unless key exists
    ///
    /// Returns `Ok(false)` if key exists.
    pub fn put_unless_key_exists<'kr, 'vr, K, V, C, KRef, VRef>(
        &mut self,
        db: &Db<K, V, C>,
        key: KRef,
        value: VRef,
    ) -> io::Result<bool>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + 'vr,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
        VRef: StorableRef<'vr, V>,
    {
        // SAFETY: passing `lmdb::MDB_NOOVERWRITE` is valid for any database
        unsafe { self.put_with_flags(db, key, value, lmdb::MDB_NOOVERWRITE) }
    }
    /// Put value into database unless key-value pair exists
    ///
    /// Returns `Ok(false)` if key-value pair exists.
    pub fn put_unless_pair_exists<'kr, 'vr, K, V, KRef, VRef>(
        &mut self,
        db: &Db<K, V, KeysDuplicate>,
        key: KRef,
        value: VRef,
    ) -> io::Result<bool>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + 'vr,
        KRef: StorableRef<'kr, K>,
        VRef: StorableRef<'vr, V>,
    {
        // SAFETY: passing `lmdb::MDB_NODUPDATA` is valid for databases with
        // the `KeysDuplicate` option, which is set for the given `db`
        unsafe { self.put_with_flags(db, key, value, lmdb::MDB_NODUPDATA) }
    }
    /// Delete all values from database that match a given key
    pub fn delete_key<'kr, K, V, C, KRef>(
        &mut self,
        db: &Db<K, V, C>,
        key: KRef,
    ) -> io::Result<bool>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable,
        C: Constraint,
        KRef: StorableRef<'kr, K>,
    {
        let backend = self.backend.get_mut();
        db.backend.assert_env_backend(&backend.env_ro.backend);
        let key: K::BytesRef<'_> = key.ref_to_bytes();
        // TODO: use `get_or_insert_owned` if stabilized
        //self.used_dbs.get_or_insert_owned(&db.backend);
        if !self.used_dbs.contains(&db.backend) {
            self.used_dbs.insert(db.backend.to_owned());
        }
        let lmdb_key = lmdb::MDB_val {
            mv_size: key.len(),
            mv_data: key.as_ptr() as *mut _,
        };
        Ok(
            // SAFETY:
            //  *  `lmdb_key.mv_data` is a valid pointer to `lmdb_key.mv_size`
            //     bytes of memory for reading.
            //  *  It is allowed to pass a NULL pointer as last argument.
            match unsafe {
                lmdb::mdb_del(
                    backend.inner,
                    db.backend.inner,
                    &lmdb_key as *const _ as *mut lmdb::MDB_val,
                    null_mut(),
                )
            } {
                lmdb::MDB_NOTFOUND => false,
                status => {
                    // SAFETY: `status` is a valid LMDB return code, and if
                    // negative, it is a valid LMDB error code
                    unsafe { check_err_code(status) }?;
                    true
                }
            },
        )
    }
    /// Delete key-value pair from database
    pub fn delete_pair<'kr, 'vr, K, V, KRef, VRef>(
        &mut self,
        db: &Db<K, V, KeysDuplicate>,
        key: KRef,
        value: VRef,
    ) -> io::Result<bool>
    where
        K: ?Sized + Storable + 'kr,
        V: ?Sized + Storable + 'vr,
        KRef: StorableRef<'kr, K>,
        VRef: StorableRef<'vr, V>,
    {
        let backend = self.backend.get_mut();
        db.backend.assert_env_backend(&backend.env_ro.backend);
        let key: K::BytesRef<'_> = key.ref_to_bytes();
        let value: V::BytesRef<'_> = value.ref_to_bytes();
        // TODO: use `get_or_insert_owned` if stabilized
        //self.used_dbs.get_or_insert_owned(&db.backend);
        if !self.used_dbs.contains(&db.backend) {
            self.used_dbs.insert(db.backend.to_owned());
        }
        let lmdb_key = lmdb::MDB_val {
            mv_size: key.len(),
            mv_data: key.as_ptr() as *mut _,
        };
        let lmdb_value = lmdb::MDB_val {
            mv_size: value.len(),
            mv_data: value.as_ptr() as *mut _,
        };
        Ok(
            // SAFETY:
            //  *  `lmdb_key.mv_data` and `lmdb_data.mv_data` are valid
            //     pointers to `lmdb_key.mv_size` and `lmdb_data.mv_size`
            //     bytes of memory, respectively, for reading.
            match unsafe {
                lmdb::mdb_del(
                    backend.inner,
                    db.backend.inner,
                    &lmdb_key as *const _ as *mut lmdb::MDB_val,
                    &lmdb_value as *const _ as *mut lmdb::MDB_val,
                )
            } {
                lmdb::MDB_NOTFOUND => false,
                status => {
                    // SAFETY: `status` is a valid LMDB return code, and if
                    // negative, it is a valid LMDB error code
                    unsafe { check_err_code(status) }?;
                    true
                }
            },
        )
    }
    /// Delete all entries in a database
    ///
    /// Use [`EnvRw::drop_db`] to completely drop the database.
    pub fn delete_all<K, V, C>(&mut self, db: &Db<K, V, C>) -> io::Result<()>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        let backend = self.backend.get_mut();
        db.backend.assert_env_backend(&backend.env_ro.backend);
        // SAFETY:
        //  *  Passing zero (`0`) as a last argument to `lmdb::mdb_drop` means
        //     the database doesn't get closed but just emptied; thus no
        //     special requirements for calling the function exist.
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_drop(backend.inner, db.backend.inner, 0))?;
        }
        Ok(())
    }
    /// Delete key-value pair at current cursor position
    pub fn cursor_delete_current<K, V, C>(&mut self, cursor: &Cursor<K, V, C>) -> io::Result<()>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
        C: Constraint,
    {
        // TODO: use `get_or_insert_owned` if stabilized
        //self.used_dbs.get_or_insert_owned(&cursor.db.backend);
        if !self.used_dbs.contains(&cursor.db.backend) {
            self.used_dbs.insert(cursor.db.backend.to_owned());
        }
        let backend = self.backend.get_mut();
        cursor.backend.assert_txn_backend(backend);
        // SAFETY:
        //  *  Regarding closing the cursor or its transaction, refer to the
        //     corresponding other SAFETY comments on closing the cursor
        //     and/or the transaction, respectively.
        //  *  Passing no flags (`0`) as last argument to
        //     `lmdb::mdb_cursor_del`is valid for any database
        //  *  The status passed to `check_err_code` is valid.
        unsafe { check_err_code(lmdb::mdb_cursor_del(cursor.backend.inner, 0)) }
    }
    /// Delete all values with same key at current cursor position
    pub fn cursor_delete_current_key<K, V>(
        &mut self,
        cursor: &Cursor<K, V, KeysDuplicate>,
    ) -> io::Result<()>
    where
        K: ?Sized + Storable,
        V: ?Sized + Storable,
    {
        // TODO: use `get_or_insert_owned` if stabilized
        //self.used_dbs.get_or_insert_owned(&cursor.db.backend);
        if !self.used_dbs.contains(&cursor.db.backend) {
            self.used_dbs.insert(cursor.db.backend.to_owned());
        }
        let backend = self.backend.get_mut();
        cursor.backend.assert_txn_backend(backend);
        // SAFETY:
        //  *  Regarding closing the cursor or its transaction, refer to the
        //     corresponding other SAFETY comments on closing the cursor
        //     and/or the transaction, respectively.
        //  *  Passing `lmdb::MDB_NODUPDATA` is valid for databases with
        //     the `KeysDuplicate` option, which is set for the given `cursor`
        //     and thus also for the corresponding database.
        //  *  The status passed to `check_err_code` is valid.
        unsafe {
            check_err_code(lmdb::mdb_cursor_del(
                cursor.backend.inner,
                lmdb::MDB_NODUPDATA as LmdbFlags,
            ))
        }
    }
}

impl<K, V, C> Cursor<K, V, C> {
    /// Associated database
    pub fn db(&self) -> &Db<K, V, C> {
        &self.db
    }
}

/// Version of underlying LMDB library
#[derive(Eq, PartialEq, Clone, Debug)]
pub struct LmdbVersion {
    /// Descriptive string
    pub string: &'static str,
    /// Major version number
    pub major: i32,
    /// Minor version number
    pub minor: i32,
    /// Patch version number
    pub patch: i32,
}

/// Retrieve version of underlying LMDB library
pub fn lmdb_version() -> LmdbVersion {
    let mut major: c_int = 0;
    let mut minor: c_int = 0;
    let mut patch: c_int = 0;
    // SAFETY: `lmdb::mdb_version` is expected to return a static C string
    // as the API documentation does not put any constraints on the usage of
    // the returned string
    let string: &CStr = unsafe {
        CStr::from_ptr(lmdb::mdb_version(
            &mut major as *mut c_int,
            &mut minor as *mut c_int,
            &mut patch as *mut c_int,
        ))
    };
    LmdbVersion {
        string: string.to_str().unwrap(),
        major: major.try_into().unwrap(),
        minor: minor.try_into().unwrap(),
        patch: patch.try_into().unwrap(),
    }
}