use std::any::TypeId;
use std::cmp::Ordering;
use std::collections::hash_map::{Entry, HashMap};
use std::ffi::{c_void, CString};
#[cfg(target_os = "linux")]
use std::fs::OpenOptions;
use std::fs::{File, Metadata};
use std::io::ErrorKind::NotFound;
#[cfg(target_os = "linux")]
use std::os::unix::fs::{MetadataExt, OpenOptionsExt};
#[cfg(unix)]
use std::os::unix::{
ffi::OsStrExt,
io::{AsRawFd, BorrowedFd, RawFd},
};
use std::panic::catch_unwind;
use std::path::{Path, PathBuf};
use std::process::abort;
use std::sync::{Arc, RwLock};
use std::time::Duration;
#[cfg(windows)]
use std::{
ffi::OsStr,
os::windows::io::{AsRawHandle, BorrowedHandle, RawHandle},
};
use std::{fmt, io, mem, ptr};
use heed_traits::{Comparator, LexicographicComparator};
use once_cell::sync::Lazy;
use synchronoise::event::SignalEvent;
use crate::cursor::MoveOperation;
use crate::database::DatabaseOpenOptions;
use crate::mdb::error::mdb_result;
use crate::mdb::ffi;
use crate::mdb::lmdb_flags::AllDatabaseFlags;
use crate::{
Database, EnvFlags, Error, Result, RoCursor, RoTxn, RwTxn, Unspecified, WriteAdmission,
};
static OPENED_ENV: Lazy<RwLock<HashMap<PathBuf, EnvEntry>>> = Lazy::new(RwLock::default);
struct EnvEntry {
env: Option<Env>,
signal_event: Arc<SignalEvent>,
options: EnvOpenOptions,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PinnedLmdbFileIdentity {
pub device: u64,
pub inode: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub struct PinnedLmdbIdentity {
pub data: PinnedLmdbFileIdentity,
pub lock: PinnedLmdbFileIdentity,
}
impl PinnedLmdbIdentity {
#[cfg(target_os = "linux")]
fn validate_before_open(&self, path: &Path) -> io::Result<()> {
if pinned_file_identity(&path.join("data.mdb"))? != self.data
|| pinned_file_identity(&path.join("lock.mdb"))? != self.lock
{
return Err(io::Error::other(
"LMDB data/lock entry differs from the caller-pinned identity",
));
}
Ok(())
}
#[cfg(not(target_os = "linux"))]
fn validate_before_open(&self, _path: &Path) -> io::Result<()> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"pinned LMDB paths are supported only on Linux",
))
}
#[cfg(target_os = "linux")]
unsafe fn validate_after_open(&self, env: *mut ffi::MDB_env, path: &Path) -> io::Result<()> {
if pinned_file_identity(&path.join("data.mdb"))? != self.data
|| pinned_file_identity(&path.join("lock.mdb"))? != self.lock
{
return Err(io::Error::other(
"pinned LMDB data/lock entry changed during open",
));
}
let mut data_fd = -1;
let status = unsafe { ffi::mdb_env_get_fd(env, &mut data_fd) };
if status != 0 {
return Err(io::Error::from_raw_os_error(status));
}
if pinned_fd_identity(data_fd)? != self.data {
return Err(io::Error::other(
"LMDB opened a different data.mdb than the pinned entry",
));
}
let mut lock_fd = -1;
let status = unsafe { ffi::mdb_env_get_lockfd(env, &mut lock_fd) };
if status != 0 {
return Err(io::Error::from_raw_os_error(status));
}
if pinned_fd_identity(lock_fd)? != self.lock {
return Err(io::Error::other(
"LMDB opened a different lock.mdb than the pinned entry",
));
}
Ok(())
}
#[cfg(not(target_os = "linux"))]
unsafe fn validate_after_open(&self, _env: *mut ffi::MDB_env, _path: &Path) -> io::Result<()> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"pinned LMDB paths are supported only on Linux",
))
}
}
#[cfg(target_os = "linux")]
fn pinned_file_identity(path: &Path) -> io::Result<PinnedLmdbFileIdentity> {
let mut options = OpenOptions::new();
options
.read(true)
.custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK);
let file = options.open(path)?;
let metadata = file.metadata()?;
if !metadata.is_file() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"pinned LMDB entry is not a regular file",
));
}
Ok(PinnedLmdbFileIdentity {
device: metadata.dev(),
inode: metadata.ino(),
})
}
#[cfg(target_os = "linux")]
fn pinned_fd_identity(fd: libc::c_int) -> io::Result<PinnedLmdbFileIdentity> {
let mut stat = mem::MaybeUninit::<libc::stat>::zeroed();
if unsafe { libc::fstat(fd, stat.as_mut_ptr()) } != 0 {
return Err(io::Error::last_os_error());
}
let stat = unsafe { stat.assume_init() };
Ok(PinnedLmdbFileIdentity {
device: stat.st_dev as u64,
inode: stat.st_ino as u64,
})
}
#[cfg(not(windows))]
fn canonicalize_path(path: &Path) -> io::Result<PathBuf> {
path.canonicalize()
}
#[cfg(windows)]
fn canonicalize_path(path: &Path) -> io::Result<PathBuf> {
let canonical = path.canonicalize()?;
let url = url::Url::from_file_path(&canonical)
.map_err(|_e| io::Error::new(io::ErrorKind::Other, "URL passing error"))?;
url.to_file_path()
.map_err(|_e| io::Error::new(io::ErrorKind::Other, "path canonicalization error"))
}
#[cfg(windows)]
trait OsStrExtLmdb {
fn as_bytes(&self) -> &[u8];
}
#[cfg(windows)]
impl OsStrExtLmdb for OsStr {
fn as_bytes(&self) -> &[u8] {
&self.to_str().unwrap().as_bytes()
}
}
#[cfg(unix)]
fn get_file_fd(file: &File) -> RawFd {
file.as_raw_fd()
}
#[cfg(windows)]
fn get_file_fd(file: &File) -> RawHandle {
file.as_raw_handle()
}
#[cfg(unix)]
unsafe fn metadata_from_fd(raw_fd: RawFd) -> io::Result<Metadata> {
let fd = BorrowedFd::borrow_raw(raw_fd);
let owned = fd.try_clone_to_owned()?;
File::from(owned).metadata()
}
#[cfg(windows)]
unsafe fn metadata_from_fd(raw_fd: RawHandle) -> io::Result<Metadata> {
let fd = BorrowedHandle::borrow_raw(raw_fd);
let owned = fd.try_clone_to_owned()?;
File::from(owned).metadata()
}
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct EnvOpenOptions {
map_size: Option<usize>,
max_readers: Option<u32>,
max_dbs: Option<u32>,
flags: EnvFlags,
#[cfg_attr(
feature = "serde",
serde(default, skip_serializing_if = "Option::is_none")
)]
write_admission: Option<WriteAdmission>,
}
impl Default for EnvOpenOptions {
fn default() -> Self {
Self::new()
}
}
impl EnvOpenOptions {
pub fn new() -> EnvOpenOptions {
EnvOpenOptions {
map_size: None,
max_readers: None,
max_dbs: None,
flags: EnvFlags::empty(),
write_admission: None,
}
}
pub fn write_admission(&mut self, admission: WriteAdmission) -> &mut Self {
self.write_admission = Some(admission);
self
}
pub fn map_size(&mut self, size: usize) -> &mut Self {
self.map_size = Some(size);
self
}
pub fn max_readers(&mut self, readers: u32) -> &mut Self {
self.max_readers = Some(readers);
self
}
pub fn max_dbs(&mut self, dbs: u32) -> &mut Self {
self.max_dbs = Some(dbs);
self
}
pub unsafe fn flags(&mut self, flags: EnvFlags) -> &mut Self {
self.flags |= flags;
self
}
pub unsafe fn open<P: AsRef<Path>>(&self, path: P) -> Result<Env> {
let path = match canonicalize_path(path.as_ref()) {
Err(err) => {
if err.kind() == NotFound && self.flags.contains(EnvFlags::NO_SUB_DIR) {
let path = path.as_ref();
match path.parent().zip(path.file_name()) {
Some((dir, file_name)) => canonicalize_path(dir)?.join(file_name),
None => return Err(err.into()),
}
} else {
return Err(err.into());
}
}
Ok(path) => path,
};
unsafe { self.open_resolved_path(path, None) }
}
pub unsafe fn open_from_pinned_path<P: AsRef<Path>>(
&self,
path: P,
pinned: PinnedLmdbIdentity,
) -> Result<Env> {
let path = path.as_ref();
if !path.is_absolute() {
return Err(Error::Io(io::Error::new(
io::ErrorKind::InvalidInput,
"pinned LMDB path must be absolute",
)));
}
pinned.validate_before_open(path)?;
unsafe { self.open_resolved_path(path.to_path_buf(), Some(pinned)) }
}
unsafe fn open_resolved_path(
&self,
path: PathBuf,
pinned: Option<PinnedLmdbIdentity>,
) -> Result<Env> {
#[cfg(not(unix))]
if self.write_admission.is_some() {
return Err(Error::Io(io::Error::new(
io::ErrorKind::Unsupported,
"LMDB write admission requires Unix file descriptors",
)));
}
let mut lock = OPENED_ENV.write().unwrap();
match lock.entry(path) {
Entry::Occupied(entry) => {
let env = entry.get().env.clone().ok_or(Error::DatabaseClosing)?;
let options = entry.get().options.clone();
if &options == self {
Ok(env)
} else {
Err(Error::BadOpenOptions { env, options })
}
}
Entry::Vacant(entry) => {
let path = entry.key();
let path_str = CString::new(path.as_os_str().as_bytes()).unwrap();
unsafe {
let mut env: *mut ffi::MDB_env = ptr::null_mut();
mdb_result(ffi::mdb_env_create(&mut env))?;
#[cfg(unix)]
if let Some(admission) = &self.write_admission {
if let Err(error) = mdb_result(ffi::mdb_env_set_write_admission(
env,
Some(crate::write_admission::admit_write),
admission.context(),
)) {
ffi::mdb_env_close(env);
return Err(error.into());
}
}
if let Some(pinned) = pinned {
mdb_result(ffi::mdb_env_set_expected_file_identity(
env,
pinned.data.device,
pinned.data.inode,
pinned.lock.device,
pinned.lock.inode,
))?;
}
if let Some(size) = self.map_size {
if size % page_size::get() != 0 {
let msg = format!(
"map size ({}) must be a multiple of the system page size ({})",
size,
page_size::get()
);
return Err(Error::Io(io::Error::new(
io::ErrorKind::InvalidInput,
msg,
)));
}
mdb_result(ffi::mdb_env_set_mapsize(env, size))?;
}
if let Some(readers) = self.max_readers {
mdb_result(ffi::mdb_env_set_maxreaders(env, readers))?;
}
if let Some(dbs) = self.max_dbs {
mdb_result(ffi::mdb_env_set_maxdbs(env, dbs))?;
}
let flags = if cfg!(feature = "read-txn-no-tls") {
self.flags | EnvFlags::NO_TLS
} else {
self.flags
};
let result = mdb_result(ffi::mdb_env_open(
env,
path_str.as_ptr(),
flags.bits(),
0o600,
));
match result {
Ok(()) => {
if let Some(pinned) = pinned {
if let Err(error) = pinned.validate_after_open(env, path) {
ffi::mdb_env_close(env);
return Err(Error::Io(error));
}
}
let signal_event = Arc::new(SignalEvent::manual(false));
let inner = EnvInner {
env,
path: path.clone(),
_write_admission: self.write_admission.clone(),
};
let env = Env(Arc::new(inner));
let cache_entry = EnvEntry {
env: Some(env.clone()),
options: self.clone(),
signal_event,
};
entry.insert(cache_entry);
Ok(env)
}
Err(e) => {
ffi::mdb_env_close(env);
Err(e.into())
}
}
}
}
}
}
}
pub fn env_closing_event<P: AsRef<Path>>(path: P) -> Option<EnvClosingEvent> {
let lock = OPENED_ENV.read().unwrap();
lock.get(path.as_ref())
.map(|e| EnvClosingEvent(e.signal_event.clone()))
}
#[derive(Clone)]
pub struct Env(Arc<EnvInner>);
impl fmt::Debug for Env {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
let EnvInner { path, .. } = self.0.as_ref();
f.debug_struct("Env")
.field("path", &path.display())
.finish_non_exhaustive()
}
}
struct EnvInner {
env: *mut ffi::MDB_env,
path: PathBuf,
_write_admission: Option<WriteAdmission>,
}
unsafe impl Send for EnvInner {}
unsafe impl Sync for EnvInner {}
impl Drop for EnvInner {
fn drop(&mut self) {
let mut lock = OPENED_ENV.write().unwrap();
match lock.remove(&self.path) {
None => panic!("It seems another env closed this env before"),
Some(EnvEntry { signal_event, .. }) => {
unsafe {
ffi::mdb_env_close(self.env);
}
signal_event.signal();
}
}
}
}
unsafe extern "C" fn custom_key_cmp_wrapper<C: Comparator>(
a: *const ffi::MDB_val,
b: *const ffi::MDB_val,
) -> i32 {
let a = unsafe { ffi::from_val(*a) };
let b = unsafe { ffi::from_val(*b) };
match catch_unwind(|| C::compare(a, b)) {
Ok(Ordering::Less) => -1,
Ok(Ordering::Equal) => 0,
Ok(Ordering::Greater) => 1,
Err(_) => abort(),
}
}
pub enum DefaultComparator {}
impl LexicographicComparator for DefaultComparator {
#[inline]
fn compare_elem(a: u8, b: u8) -> Ordering {
a.cmp(&b)
}
#[inline]
fn successor(elem: u8) -> Option<u8> {
match elem {
u8::MAX => None,
elem => Some(elem + 1),
}
}
#[inline]
fn predecessor(elem: u8) -> Option<u8> {
match elem {
u8::MIN => None,
elem => Some(elem - 1),
}
}
#[inline]
fn max_elem() -> u8 {
u8::MAX
}
#[inline]
fn min_elem() -> u8 {
u8::MIN
}
}
#[derive(Debug, Copy, Clone)]
pub enum CompactionOption {
Enabled,
Disabled,
}
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum FlagSetMode {
Enable,
Disable,
}
impl FlagSetMode {
fn as_mdb_env_set_flags_input(self) -> i32 {
match self {
Self::Enable => 1,
Self::Disable => 0,
}
}
}
impl Env {
pub(crate) fn env_mut_ptr(&self) -> *mut ffi::MDB_env {
self.0.env
}
pub fn real_disk_size(&self) -> Result<u64> {
let mut fd = mem::MaybeUninit::uninit();
unsafe { mdb_result(ffi::mdb_env_get_fd(self.env_mut_ptr(), fd.as_mut_ptr()))? };
let fd = unsafe { fd.assume_init() };
let metadata = unsafe { metadata_from_fd(fd)? };
Ok(metadata.len())
}
pub fn flags(&self) -> Result<Option<EnvFlags>> {
self.get_flags().map(EnvFlags::from_bits)
}
pub unsafe fn set_flags(&self, flags: EnvFlags, mode: FlagSetMode) -> Result<()> {
mdb_result(unsafe {
ffi::mdb_env_set_flags(
self.env_mut_ptr(),
flags.bits(),
mode.as_mdb_env_set_flags_input(),
)
})
.map_err(Into::into)
}
pub fn get_flags(&self) -> Result<u32> {
let mut flags = mem::MaybeUninit::uninit();
unsafe {
mdb_result(ffi::mdb_env_get_flags(
self.env_mut_ptr(),
flags.as_mut_ptr(),
))?
};
let flags = unsafe { flags.assume_init() };
Ok(flags)
}
pub fn info(&self) -> EnvInfo {
let mut raw_info = mem::MaybeUninit::uninit();
unsafe { ffi::mdb_env_info(self.0.env, raw_info.as_mut_ptr()) };
let raw_info = unsafe { raw_info.assume_init() };
EnvInfo {
map_addr: raw_info.me_mapaddr,
map_size: raw_info.me_mapsize,
last_page_number: raw_info.me_last_pgno,
last_txn_id: raw_info.me_last_txnid,
maximum_number_of_readers: raw_info.me_maxreaders,
number_of_readers: raw_info.me_numreaders,
}
}
pub fn non_free_pages_size(&self) -> Result<u64> {
let compute_size = |stat: ffi::MDB_stat| {
(stat.ms_leaf_pages + stat.ms_branch_pages + stat.ms_overflow_pages) as u64
* stat.ms_psize as u64
};
let mut size = 0;
let mut stat = mem::MaybeUninit::uninit();
unsafe { mdb_result(ffi::mdb_env_stat(self.env_mut_ptr(), stat.as_mut_ptr()))? };
let stat = unsafe { stat.assume_init() };
size += compute_size(stat);
let rtxn = self.read_txn()?;
let dbi = self.raw_open_dbi::<DefaultComparator>(rtxn.txn, None, 0)?;
let mut cursor = RoCursor::new(&rtxn, dbi)?;
while let Some((key, _value)) = cursor.move_on_next(MoveOperation::NoDup)? {
if key.contains(&0) {
continue;
}
let key = String::from_utf8(key.to_vec()).unwrap();
if let Ok(dbi) = self.raw_open_dbi::<DefaultComparator>(rtxn.txn, Some(&key), 0) {
let mut stat = mem::MaybeUninit::uninit();
unsafe { mdb_result(ffi::mdb_stat(rtxn.txn, dbi, stat.as_mut_ptr()))? };
let stat = unsafe { stat.assume_init() };
size += compute_size(stat);
}
}
Ok(size)
}
pub fn database_options(&self) -> DatabaseOpenOptions<'_, '_, Unspecified, Unspecified> {
DatabaseOpenOptions::new(self)
}
pub fn open_database<KC, DC>(
&self,
rtxn: &RoTxn,
name: Option<&str>,
) -> Result<Option<Database<KC, DC>>>
where
KC: 'static,
DC: 'static,
{
let mut options = self.database_options().types::<KC, DC>();
if let Some(name) = name {
options.name(name);
}
options.open(rtxn)
}
pub fn create_database<KC, DC>(
&self,
wtxn: &mut RwTxn,
name: Option<&str>,
) -> Result<Database<KC, DC>>
where
KC: 'static,
DC: 'static,
{
let mut options = self.database_options().types::<KC, DC>();
if let Some(name) = name {
options.name(name);
}
options.create(wtxn)
}
pub(crate) fn raw_init_database<C: Comparator + 'static>(
&self,
raw_txn: *mut ffi::MDB_txn,
name: Option<&str>,
flags: AllDatabaseFlags,
) -> Result<u32> {
match self.raw_open_dbi::<C>(raw_txn, name, flags.bits()) {
Ok(dbi) => Ok(dbi),
Err(e) => Err(e.into()),
}
}
fn raw_open_dbi<C: Comparator + 'static>(
&self,
raw_txn: *mut ffi::MDB_txn,
name: Option<&str>,
flags: u32,
) -> std::result::Result<u32, crate::mdb::lmdb_error::Error> {
let mut dbi = 0;
let name = name.map(|n| CString::new(n).unwrap());
let name_ptr = match name {
Some(ref name) => name.as_bytes_with_nul().as_ptr() as *const _,
None => ptr::null(),
};
unsafe {
mdb_result(ffi::mdb_dbi_open(raw_txn, name_ptr, flags, &mut dbi))?;
if TypeId::of::<C>() != TypeId::of::<DefaultComparator>() {
mdb_result(ffi::mdb_set_compare(
raw_txn,
dbi,
Some(custom_key_cmp_wrapper::<C>),
))?;
}
};
Ok(dbi)
}
pub fn write_txn(&self) -> Result<RwTxn<'_>> {
RwTxn::new(self)
}
pub fn nested_write_txn<'p>(&'p self, parent: &'p mut RwTxn) -> Result<RwTxn<'p>> {
RwTxn::nested(self, parent)
}
pub fn read_txn(&self) -> Result<RoTxn<'_>> {
RoTxn::new(self)
}
pub fn static_read_txn(self) -> Result<RoTxn<'static>> {
RoTxn::static_read_txn(self)
}
pub fn copy_to_file<P: AsRef<Path>>(&self, path: P, option: CompactionOption) -> Result<File> {
let file = File::options().create_new(true).write(true).open(&path)?;
let fd = get_file_fd(&file);
unsafe { self.copy_to_fd(fd, option)? };
let file = File::open(path)?;
Ok(file)
}
pub unsafe fn copy_to_fd(
&self,
fd: ffi::mdb_filehandle_t,
option: CompactionOption,
) -> Result<()> {
let flags = if let CompactionOption::Enabled = option {
ffi::MDB_CP_COMPACT
} else {
0
};
mdb_result(ffi::mdb_env_copyfd2(self.0.env, fd, flags))?;
Ok(())
}
pub fn force_sync(&self) -> Result<()> {
unsafe { mdb_result(ffi::mdb_env_sync(self.0.env, 1))? }
Ok(())
}
pub fn path(&self) -> &Path {
&self.0.path
}
pub fn prepare_for_closing(self) -> EnvClosingEvent {
let mut lock = OPENED_ENV.write().unwrap();
match lock.get_mut(self.path()) {
None => panic!("cannot find the env that we are trying to close"),
Some(EnvEntry {
env, signal_event, ..
}) => {
let _env = env.take();
let signal_event = signal_event.clone();
drop(lock);
EnvClosingEvent(signal_event)
}
}
}
pub fn clear_stale_readers(&self) -> Result<usize> {
let mut dead: i32 = 0;
unsafe { mdb_result(ffi::mdb_reader_check(self.0.env, &mut dead))? }
Ok(dead as usize)
}
pub unsafe fn resize(&self, new_size: usize) -> Result<()> {
if new_size % page_size::get() != 0 {
let msg = format!(
"map size ({}) must be a multiple of the system page size ({})",
new_size,
page_size::get()
);
return Err(Error::Io(io::Error::new(io::ErrorKind::InvalidInput, msg)));
}
mdb_result(unsafe { ffi::mdb_env_set_mapsize(self.env_mut_ptr(), new_size) })
.map_err(Into::into)
}
pub fn max_key_size(&self) -> usize {
let maxsize: i32 = unsafe { ffi::mdb_env_get_maxkeysize(self.env_mut_ptr()) };
maxsize as usize
}
}
#[derive(Debug, Clone, Copy)]
pub struct EnvInfo {
pub map_addr: *mut c_void,
pub map_size: usize,
pub last_page_number: usize,
pub last_txn_id: usize,
pub maximum_number_of_readers: u32,
pub number_of_readers: u32,
}
#[derive(Clone)]
pub struct EnvClosingEvent(Arc<SignalEvent>);
impl EnvClosingEvent {
pub fn wait(&self) {
self.0.wait()
}
pub fn wait_timeout(&self, timeout: Duration) -> bool {
self.0.wait_timeout(timeout)
}
}
impl fmt::Debug for EnvClosingEvent {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("EnvClosingEvent").finish()
}
}
#[cfg(test)]
mod tests {
use std::io::ErrorKind;
use std::time::Duration;
use std::{fs, thread};
use crate::types::*;
use crate::{env_closing_event, EnvOpenOptions, Error};
#[test]
fn close_env() {
let dir = tempfile::tempdir().unwrap();
let env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .max_dbs(30)
.open(dir.path())
.unwrap()
};
let env_cloned = env.clone();
thread::spawn(move || {
let _env = env_cloned;
thread::sleep(Duration::from_secs(1));
});
let mut wtxn = env.write_txn().unwrap();
let db = env.create_database::<Str, Str>(&mut wtxn, None).unwrap();
wtxn.commit().unwrap();
let mut wtxn = env.write_txn().unwrap();
db.put(&mut wtxn, "hello", "hello").unwrap();
db.put(&mut wtxn, "world", "world").unwrap();
let mut iter = db.iter(&wtxn).unwrap();
assert_eq!(iter.next().transpose().unwrap(), Some(("hello", "hello")));
assert_eq!(iter.next().transpose().unwrap(), Some(("world", "world")));
assert_eq!(iter.next().transpose().unwrap(), None);
drop(iter);
wtxn.commit().unwrap();
let signal_event = env.prepare_for_closing();
eprintln!("waiting for the env to be closed");
signal_event.wait();
eprintln!("env closed successfully");
assert!(env_closing_event(dir.path()).is_none());
}
#[test]
fn reopen_env_with_different_options_is_err() {
let dir = tempfile::tempdir().unwrap();
let _env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .open(dir.path())
.unwrap()
};
let result = unsafe {
EnvOpenOptions::new()
.map_size(12 * 1024 * 1024) .open(dir.path())
};
assert!(matches!(result, Err(Error::BadOpenOptions { .. })));
}
#[test]
fn open_env_with_named_path() {
let dir = tempfile::tempdir().unwrap();
fs::create_dir_all(dir.path().join("babar.mdb")).unwrap();
let _env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .open(dir.path().join("babar.mdb"))
.unwrap()
};
let _env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .open(dir.path().join("babar.mdb"))
.unwrap()
};
}
#[test]
#[cfg(not(windows))]
fn open_database_with_writemap_flag() {
let dir = tempfile::tempdir().unwrap();
let mut envbuilder = EnvOpenOptions::new();
envbuilder.map_size(10 * 1024 * 1024); envbuilder.max_dbs(10);
unsafe { envbuilder.flags(crate::EnvFlags::WRITE_MAP) };
let env = unsafe { envbuilder.open(dir.path()).unwrap() };
let mut wtxn = env.write_txn().unwrap();
let _db = env
.create_database::<Str, Str>(&mut wtxn, Some("my-super-db"))
.unwrap();
wtxn.commit().unwrap();
}
#[test]
fn open_database_with_nosubdir() {
let dir = tempfile::tempdir().unwrap();
let mut envbuilder = EnvOpenOptions::new();
unsafe { envbuilder.flags(crate::EnvFlags::NO_SUB_DIR) };
let _env = unsafe { envbuilder.open(dir.path().join("data.mdb")).unwrap() };
}
#[test]
fn create_database_without_commit() {
let dir = tempfile::tempdir().unwrap();
let env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .max_dbs(10)
.open(dir.path())
.unwrap()
};
let mut wtxn = env.write_txn().unwrap();
let _db = env
.create_database::<Str, Str>(&mut wtxn, Some("my-super-db"))
.unwrap();
wtxn.abort();
let rtxn = env.read_txn().unwrap();
let option = env
.open_database::<Str, Str>(&rtxn, Some("my-super-db"))
.unwrap();
assert!(option.is_none());
}
#[test]
fn open_already_existing_database() {
let dir = tempfile::tempdir().unwrap();
let env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .max_dbs(10)
.open(dir.path())
.unwrap()
};
let mut wtxn = env.write_txn().unwrap();
let _db = env
.create_database::<Str, Str>(&mut wtxn, Some("my-super-db"))
.unwrap();
wtxn.commit().unwrap();
env.prepare_for_closing().wait();
let env = unsafe {
EnvOpenOptions::new()
.map_size(10 * 1024 * 1024) .max_dbs(10)
.open(dir.path())
.unwrap()
};
let rtxn = env.read_txn().unwrap();
let option = env
.open_database::<Str, Str>(&rtxn, Some("my-super-db"))
.unwrap();
assert!(option.is_some());
}
#[test]
fn resize_database() {
let dir = tempfile::tempdir().unwrap();
let page_size = page_size::get();
let env = unsafe {
EnvOpenOptions::new()
.map_size(9 * page_size)
.max_dbs(1)
.open(dir.path())
.unwrap()
};
let mut wtxn = env.write_txn().unwrap();
let db = env
.create_database::<Str, Str>(&mut wtxn, Some("my-super-db"))
.unwrap();
wtxn.commit().unwrap();
let mut wtxn = env.write_txn().unwrap();
for i in 0..64 {
db.put(&mut wtxn, &i.to_string(), "world").unwrap();
}
wtxn.commit().unwrap();
let mut wtxn = env.write_txn().unwrap();
for i in 64..128 {
db.put(&mut wtxn, &i.to_string(), "world").unwrap();
}
wtxn.commit()
.expect_err("cannot commit a transaction that would reach the map size limit");
unsafe {
env.resize(10 * page_size).unwrap();
}
let mut wtxn = env.write_txn().unwrap();
for i in 64..128 {
db.put(&mut wtxn, &i.to_string(), "world").unwrap();
}
wtxn.commit()
.expect("transaction should commit after resizing the map size");
assert_eq!(10 * page_size, env.info().map_size);
}
#[test]
fn open_read_only_without_no_env_opened_before() {
let expected_data0 = "Data Expected db0";
let dir = tempfile::tempdir().unwrap();
{
let env = unsafe {
EnvOpenOptions::new()
.map_size(16 * 1024 * 1024 * 1024) .max_dbs(32)
.open(dir.path())
.unwrap()
};
let mut wtxn = env.write_txn().unwrap();
let database0 = env
.create_database::<Str, Str>(&mut wtxn, Some("shared0"))
.unwrap();
wtxn.commit().unwrap();
let mut wtxn = env.write_txn().unwrap();
database0.put(&mut wtxn, "shared0", expected_data0).unwrap();
wtxn.commit().unwrap();
env.prepare_for_closing().wait();
}
{
let env = unsafe {
EnvOpenOptions::new()
.map_size(16 * 1024 * 1024 * 1024) .max_dbs(32)
.open(dir.path())
.unwrap()
};
let database0 = {
let rtxn = env.read_txn().unwrap();
let database0 = env
.open_database::<Str, Str>(&rtxn, Some("shared0"))
.unwrap()
.unwrap();
rtxn.commit().unwrap();
database0
};
{
let rtxn = env.read_txn().unwrap();
let value = database0.get(&rtxn, "shared0").unwrap().unwrap();
assert_eq!(value, expected_data0);
}
env.prepare_for_closing().wait();
}
{
let env = unsafe {
EnvOpenOptions::new()
.map_size(16 * 1024 * 1024 * 1024) .max_dbs(32)
.open(dir.path())
.unwrap()
};
let database0 = {
let rtxn = env.read_txn().unwrap();
let database0 = env
.open_database::<Str, Str>(&rtxn, Some("shared0"))
.unwrap()
.unwrap();
drop(rtxn);
database0
};
{
let rtxn = env.read_txn().unwrap();
let err = dbg!(database0.get(&rtxn, "shared0"));
if cfg!(windows) {
assert!(err.is_err());
} else {
assert!(
matches!(err, Err(Error::Io(ref e)) if e.kind() == ErrorKind::InvalidInput)
);
}
}
env.prepare_for_closing().wait();
}
}
#[test]
fn max_key_size() {
let dir = tempfile::tempdir().unwrap();
let env = unsafe {
EnvOpenOptions::new()
.open(dir.path().join(dir.path()))
.unwrap()
};
let maxkeysize = env.max_key_size();
eprintln!("maxkeysize: {}", maxkeysize);
if cfg!(feature = "longer-keys") {
assert!(maxkeysize > 511);
} else {
assert_eq!(maxkeysize, 511);
}
}
}