use crate::{
ack::{AckTicket, Completion, SyncTrigger, TEpoch},
error::{BindModule, ErrCode, FrozenError, FrozenResult},
file::{File, FileCfg},
hints,
};
use core::marker::PhantomData;
use std::sync::{Arc, Condvar, Mutex, RwLock, atomic};
mod interface;
mod page_size;
pub use page_size::PageSize;
#[cfg(any(target_os = "linux", target_os = "macos"))]
mod posix;
#[cfg(target_os = "windows")]
mod windows;
pub(in crate::memmap) use interface::MemMapInterface;
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub(in crate::memmap) type PlatformMemMap = posix::POSIXMemMap;
#[cfg(target_os = "windows")]
pub(in crate::memmap) type PlatformMemMap = windows::WINMemMap;
pub(in crate::memmap) mod err {
use super::{ErrCode, FrozenError, FrozenResult};
const ERRDOMAIN: u8 = 0x10;
pub const HCF: ErrCode = ErrCode::new(0x02, "hault and catch fire");
pub const UNK: ErrCode = ErrCode::new(0x04, "unknown error");
pub const NMM: ErrCode = ErrCode::new(0x06, "not enough memory available on the device");
pub const SYN: ErrCode = ErrCode::new(0x08, "failed to sync/flush data to storage device");
pub const PRM: ErrCode = ErrCode::new(0x0A, "missing permissions for IO");
pub const FXE: ErrCode = ErrCode::new(0x10, "unable to spawn flush_tx");
pub const ZRO: ErrCode = ErrCode::new(0x0C, "type T must not be zero sized");
pub const DRP: ErrCode = ErrCode::new(0x12, "type T must not implement `Drop`");
pub const ALN: ErrCode = ErrCode::new(0x14, "type T must be 8-bytes aligned");
pub const SZE: ErrCode = ErrCode::new(0x16, "`size_of::<T>()` must be multiple of 8 bytes");
pub const INV: ErrCode =
ErrCode::new(0x18, "`size_of::<T>()` must not exceed system page size");
pub const DIR: ErrCode = ErrCode::new(0x1A, "directory does not exist");
pub const PTH: ErrCode = ErrCode::new(0x1C, "invalid path to file");
pub const EXS: ErrCode = ErrCode::new(0x1E, "file already exists");
pub const NEX: ErrCode = ErrCode::new(0x20, "file does not exist");
pub const GRW: ErrCode = ErrCode::new(0x22, "unable to grow memory-mapped file");
pub const CAS: ErrCode = ErrCode::new(0x24, "compare and swap check failed");
#[inline]
pub(in crate::memmap) fn raw_error<R, E: std::fmt::Display>(
code: ErrCode,
error: E,
) -> FrozenResult<R> {
let err = FrozenError::unbound_raw(ERRDOMAIN, code, error);
Err(err)
}
#[inline]
pub(in crate::memmap) fn default_error<R>(code: ErrCode) -> FrozenResult<R> {
let err = FrozenError::unbound(ERRDOMAIN, code, "");
Err(err)
}
#[inline]
pub(in crate::memmap) fn make_raw_error<E: std::fmt::Display>(
code: ErrCode,
error: E,
) -> FrozenError {
FrozenError::unbound_raw(ERRDOMAIN, code, error)
}
}
#[derive(Debug, Clone)]
pub struct MemMapCfg {
pub module_id: u8,
pub path: std::path::PathBuf,
pub version: u32,
pub initial_t: usize,
pub sync_interval: Option<std::time::Duration>,
pub page_size: Option<PageSize>,
}
impl MemMapCfg {
#[inline]
fn versioned_path(&self) -> std::path::PathBuf {
let parent = self.path.parent();
let stem = self.path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
let extension = self.path.extension().and_then(|e| e.to_str());
let versioned_filename = match extension {
Some(ext) if !ext.is_empty() => format!("{}_{}.{}", stem, self.version, ext),
_ => format!("{}_{}", stem, self.version),
};
match parent {
Some(p) if !p.as_os_str().is_empty() => p.join(versioned_filename),
_ => std::path::PathBuf::from(versioned_filename),
}
}
}
const WRITER_BIT: u32 = 0x8000_0000;
const L1_CONTENTION: usize = 0x10;
const L2_CONTENTION: usize = 0x20;
#[derive(Debug)]
struct SlotLocks {
locks: Box<[atomic::AtomicU32]>,
}
impl SlotLocks {
fn new(total_slots: usize) -> Self {
let mut slots = Vec::with_capacity(total_slots);
for _ in 0..total_slots {
slots.push(atomic::AtomicU32::new(0));
}
Self { locks: slots.into_boxed_slice() }
}
fn grow(&mut self, additional_slots: usize) {
if additional_slots == 0 {
return;
}
let total = self.locks.len() + additional_slots;
let mut slots = Vec::with_capacity(total);
for val in self.locks.iter() {
slots.push(atomic::AtomicU32::new(val.load(atomic::Ordering::Relaxed)));
}
for _ in 0..additional_slots {
slots.push(atomic::AtomicU32::new(0));
}
self.locks = slots.into_boxed_slice();
}
#[inline(always)]
fn read_lock(&self, index: usize) -> SlotReadGuard<'_> {
let val = &self.locks[index];
let mut spins = 0;
loop {
let state = val.load(atomic::Ordering::Relaxed);
if hints::likely(state & WRITER_BIT == 0)
&& val
.compare_exchange_weak(
state,
state + 1,
atomic::Ordering::Acquire,
atomic::Ordering::Relaxed,
)
.is_ok()
{
return SlotReadGuard(val);
}
if hints::likely(spins < L1_CONTENTION) {
std::hint::spin_loop();
} else if spins < L2_CONTENTION {
std::thread::yield_now();
} else {
let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
std::thread::sleep(std::time::Duration::from_nanos(ns));
}
spins += 1;
}
}
#[inline(always)]
fn write_lock(&self, index: usize) -> SlotWriteGuard<'_> {
let val = &self.locks[index];
let mut spins = 0;
loop {
if val
.compare_exchange_weak(
0,
WRITER_BIT,
atomic::Ordering::Acquire,
atomic::Ordering::Relaxed,
)
.is_ok()
{
return SlotWriteGuard(val);
}
if hints::likely(spins < L1_CONTENTION) {
std::hint::spin_loop();
} else if spins < L2_CONTENTION {
std::thread::yield_now();
} else {
let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
std::thread::sleep(std::time::Duration::from_nanos(ns));
}
spins += 1;
}
}
}
struct SlotReadGuard<'a>(&'a atomic::AtomicU32);
impl Drop for SlotReadGuard<'_> {
#[inline(always)]
fn drop(&mut self) {
self.0.fetch_sub(1, atomic::Ordering::Release);
}
}
struct SlotWriteGuard<'a>(&'a atomic::AtomicU32);
impl Drop for SlotWriteGuard<'_> {
#[inline(always)]
fn drop(&mut self) {
self.0.store(0, atomic::Ordering::Release);
}
}
#[derive(Debug)]
struct MemMapInner {
cfg: MemMapCfg,
page_size: PageSize,
mmap: RwLock<Option<PlatformMemMap>>,
slot_locks: RwLock<SlotLocks>,
length: atomic::AtomicUsize,
completion: Arc<Completion>,
sync_condvar: Condvar,
sync_mutex: Mutex<bool>,
shutdown: atomic::AtomicBool,
is_parked: atomic::AtomicBool,
}
impl SyncTrigger for MemMapInner {
fn trigger_sync(&self) -> FrozenResult<()> {
if self.cfg.sync_interval.is_some() {
{
let mut guard = self.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
*guard = true;
}
self.sync_condvar.notify_one();
Ok(())
} else {
self.sync_internal()
}
}
}
impl MemMapInner {
fn sync_mmap_if_needed(&self, mmap: &PlatformMemMap, sync_len: usize) -> FrozenResult<()> {
let target_epoch = self.completion.read_current_epoch();
let durable_epoch = self.completion.read_durable_epoch();
if target_epoch <= durable_epoch {
return Ok(());
}
let sync_res = mmap.sync(sync_len);
match sync_res {
Ok(()) => {
self.completion.mark_epoch_as_durable(target_epoch);
self.completion.del_err();
self.completion.notify_all_listeners();
Ok(())
}
Err(e) => {
let e = e.with_module(self.cfg.module_id);
self.completion.set_err(e.clone());
self.completion.notify_all_listeners();
Err(e)
}
}
}
fn sync_internal(&self) -> FrozenResult<()> {
let guard = self.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return Ok(()),
};
let length = self.length.load(atomic::Ordering::Acquire);
self.sync_mmap_if_needed(mmap, length)
}
}
pub struct MemMap<T: Sized + Clone> {
cfg: MemMapCfg,
file: Option<File>,
inner: Arc<MemMapInner>,
bg_thread: Option<std::thread::JoinHandle<()>>,
_marker: PhantomData<T>,
}
impl<T> core::fmt::Debug for MemMap<T>
where
T: Sized + Clone,
{
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("MemMap").field("cfg", &self.cfg).field("inner", &self.inner).finish()
}
}
unsafe impl<T> Send for MemMap<T> where T: Sized + Clone {}
unsafe impl<T> Sync for MemMap<T> where T: Sized + Clone {}
impl<T> MemMap<T>
where
T: Sized + Clone,
{
fn validate_t(cfg: &MemMapCfg) -> FrozenResult<PageSize> {
let mid = cfg.module_id;
if core::mem::needs_drop::<T>() {
return err::default_error(err::DRP).with_module(mid);
}
let size = core::mem::size_of::<T>();
if size == 0 {
return err::default_error(err::ZRO).with_module(mid);
}
if size % 8 != 0 {
return err::default_error(err::SZE).with_module(mid);
}
let align = core::mem::align_of::<T>();
if align != 8 {
return err::default_error(err::ALN).with_module(mid);
}
let page_size = match cfg.page_size {
Some(ps) => ps.validate_platform().with_module(mid)?,
None => PlatformMemMap::page_size().with_module(mid)?,
};
if size > page_size.as_usize() {
return err::default_error(err::INV).with_module(mid);
}
if page_size.as_usize() % size != 0 {
return err::raw_error(
err::INV,
format!("page size {page_size} must be divisible by size_of::<T>() ({size})"),
)
.with_module(mid);
}
if cfg.initial_t == 0 {
return err::default_error(err::INV).with_module(mid);
}
let initial_bytes = match cfg.initial_t.checked_mul(size) {
Some(b) => b,
None => {
return err::raw_error(err::INV, "initial_t calculation overflowed usize")
.with_module(mid);
}
};
#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
{
if initial_bytes < 4096 || initial_bytes % 4096 != 0 {
return err::raw_error(
err::INV,
format!(
"initial allocation (initial_t * size_of::<T>() = {initial_bytes} bytes) must be at least 4096 and a multiple of 4096"
),
)
.with_module(mid);
}
}
#[cfg(not(all(target_os = "macos", target_arch = "aarch64")))]
{
if initial_bytes % page_size.as_usize() != 0 {
return err::raw_error(
err::INV,
format!(
"initial allocation (initial_t * size_of::<T>() = {initial_bytes} bytes) must be an exact multiple of page size ({page_size})"
),
)
.with_module(mid);
}
}
Ok(page_size)
}
fn validate_path(cfg: &MemMapCfg) -> FrozenResult<()> {
let mid = cfg.module_id;
if cfg.path.as_os_str().is_empty() {
return err::default_error(err::PTH).with_module(mid);
}
let stem = cfg.path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
if stem.is_empty() {
return err::default_error(err::PTH).with_module(mid);
}
if cfg.path.exists() && cfg.path.is_dir() {
return err::default_error(err::PTH).with_module(mid);
}
if let Some(parent) = cfg.path.parent() {
if !parent.as_os_str().is_empty() && (!parent.exists() || !parent.is_dir()) {
return err::default_error(err::DIR).with_module(mid);
}
}
Ok(())
}
fn file_cfg(cfg: &MemMapCfg) -> FileCfg {
FileCfg {
module_id: cfg.module_id,
path: cfg.versioned_path(),
buffer_size: core::mem::size_of::<T>(),
initial_available_buffers: cfg.initial_t,
sync_interval: None,
}
}
fn from_components(
cfg: MemMapCfg,
mmap: PlatformMemMap,
file: File,
page_size: PageSize,
) -> FrozenResult<Self> {
let length = mmap.length();
let completion = Arc::new(Completion::default());
let sync_interval = cfg.sync_interval;
let total_slots = length / core::mem::size_of::<T>();
let slot_locks = SlotLocks::new(total_slots);
let inner = Arc::new(MemMapInner {
cfg: cfg.clone(),
page_size,
mmap: RwLock::new(Some(mmap)),
slot_locks: RwLock::new(slot_locks),
length: atomic::AtomicUsize::new(length),
completion: completion.clone(),
sync_condvar: Condvar::new(),
sync_mutex: Mutex::new(false),
shutdown: atomic::AtomicBool::new(false),
is_parked: atomic::AtomicBool::new(false),
});
let weak_inner: std::sync::Weak<dyn SyncTrigger> =
Arc::downgrade(&inner) as std::sync::Weak<dyn SyncTrigger>;
completion.set_sync_trigger(weak_inner);
let bg_thread = if let Some(interval) = sync_interval {
let worker_inner = Arc::clone(&inner);
let handle = std::thread::Builder::new().name("frozen-mmap-sync-worker".into()).spawn(
move || {
let mut guard =
worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
while !worker_inner.shutdown.load(atomic::Ordering::Acquire) {
if *guard {
*guard = false;
drop(guard);
let _ = worker_inner.sync_internal();
guard =
worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
continue;
}
let target_epoch = worker_inner.completion.read_current_epoch();
let durable_epoch = worker_inner.completion.read_durable_epoch();
if target_epoch == durable_epoch {
worker_inner.is_parked.store(true, atomic::Ordering::Release);
if worker_inner.completion.read_current_epoch() != durable_epoch {
worker_inner.is_parked.store(false, atomic::Ordering::Release);
continue;
}
let new_guard = worker_inner
.sync_condvar
.wait(guard)
.unwrap_or_else(|e| e.into_inner());
guard = new_guard;
worker_inner.is_parked.store(false, atomic::Ordering::Release);
if worker_inner.shutdown.load(atomic::Ordering::Acquire) || *guard {
continue;
}
if worker_inner.completion.read_current_epoch()
== worker_inner.completion.read_durable_epoch()
{
continue;
}
let (new_guard, _) = worker_inner
.sync_condvar
.wait_timeout(guard, interval)
.unwrap_or_else(|e| e.into_inner());
guard = new_guard;
} else {
worker_inner.is_parked.store(false, atomic::Ordering::Release);
let (new_guard, _) = worker_inner
.sync_condvar
.wait_timeout(guard, interval)
.unwrap_or_else(|e| e.into_inner());
guard = new_guard;
}
if worker_inner.shutdown.load(atomic::Ordering::Acquire) {
break;
}
*guard = false;
drop(guard);
let _ = worker_inner.sync_internal();
guard = worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
}
},
);
let handle = match handle {
Ok(h) => h,
Err(spawn_err) => {
let mut mmap_guard = inner.mmap.write().unwrap_or_else(|e| e.into_inner());
let mut err = err::make_raw_error(err::FXE, spawn_err);
if let Some(m) = mmap_guard.take() {
if let Err(unmap_err) = m.unmap() {
err.add_suppressed(unmap_err);
}
}
return Err(err.with_module(cfg.module_id));
}
};
Some(handle)
} else {
None
};
Ok(Self { cfg, file: Some(file), inner, bg_thread, _marker: PhantomData })
}
pub fn new(cfg: MemMapCfg) -> FrozenResult<Self> {
let page_size = Self::validate_t(&cfg)?;
Self::validate_path(&cfg)?;
let vpath = cfg.versioned_path();
if vpath.exists() {
return err::default_error(err::EXS).with_module(cfg.module_id);
}
let file_cfg = Self::file_cfg(&cfg);
let file = match File::new(file_cfg) {
Ok(f) => f,
Err(e) if e.reason == err::EXS.reason => {
return err::default_error(err::EXS).with_module(cfg.module_id);
}
Err(e) => return Err(e),
};
let mmap = match PlatformMemMap::new(file.fd(), file.length()) {
Ok(m) => m,
Err(mut e) => {
if let Err(del_err) = file.delete() {
e.add_suppressed(del_err);
}
return Err(e.with_module(cfg.module_id));
}
};
Self::from_components(cfg, mmap, file, page_size)
}
pub fn open(cfg: MemMapCfg) -> FrozenResult<Self> {
let page_size = Self::validate_t(&cfg)?;
Self::validate_path(&cfg)?;
let vpath = cfg.versioned_path();
if !vpath.exists() {
return err::default_error(err::NEX).with_module(cfg.module_id);
}
if vpath.is_dir() {
return err::default_error(err::PTH).with_module(cfg.module_id);
}
let file_cfg = Self::file_cfg(&cfg);
let file = File::open(file_cfg)?;
let mmap = PlatformMemMap::new(file.fd(), file.length()).with_module(cfg.module_id)?;
Self::from_components(cfg, mmap, file, page_size)
}
pub fn open_or_create(cfg: MemMapCfg) -> FrozenResult<Self> {
let page_size = Self::validate_t(&cfg)?;
Self::validate_path(&cfg)?;
let vpath = cfg.versioned_path();
if vpath.exists() && vpath.is_dir() {
return err::default_error(err::PTH).with_module(cfg.module_id);
}
let file_cfg = Self::file_cfg(&cfg);
let file = File::open_or_create(file_cfg)?;
let mmap = PlatformMemMap::new(file.fd(), file.length()).with_module(cfg.module_id)?;
Self::from_components(cfg, mmap, file, page_size)
}
#[inline]
pub fn page_size() -> FrozenResult<PageSize> {
PlatformMemMap::page_size()
}
#[inline]
pub fn length(&self) -> usize {
self.inner.length.load(atomic::Ordering::Acquire)
}
pub fn grow(&self, pages: usize) -> FrozenResult<()> {
let mid = self.cfg.module_id;
if pages == 0 {
return Ok(());
}
let page_size = self.inner.page_size.as_usize();
let bytes_to_add = match pages.checked_mul(page_size) {
Some(bytes) => bytes,
None => {
return err::raw_error(err::GRW, "file growth calculation overflowed usize")
.with_module(mid);
}
};
let size = core::mem::size_of::<T>();
let additional_slots = bytes_to_add / size;
let file = match self.file.as_ref() {
Some(f) => f,
None => return err::default_error(err::INV).with_module(mid),
};
let mut slot_locks_guard = self.inner.slot_locks.write().unwrap_or_else(|e| e.into_inner());
let mut mmap_guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
let mmap = match mmap_guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
let curr_len = self.inner.length.load(atomic::Ordering::Acquire);
let new_length = match curr_len.checked_add(bytes_to_add) {
Some(len) => len,
None => {
return err::raw_error(err::GRW, "file growth calculation overflowed usize")
.with_module(mid);
}
};
self.inner.sync_mmap_if_needed(mmap, curr_len)?;
let old_mmap = match mmap_guard.take() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
if let Err(e) = old_mmap.unmap().with_module(mid) {
if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
*mmap_guard = Some(restored);
}
return Err(e);
}
if let Err(e) = file.grow(additional_slots).with_module(mid) {
if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
*mmap_guard = Some(restored);
}
return Err(e);
}
if let Err(e) = file.sync().with_module(mid) {
if let Ok(restored) = PlatformMemMap::new(file.fd(), new_length).with_module(mid) {
*mmap_guard = Some(restored);
self.inner.length.store(new_length, atomic::Ordering::Release);
slot_locks_guard.grow(additional_slots);
}
return Err(e);
}
let new_mmap = match PlatformMemMap::new(file.fd(), new_length).with_module(mid) {
Ok(m) => m,
Err(e) => {
if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
*mmap_guard = Some(restored);
}
return Err(e);
}
};
*mmap_guard = Some(new_mmap);
slot_locks_guard.grow(additional_slots);
self.inner.length.store(new_length, atomic::Ordering::Release);
self.inner.completion.increment_current_epoch();
Ok(())
}
#[inline]
pub fn cfg(&self) -> &MemMapCfg {
&self.cfg
}
#[inline]
pub fn current_epoch(&self) -> TEpoch {
self.inner.completion.read_current_epoch()
}
#[inline]
pub fn durable_epoch(&self) -> TEpoch {
self.inner.completion.read_durable_epoch()
}
#[inline]
pub fn sync(&self) -> FrozenResult<()> {
self.inner.sync_internal()
}
#[inline(always)]
pub fn read<R, F>(&self, index: usize, f: F) -> FrozenResult<R>
where
F: FnOnce(&T) -> R,
{
let mid = self.cfg.module_id;
let size = core::mem::size_of::<T>();
let offset = match index.checked_mul(size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
return err::default_error(err::HCF).with_module(mid);
}
let _slot_guard = slot_locks_guard.read_lock(index);
let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
let ptr = unsafe { mmap.as_ptr::<T>(offset) };
let target_ref = unsafe { &*ptr };
let res = f(target_ref);
drop(guard);
drop(_slot_guard);
drop(slot_locks_guard);
Ok(res)
}
#[inline(always)]
pub fn read_exact<Target, R, F>(&self, start: usize, f: F) -> FrozenResult<R>
where
F: FnOnce(&Target) -> R,
{
let mid = self.cfg.module_id;
let slot_size = core::mem::size_of::<T>();
let target_size = core::mem::size_of::<Target>();
if hints::unlikely(target_size == 0) {
return err::default_error(err::ZRO).with_module(mid);
}
if hints::unlikely(target_size % slot_size != 0) {
return err::default_error(err::SZE).with_module(mid);
}
if hints::unlikely(core::mem::align_of::<Target>() > core::mem::align_of::<T>()) {
return err::default_error(err::ALN).with_module(mid);
}
let n = target_size / slot_size;
let end_slot = match start.checked_add(n) {
Some(end) => end,
None => return err::default_error(err::INV).with_module(mid),
};
let offset = match start.checked_mul(slot_size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
let end_offset = match end_slot.checked_mul(slot_size) {
Some(end) => end,
None => return err::default_error(err::INV).with_module(mid),
};
let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
if hints::unlikely(end_offset > self.length()) {
return err::default_error(err::HCF).with_module(mid);
}
let mut slot_guards = Vec::with_capacity(n);
for slot in start..end_slot {
slot_guards.push(slot_locks_guard.read_lock(slot));
}
let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
let ptr = unsafe { mmap.as_ptr::<Target>(offset) };
if hints::unlikely((ptr as usize) % core::mem::align_of::<Target>() != 0) {
return err::default_error(err::ALN).with_module(mid);
}
let target_ref = unsafe { &*ptr };
let res = f(target_ref);
drop(guard);
drop(slot_guards);
drop(slot_locks_guard);
Ok(res)
}
#[inline(always)]
pub fn write<R, F>(&self, index: usize, f: F) -> FrozenResult<(AckTicket, R)>
where
F: FnOnce(&mut T) -> R,
{
let mid = self.cfg.module_id;
let size = core::mem::size_of::<T>();
let offset = match index.checked_mul(size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
return err::default_error(err::HCF).with_module(mid);
}
let _slot_guard = slot_locks_guard.write_lock(index);
let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
let res = f(unsafe { &mut *ptr });
drop(guard);
drop(_slot_guard);
drop(slot_locks_guard);
let epoch = self.inner.completion.increment_current_epoch();
if self.inner.is_parked.load(atomic::Ordering::Acquire) {
let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
self.inner.sync_condvar.notify_one();
}
}
Ok((AckTicket::new(epoch, self.inner.completion.clone()), res))
}
#[inline(always)]
pub fn write_exact<Target, R, F>(&self, start: usize, f: F) -> FrozenResult<(AckTicket, R)>
where
F: FnOnce(&mut Target) -> R,
{
let mid = self.cfg.module_id;
let slot_size = core::mem::size_of::<T>();
let target_size = core::mem::size_of::<Target>();
if hints::unlikely(target_size == 0) {
return err::default_error(err::ZRO).with_module(mid);
}
if hints::unlikely(target_size % slot_size != 0) {
return err::default_error(err::SZE).with_module(mid);
}
if hints::unlikely(core::mem::align_of::<Target>() > core::mem::align_of::<T>()) {
return err::default_error(err::ALN).with_module(mid);
}
let n = target_size / slot_size;
let end_slot = match start.checked_add(n) {
Some(end) => end,
None => return err::default_error(err::INV).with_module(mid),
};
let offset = match start.checked_mul(slot_size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
let end_offset = match end_slot.checked_mul(slot_size) {
Some(end) => end,
None => return err::default_error(err::INV).with_module(mid),
};
let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
if hints::unlikely(end_offset > self.length()) {
return err::default_error(err::HCF).with_module(mid);
}
let mut slot_guards = Vec::with_capacity(n);
for slot in start..end_slot {
slot_guards.push(slot_locks_guard.write_lock(slot));
}
let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
let ptr = unsafe { mmap.as_mut_ptr::<Target>(offset) };
if hints::unlikely((ptr as usize) % core::mem::align_of::<Target>() != 0) {
return err::default_error(err::ALN).with_module(mid);
}
let target_ref = unsafe { &mut *ptr };
let res = f(target_ref);
drop(guard);
drop(slot_guards);
drop(slot_locks_guard);
let epoch = self.inner.completion.increment_current_epoch();
if self.inner.is_parked.load(atomic::Ordering::Acquire) {
let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
self.inner.sync_condvar.notify_one();
}
}
Ok((AckTicket::new(epoch, self.inner.completion.clone()), res))
}
#[cfg(test)]
pub(crate) fn is_parked(&self) -> bool {
self.inner.is_parked.load(atomic::Ordering::Acquire)
}
pub fn delete(mut self) -> FrozenResult<()> {
let mid = self.cfg.module_id;
self.inner.shutdown.store(true, atomic::Ordering::Release);
{
let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
*guard = true;
}
self.inner.sync_condvar.notify_all();
if let Some(handle) = self.bg_thread.take() {
let _ = handle.join();
}
let mut guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
let unmap_res =
if let Some(mmap) = guard.take() { mmap.unmap().with_module(mid) } else { Ok(()) };
let file = match self.file.take() {
Some(f) => f,
None => {
unmap_res?;
return err::default_error(err::INV).with_module(mid);
}
};
let delete_res = file.delete().with_module(mid);
match (unmap_res, delete_res) {
(Err(mut unmap_err), Err(del_err)) => {
unmap_err.add_suppressed(del_err);
Err(unmap_err)
}
(Err(unmap_err), Ok(())) => Err(unmap_err),
(Ok(()), Err(del_err)) => Err(del_err),
(Ok(()), Ok(())) => Ok(()),
}
}
}
impl<T> MemMap<T>
where
T: Sized + Clone + PartialEq,
{
pub fn compare_and_write(
&self,
expected: &T,
val: &T,
index: usize,
) -> FrozenResult<AckTicket> {
let mid = self.cfg.module_id;
let size = core::mem::size_of::<T>();
let offset = match index.checked_mul(size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
return err::default_error(err::HCF).with_module(mid);
}
let _slot_guard = slot_locks_guard.write_lock(index);
let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
let current = unsafe { ptr.read() };
if current != *expected {
drop(guard);
drop(_slot_guard);
drop(slot_locks_guard);
return err::raw_error(
err::CAS,
format!("state conflict at index {}: slot data has changed", index),
)
.with_module(mid);
}
unsafe { core::ptr::write(ptr, val.clone()) };
drop(guard);
drop(_slot_guard);
drop(slot_locks_guard);
let epoch = self.inner.completion.increment_current_epoch();
if self.inner.is_parked.load(atomic::Ordering::Acquire) {
let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
self.inner.sync_condvar.notify_one();
}
}
Ok(AckTicket::new(epoch, self.inner.completion.clone()))
}
#[inline]
pub fn new_tx(&self) -> MemMapTransaction<'_, T> {
MemMapTransaction { memmap: self, ops: Vec::new() }
}
}
#[derive(Debug)]
pub struct MemMapTransaction<'a, T: Sized + Clone + PartialEq> {
memmap: &'a MemMap<T>,
ops: Vec<(usize, T, T)>,
}
impl<T> MemMapTransaction<'_, T>
where
T: Sized + Clone + PartialEq,
{
#[inline(always)]
pub fn write(&mut self, expected: &T, new: &T, index: usize) -> FrozenResult<&mut Self> {
let mid = self.memmap.cfg.module_id;
if let Some(err) = self.memmap.inner.completion.get_err() {
return Err(err);
}
if let Some((last_idx, _, _)) = self.ops.last() {
if index <= *last_idx {
return err::raw_error(
err::HCF,
"tx writes must be strictly ordered, with no more then single ops on given index",
)
.with_module(mid);
}
}
let size = core::mem::size_of::<T>();
let offset = match index.checked_mul(size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.memmap.length())) {
return err::default_error(err::HCF).with_module(mid);
}
self.ops.push((index, expected.clone(), new.clone()));
Ok(self)
}
pub fn commit(self) -> FrozenResult<AckTicket> {
let mid = self.memmap.cfg.module_id;
if self.ops.is_empty() {
return err::raw_error(err::INV, "empty transaction").with_module(mid);
}
if let Some(err) = self.memmap.inner.completion.get_err() {
return Err(err);
}
let size = core::mem::size_of::<T>();
let slot_locks_guard =
self.memmap.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
let current_length = self.memmap.length();
for (index, _, _) in &self.ops {
let offset = match index.checked_mul(size) {
Some(off) => off,
None => return err::default_error(err::INV).with_module(mid),
};
if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > current_length)) {
return err::default_error(err::HCF).with_module(mid);
}
}
let mut slot_guards = Vec::with_capacity(self.ops.len());
for (index, _, _) in &self.ops {
slot_guards.push(slot_locks_guard.write_lock(*index));
}
let guard = self.memmap.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
let mmap = match guard.as_ref() {
Some(m) => m,
None => return err::default_error(err::INV).with_module(mid),
};
for (index, expected, _) in &self.ops {
let offset = *index * size;
let ptr = unsafe { mmap.as_ptr::<T>(offset) };
let current = unsafe { ptr.read() };
if current != *expected {
drop(guard);
drop(slot_guards);
drop(slot_locks_guard);
return err::raw_error(
err::CAS,
format!("state conflict at index {}: slot data has changed", index),
)
.with_module(mid);
}
}
for (index, _, new) in self.ops {
let offset = index * size;
let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
unsafe { core::ptr::write(ptr, new) };
}
drop(guard);
drop(slot_guards);
drop(slot_locks_guard);
let epoch = self.memmap.inner.completion.increment_current_epoch();
if self.memmap.inner.is_parked.load(atomic::Ordering::Acquire) {
let _guard = self.memmap.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
if self.memmap.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
self.memmap.inner.sync_condvar.notify_one();
}
}
Ok(AckTicket::new(epoch, self.memmap.inner.completion.clone()))
}
#[inline]
pub fn len(&self) -> usize {
self.ops.len()
}
#[inline]
pub fn is_empty(&self) -> bool {
self.ops.is_empty()
}
}
impl<T> Drop for MemMap<T>
where
T: Sized + Clone,
{
fn drop(&mut self) {
self.inner.shutdown.store(true, atomic::Ordering::Release);
{
let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
*guard = true;
}
self.inner.sync_condvar.notify_all();
if let Some(handle) = self.bg_thread.take() {
let _ = handle.join();
}
let _ = self.inner.sync_internal();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
const MID: u8 = 0;
const INIT_T: usize = 512;
#[derive(Debug, Clone)]
#[repr(C, align(8))]
struct ValidT {
a: u64,
b: u64,
}
#[derive(Debug, Clone)]
#[repr(C)]
struct BadAlignT {
a: u32,
b: u32,
}
#[derive(Debug, Clone)]
#[repr(C)]
struct BadSizeT {
a: [u8; 0x0C],
}
#[derive(Debug, Clone)]
#[repr(C, align(8))]
struct DropT(u64);
impl Drop for DropT {
fn drop(&mut self) {}
}
#[derive(Debug, Clone)]
#[repr(C, align(8))]
struct ZstT;
#[derive(Debug, Clone)]
#[repr(C, align(8))]
struct OversizedT {
data: [u8; 0x20000],
}
fn tmp_cfg() -> (tempfile::TempDir, MemMapCfg) {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("tmp_memmap");
let cfg = MemMapCfg {
module_id: MID,
path,
version: 1,
initial_t: INIT_T,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
(dir, cfg)
}
#[test]
fn ok_validate_t() {
let (_dir, cfg) = tmp_cfg();
assert!(MemMap::<u64>::validate_t(&cfg).is_ok());
let mut cfg_valid = cfg.clone();
cfg_valid.initial_t = 256;
assert!(MemMap::<ValidT>::validate_t(&cfg_valid).is_ok());
}
#[test]
fn err_validate_t_invariants() {
let (_dir, cfg) = tmp_cfg();
assert_eq!(MemMap::<DropT>::validate_t(&cfg).unwrap_err().reason, err::DRP.reason);
assert_eq!(MemMap::<BadAlignT>::validate_t(&cfg).unwrap_err().reason, err::ALN.reason);
assert_eq!(MemMap::<ZstT>::validate_t(&cfg).unwrap_err().reason, err::ZRO.reason);
assert_eq!(MemMap::<BadSizeT>::validate_t(&cfg).unwrap_err().reason, err::SZE.reason);
assert_eq!(MemMap::<OversizedT>::validate_t(&cfg).unwrap_err().reason, err::INV.reason);
let mut zero_cfg = cfg.clone();
zero_cfg.initial_t = 0;
assert_eq!(MemMap::<u64>::validate_t(&zero_cfg).unwrap_err().reason, err::INV.reason);
let mut unaligned_cfg = cfg.clone();
unaligned_cfg.initial_t = 511;
assert_eq!(MemMap::<u64>::validate_t(&unaligned_cfg).unwrap_err().reason, err::INV.reason);
}
#[test]
fn ok_new_sets_file_sync_interval_none() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
assert_eq!(mmap.cfg().sync_interval, Some(Duration::from_millis(0x64)));
}
#[test]
fn err_new_when_file_exists() {
let (_dir, cfg) = tmp_cfg();
let _mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
let err = MemMap::<u64>::new(cfg).unwrap_err();
assert_eq!(err.reason, err::EXS.reason);
}
#[test]
fn err_new_when_t_invalid() {
let (_dir, cfg) = tmp_cfg();
assert_eq!(MemMap::<DropT>::new(cfg.clone()).unwrap_err().reason, err::DRP.reason);
assert!(!cfg.versioned_path().exists());
}
#[test]
fn ok_open() {
let (_dir, cfg) = tmp_cfg();
{
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
}
let opened = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(opened.length(), core::mem::size_of::<u64>() * INIT_T);
}
#[test]
fn err_open_when_file_missing() {
let (_dir, cfg) = tmp_cfg();
let err = MemMap::<u64>::open(cfg).unwrap_err();
assert_eq!(err.reason, err::NEX.reason);
}
#[test]
fn ok_open_or_create() {
let (_dir, cfg) = tmp_cfg();
{
let mmap = MemMap::<u64>::open_or_create(cfg.clone()).unwrap();
assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
}
let reopened = MemMap::<u64>::open_or_create(cfg).unwrap();
assert_eq!(reopened.length(), core::mem::size_of::<u64>() * INIT_T);
}
#[test]
fn err_path_when_dir_missing() {
let dir = tempfile::tempdir().unwrap();
let non_existent_dir = dir.path().join("missing_dir_xyz");
let cfg = MemMapCfg {
module_id: MID,
path: non_existent_dir.join("file.bin"),
version: 1,
initial_t: INIT_T,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
assert_eq!(err_new.reason, err::DIR.reason);
let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
assert_eq!(err_open.reason, err::DIR.reason);
let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
assert_eq!(err_open_or_create.reason, err::DIR.reason);
}
#[test]
fn err_path_when_parent_is_not_dir() {
let dir = tempfile::tempdir().unwrap();
let file_path = dir.path().join("regular_file");
std::fs::write(&file_path, b"test").unwrap();
let cfg = MemMapCfg {
module_id: MID,
path: file_path.join("subfile.bin"),
version: 1,
initial_t: INIT_T,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
assert_eq!(err_new.reason, err::DIR.reason);
let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
assert_eq!(err_open.reason, err::DIR.reason);
let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
assert_eq!(err_open_or_create.reason, err::DIR.reason);
}
#[test]
fn err_path_when_empty() {
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from(""),
version: 1,
initial_t: INIT_T,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
assert_eq!(err_new.reason, err::PTH.reason);
let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
assert_eq!(err_open.reason, err::PTH.reason);
let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
assert_eq!(err_open_or_create.reason, err::PTH.reason);
}
#[test]
fn err_path_when_is_directory() {
let dir = tempfile::tempdir().unwrap();
let cfg = MemMapCfg {
module_id: MID,
path: dir.path().to_path_buf(),
version: 1,
initial_t: INIT_T,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
assert_eq!(err_new.reason, err::PTH.reason);
let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
assert_eq!(err_open.reason, err::PTH.reason);
let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
assert_eq!(err_open_or_create.reason, err::PTH.reason);
}
#[test]
fn err_path_when_no_filename() {
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from("/"),
version: 1,
initial_t: INIT_T,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
assert_eq!(err_new.reason, err::PTH.reason);
let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
assert_eq!(err_open.reason, err::PTH.reason);
let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
assert_eq!(err_open_or_create.reason, err::PTH.reason);
}
#[test]
fn ok_delete() {
let (_dir, cfg) = tmp_cfg();
let path = cfg.versioned_path();
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert!(path.exists());
assert!(mmap.delete().is_ok());
assert!(!path.exists());
}
#[test]
fn ok_versioned_path_resolution() {
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from("data.bin"),
version: 1,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("data_1.bin"));
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from("datastore"),
version: 2,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("datastore_2"));
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from("archive.tar.gz"),
version: 3,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("archive.tar_3.gz"));
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from("/var/data/store.db"),
version: 0x0A,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("/var/data/store_10.db"));
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from("dir/sub/file.dat"),
version: 0,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("dir/sub/file_0.dat"));
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from(".env"),
version: 5,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from(".env_5"));
let cfg = MemMapCfg {
module_id: MID,
path: std::path::PathBuf::from(".config.json"),
version: 1,
initial_t: 2,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg.versioned_path(), std::path::PathBuf::from(".config_1.json"));
}
#[test]
fn ok_versioned_path_and_multiversion() {
let dir = tempfile::tempdir().unwrap();
let base_path = dir.path().join("store.dat");
let cfg_v1 = MemMapCfg {
module_id: MID,
path: base_path.clone(),
version: 1,
initial_t: 512,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg_v1.versioned_path(), dir.path().join("store_1.dat"));
let cfg_v2 = MemMapCfg {
module_id: MID,
path: base_path.clone(),
version: 2,
initial_t: 256,
sync_interval: Some(Duration::from_millis(0x64)),
page_size: None,
};
assert_eq!(cfg_v2.versioned_path(), dir.path().join("store_2.dat"));
let mmap_v1 = MemMap::<u64>::new(cfg_v1.clone()).unwrap();
let mmap_v2 = MemMap::<ValidT>::new(cfg_v2.clone()).unwrap();
assert!(cfg_v1.versioned_path().exists());
assert!(cfg_v2.versioned_path().exists());
assert_eq!(mmap_v1.length(), core::mem::size_of::<u64>() * 512);
assert_eq!(mmap_v2.length(), core::mem::size_of::<ValidT>() * 256);
assert!(mmap_v1.delete().is_ok());
assert!(!cfg_v1.versioned_path().exists());
assert!(cfg_v2.versioned_path().exists());
assert!(mmap_v2.delete().is_ok());
assert!(!cfg_v2.versioned_path().exists());
}
#[test]
fn ok_write_read_single() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
let (ticket, _) = mmap.write(0, |v| *v = 0xDEAD_BEEF_CAFE_BABE).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xDEAD_BEEF_CAFE_BABE);
assert_eq!(ticket.epoch(), 1);
}
#[test]
fn ok_write_read_multiple_indices() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
for i in 0..INIT_T {
let val = (i as u64 + 1).wrapping_mul(0x1111_2222_3333_4444);
mmap.write(i, |v| *v = val).unwrap();
}
for i in 0..INIT_T {
let expected = (i as u64 + 1).wrapping_mul(0x1111_2222_3333_4444);
assert_eq!(mmap.read(i, |v| *v).unwrap(), expected);
}
}
#[test]
fn ok_write_read_struct() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<ValidT>::new(cfg).unwrap();
let val0 = ValidT { a: 0x0A, b: 0x14 };
let val1 = ValidT { a: 0x64, b: 0xC8 };
mmap.write(0, |v| *v = val0).unwrap();
mmap.write(1, |v| *v = val1).unwrap();
let r0 = mmap.read(0, |v| v.clone()).unwrap();
let r1 = mmap.read(1, |v| v.clone()).unwrap();
assert_eq!(r0.a, 0x0A);
assert_eq!(r0.b, 0x14);
assert_eq!(r1.a, 0x64);
assert_eq!(r1.b, 0xC8);
}
#[test]
fn ok_read_callback_zero_copy_projection() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<ValidT>::new(cfg).unwrap();
let val = ValidT { a: 0x1234, b: 0x5678 };
mmap.write(0, |v| *v = val).unwrap();
let a_val = mmap.read(0, |v| v.a).unwrap();
assert_eq!(a_val, 0x1234);
let sum = mmap.read(0, |v| v.a + v.b).unwrap();
assert_eq!(sum, 0x1234 + 0x5678);
let is_match = mmap.read(0, |v| v.b == 0x5678).unwrap();
assert!(is_match);
}
#[test]
fn ok_read_callback_unwind_releases_lock() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
mmap.write(0, |v| *v = 0x42).unwrap();
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _ = mmap.read(0, |_| panic!("simulated panic in read callback"));
}));
assert!(result.is_err());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x42);
assert!(mmap.write(0, |v| *v = 0x99).is_ok());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x99);
}
#[test]
fn ok_write_callback_in_place_mutation() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<ValidT>::new(cfg).unwrap();
mmap.write(0, |v| {
v.a = 0x1111;
v.b = 0x2222;
})
.unwrap();
mmap.write(0, |v| {
v.b = 0x9999;
})
.unwrap();
let val = mmap.read(0, |v| v.clone()).unwrap();
assert_eq!(val.a, 0x1111);
assert_eq!(val.b, 0x9999);
}
#[test]
fn ok_write_callback_returns_derived_value() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let (ticket, old_prev) = mmap
.write(0, |v| {
let prev = *v;
*v = 42;
prev
})
.unwrap();
assert_eq!(old_prev, 0);
assert_eq!(ticket.epoch(), 1);
assert_eq!(mmap.read(0, |v| *v).unwrap(), 42);
}
#[test]
fn ok_write_callback_unwind_releases_lock() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let epoch_before = mmap.current_epoch();
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _ = mmap.write(0, |v| {
*v = 0xBAD;
panic!("intentional panic inside write callback");
});
}));
assert!(result.is_err());
assert_eq!(mmap.current_epoch(), epoch_before);
assert!(mmap.write(0, |v| *v = 0x600D).is_ok());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x600D);
}
#[test]
fn ok_concurrent_writes_and_reads_single_slot() {
let (_dir, mut cfg) = tmp_cfg();
cfg.initial_t = 256;
let mmap = Arc::new(MemMap::<ValidT>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(4));
let mut handles = Vec::new();
for t in 0..2 {
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for i in 0..0xC8 {
let val = ((t + 1) as u64) << 0x20 | (i as u64);
let _ = m.write(0, |node| {
node.a = val;
node.b = val;
});
}
}));
}
for _ in 0..2 {
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for _ in 0..0x190 {
let _ = m.read(0, |node| {
assert_eq!(node.a, node.b);
});
}
}));
}
for h in handles {
h.join().unwrap();
}
mmap.read(0, |node| {
assert_eq!(node.a, node.b);
})
.unwrap();
}
#[test]
fn err_write_read_out_of_bounds() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let err_r = mmap.read(INIT_T, |_| ()).unwrap_err();
assert_eq!(err_r.reason, err::HCF.reason);
let err_w = mmap.write(INIT_T, |_| ()).unwrap_err();
assert_eq!(err_w.reason, err::HCF.reason);
let err_r_large = mmap.read(INIT_T + 0x64, |_| ()).unwrap_err();
assert_eq!(err_r_large.reason, err::HCF.reason);
let err_w_large = mmap.write(INIT_T + 0x64, |_| ()).unwrap_err();
assert_eq!(err_w_large.reason, err::HCF.reason);
}
#[test]
fn err_write_read_overflow_index() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let err_r = mmap.read(usize::MAX, |_| ()).unwrap_err();
assert_eq!(err_r.reason, err::INV.reason);
let err_w = mmap.write(usize::MAX, |_| ()).unwrap_err();
assert_eq!(err_w.reason, err::INV.reason);
}
mod read_exact {
use super::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C)]
struct TestNode {
id: u64,
flags: u64,
keys: [u64; 2],
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C)]
struct TestBadSizeTarget([u8; 12]);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C, align(16))]
struct TestOveralignedTarget {
_a: u64,
_b: u64,
}
#[test]
fn ok_read_exact_single_and_multi_slot() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
mmap.write(0, |v| *v = 0x1111).unwrap();
mmap.write(1, |v| *v = 0x2222).unwrap();
mmap.write(2, |v| *v = 0x3333).unwrap();
mmap.write(3, |v| *v = 0x4444).unwrap();
let single = mmap.read_exact(0, |val: &u64| *val).unwrap();
assert_eq!(single, 0x1111);
let node = mmap.read_exact(0, |n: &TestNode| *n).unwrap();
assert_eq!(node, TestNode { id: 0x1111, flags: 0x2222, keys: [0x3333, 0x4444] });
let keys = mmap.read_exact(2, |k: &[u64; 2]| *k).unwrap();
assert_eq!(keys, [0x3333, 0x4444]);
let flag_is_set = mmap.read_exact(0, |n: &TestNode| n.flags == 0x2222).unwrap();
assert!(flag_is_set);
}
#[test]
fn err_read_exact_invariants() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let err_zro = mmap.read_exact::<(), _, _>(0, |_| ()).unwrap_err();
assert_eq!(err_zro.reason, err::ZRO.reason);
let err_sze = mmap.read_exact::<TestBadSizeTarget, _, _>(0, |_| ()).unwrap_err();
assert_eq!(err_sze.reason, err::SZE.reason);
let err_aln = mmap.read_exact::<TestOveralignedTarget, _, _>(0, |_| ()).unwrap_err();
assert_eq!(err_aln.reason, err::ALN.reason);
let err_inv = mmap.read_exact::<u64, _, _>(usize::MAX, |_| ()).unwrap_err();
assert_eq!(err_inv.reason, err::INV.reason);
let err_inv_add =
mmap.read_exact::<TestNode, _, _>(usize::MAX - 1, |_| ()).unwrap_err();
assert_eq!(err_inv_add.reason, err::INV.reason);
let err_hcf_start = mmap.read_exact::<u64, _, _>(INIT_T, |_| ()).unwrap_err();
assert_eq!(err_hcf_start.reason, err::HCF.reason);
let err_hcf_end = mmap.read_exact::<TestNode, _, _>(INIT_T - 2, |_| ()).unwrap_err();
assert_eq!(err_hcf_end.reason, err::HCF.reason);
}
#[test]
fn ok_read_exact_concurrent_readers() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
mmap.write(0, |v| *v = 0xCAFE).unwrap();
mmap.write(1, |v| *v = 0xBABE).unwrap();
mmap.write(2, |v| *v = 0x1234).unwrap();
mmap.write(3, |v| *v = 0x5678).unwrap();
let barrier = Arc::new(std::sync::Barrier::new(4));
let mut handles = Vec::new();
for _ in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for _ in 0..0x1F4 {
let id = mmap_clone.read_exact(0, |n: &TestNode| n.id).unwrap();
assert_eq!(id, 0xCAFE);
let k1 = mmap_clone.read_exact(0, |n: &TestNode| n.keys[1]).unwrap();
assert_eq!(k1, 0x5678);
}
}));
}
for h in handles {
h.join().unwrap();
}
}
#[test]
fn ok_read_exact_concurrent_writers_isolation() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(5));
let mut handles = Vec::new();
{
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x1F4 {
let v = i as u64;
let _ = mmap_clone.write(0, |val| *val = v).unwrap();
let _ = mmap_clone.write(1, |val| *val = v).unwrap();
}
}));
}
for _ in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for _ in 0..0x1F4 {
mmap_clone
.read_exact(0, |pair: &[u64; 2]| {
assert!(pair[0] <= 0x1F4 && pair[1] <= 0x1F4);
})
.unwrap();
}
}));
}
for h in handles {
h.join().unwrap();
}
}
#[test]
fn ok_read_exact_unwind_releases_locks() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
mmap.write(0, |v| *v = 0x1111).unwrap();
mmap.write(1, |v| *v = 0x2222).unwrap();
let panic_res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _ = mmap.read_exact(0, |_pair: &[u64; 2]| {
panic!("intentional panic inside read_exact callback");
});
}));
assert!(panic_res.is_err());
let (ticket, _) = mmap.write(0, |v| *v = 0x9999).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999);
assert_eq!(ticket.epoch(), 3);
let (ticket2, _) = mmap.write(1, |v| *v = 0x8888).unwrap();
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x8888);
assert_eq!(ticket2.epoch(), 4);
}
}
mod write_exact {
use super::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C)]
struct TestNode {
id: u64,
flags: u64,
keys: [u64; 2],
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C)]
struct TestBadSizeTarget([u8; 12]);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C, align(16))]
struct TestOveralignedTarget {
_a: u64,
_b: u64,
}
#[test]
fn ok_write_exact_single_and_multi_slot() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let (ticket_single, res) = mmap
.write_exact(0, |val: &mut u64| {
*val = 0xAAAA;
42u32
})
.unwrap();
assert_eq!(res, 42);
assert_eq!(ticket_single.epoch(), 1);
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xAAAA);
let (ticket_multi, _) = mmap
.write_exact(0, |node: &mut TestNode| {
node.id = 0x1111;
node.flags = 0x2222;
node.keys = [0x3333, 0x4444];
})
.unwrap();
assert_eq!(ticket_multi.epoch(), 2);
assert_eq!(
mmap.read_exact(0, |n: &TestNode| *n).unwrap(),
TestNode { id: 0x1111, flags: 0x2222, keys: [0x3333, 0x4444] }
);
let (ticket_partial, _) = mmap
.write_exact(2, |keys: &mut [u64; 2]| {
keys[0] = 0x5555;
keys[1] = 0x6666;
})
.unwrap();
assert_eq!(ticket_partial.epoch(), 3);
assert_eq!(mmap.read(2, |v| *v).unwrap(), 0x5555);
assert_eq!(mmap.read(3, |v| *v).unwrap(), 0x6666);
}
#[test]
fn err_write_exact_invariants() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let err_zro = mmap.write_exact::<(), _, _>(0, |_| ()).unwrap_err();
assert_eq!(err_zro.reason, err::ZRO.reason);
let err_sze = mmap.write_exact::<TestBadSizeTarget, _, _>(0, |_| ()).unwrap_err();
assert_eq!(err_sze.reason, err::SZE.reason);
let err_aln = mmap.write_exact::<TestOveralignedTarget, _, _>(0, |_| ()).unwrap_err();
assert_eq!(err_aln.reason, err::ALN.reason);
let err_inv = mmap.write_exact::<u64, _, _>(usize::MAX, |_| ()).unwrap_err();
assert_eq!(err_inv.reason, err::INV.reason);
let err_inv_add =
mmap.write_exact::<TestNode, _, _>(usize::MAX - 1, |_| ()).unwrap_err();
assert_eq!(err_inv_add.reason, err::INV.reason);
let err_hcf_start = mmap.write_exact::<u64, _, _>(INIT_T, |_| ()).unwrap_err();
assert_eq!(err_hcf_start.reason, err::HCF.reason);
let err_hcf_end = mmap.write_exact::<TestNode, _, _>(INIT_T - 2, |_| ()).unwrap_err();
assert_eq!(err_hcf_end.reason, err::HCF.reason);
}
#[test]
fn ok_write_exact_concurrent_writers() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(2));
let mut handles = Vec::new();
{
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x1F4 {
let _ = mmap_clone
.write_exact(0, |node: &mut TestNode| {
node.id = i as u64;
node.flags = (i as u64) + 1;
node.keys = [(i as u64) + 2, (i as u64) + 3];
})
.unwrap();
}
}));
}
{
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x1F4 {
let _ = mmap_clone
.write_exact(4, |node: &mut TestNode| {
node.id = (i as u64) * 10;
node.flags = (i as u64) * 10 + 1;
node.keys = [(i as u64) * 10 + 2, (i as u64) * 10 + 3];
})
.unwrap();
}
}));
}
for h in handles {
h.join().unwrap();
}
let node0 = mmap.read_exact(0, |n: &TestNode| *n).unwrap();
assert_eq!(node0.id, 0x1F3);
assert_eq!(node0.flags, 0x1F4);
assert_eq!(node0.keys, [0x1F5, 0x1F6]);
let node1 = mmap.read_exact(4, |n: &TestNode| *n).unwrap();
assert_eq!(node1.id, 0x1F3 * 10);
assert_eq!(node1.flags, 0x1F3 * 10 + 1);
assert_eq!(node1.keys, [0x1F3 * 10 + 2, 0x1F3 * 10 + 3]);
}
#[test]
fn ok_write_exact_concurrent_readers_writers_isolation() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(5));
let mut handles = Vec::new();
{
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x1F4 {
let v = i as u64;
let _ = mmap_clone
.write_exact(0, |pair: &mut [u64; 2]| {
pair[0] = v;
pair[1] = v;
})
.unwrap();
}
}));
}
for _ in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for _ in 0..0x1F4 {
mmap_clone
.read_exact(0, |pair: &[u64; 2]| {
assert_eq!(pair[0], pair[1]);
})
.unwrap();
}
}));
}
for h in handles {
h.join().unwrap();
}
}
#[test]
fn ok_write_exact_unwind_releases_locks() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let panic_res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _ = mmap.write_exact(0, |_pair: &mut [u64; 2]| {
panic!("intentional panic inside write_exact callback");
});
}));
assert!(panic_res.is_err());
let (ticket, _) = mmap.write(0, |v| *v = 0x9999).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999);
assert_eq!(ticket.epoch(), 1);
let (ticket2, _) = mmap.write(1, |v| *v = 0x8888).unwrap();
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x8888);
assert_eq!(ticket2.epoch(), 2);
}
#[test]
fn ok_write_exact_durability() {
let (_dir, cfg) = tmp_cfg();
{
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
let (ticket, _) = mmap
.write_exact(0, |node: &mut TestNode| {
node.id = 0xDEAD;
node.flags = 0xBEEF;
node.keys = [0xCAFE, 0xBABE];
})
.unwrap();
mmap.inner.sync_internal().unwrap();
assert_eq!(ticket.epoch(), 1);
}
let reopened = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(
reopened.read_exact(0, |n: &TestNode| *n).unwrap(),
TestNode { id: 0xDEAD, flags: 0xBEEF, keys: [0xCAFE, 0xBABE] }
);
}
}
#[test]
fn ok_durability_across_reopen() {
let (_dir, cfg) = tmp_cfg();
{
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
mmap.write(1, |v| *v = 0xCAFE_BABE_1234_5678).unwrap();
mmap.inner.sync_internal().unwrap();
}
let reopened = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(reopened.read(1, |v| *v).unwrap(), 0xCAFE_BABE_1234_5678);
}
#[test]
fn ok_background_sync_advances_epoch() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_millis(0x19));
let mmap = MemMap::<u64>::new(cfg).unwrap();
let (ticket, _) = mmap.write(0, |v| *v = 0x1234).unwrap();
let epoch = ticket.wait().unwrap();
assert_eq!(epoch, ticket.epoch());
assert!(ticket.is_durable());
assert!(mmap.durable_epoch() >= ticket.epoch());
}
#[test]
fn ok_ticket_force_wakes_bg_worker() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_secs(0x0A));
let mmap = MemMap::<u64>::new(cfg).unwrap();
let (ticket, _) = mmap.write(0, |v| *v = 0xABCD).unwrap();
assert!(!ticket.is_durable());
let start = std::time::Instant::now();
let epoch = ticket.force().unwrap();
let elapsed = start.elapsed();
assert_eq!(epoch, ticket.epoch());
assert!(ticket.is_durable());
assert!(elapsed < Duration::from_secs(2));
}
#[test]
fn ok_drop_persists_unflushed_writes() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_secs(0x1E));
{
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
let (_ticket, _) = mmap.write(0, |v| *v = 0xBEEF_CAFE).unwrap();
drop(mmap);
}
let reopened = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(reopened.read(0, |v| *v).unwrap(), 0xBEEF_CAFE);
}
#[test]
fn ok_concurrent_slot_locking() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_secs(0x3C));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(8));
let mut handles = Vec::new();
for t in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x1F4 {
let val = ((t + 1) as u64) << 0x20 | (i as u64);
let _ = mmap_clone.write(0, |slot| *slot = val).unwrap();
}
}));
}
for _ in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for _ in 0..0x3E8 {
let val = mmap_clone.read(0, |v| *v).unwrap();
let thread_id = (val >> 0x20) as usize;
assert!(thread_id <= 4, "read value must be valid without torn reads");
}
}));
}
for h in handles {
h.join().unwrap();
}
let barrier2 = Arc::new(std::sync::Barrier::new(4));
let mut read_handles = Vec::new();
for _ in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier2);
read_handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for _ in 0..0x1F4 {
mmap_clone.read(0, |_| ()).unwrap();
}
}));
}
for h in read_handles {
h.join().unwrap();
}
}
#[test]
fn err_concurrent_new_same_path() {
let (_dir, cfg) = tmp_cfg();
let barrier = Arc::new(std::sync::Barrier::new(2));
let t1 = {
let cfg = cfg.clone();
let b = Arc::clone(&barrier);
std::thread::spawn(move || {
b.wait();
MemMap::<u64>::new(cfg)
})
};
let t2 = {
let b = Arc::clone(&barrier);
std::thread::spawn(move || {
b.wait();
MemMap::<u64>::new(cfg)
})
};
let r1 = t1.join().unwrap();
let r2 = t2.join().unwrap();
match (r1, r2) {
(Ok(m), Err(e)) | (Err(e), Ok(m)) => {
assert_eq!(e.reason, err::EXS.reason);
assert!(m.delete().is_ok());
}
(Ok(_), Ok(_)) => {
panic!("Concurrent MemMap::new on identical path must not both succeed");
}
(Err(e1), Err(e2)) => {
panic!("Unexpected error on both threads: {:?}, {:?}", e1, e2);
}
}
}
#[test]
fn err_concurrent_open_same_path() {
let (_dir, cfg) = tmp_cfg();
{
let m = MemMap::<u64>::new(cfg.clone()).unwrap();
assert_eq!(m.length(), core::mem::size_of::<u64>() * INIT_T);
}
let barrier = Arc::new(std::sync::Barrier::new(2));
let t1 = {
let cfg = cfg.clone();
let b = Arc::clone(&barrier);
std::thread::spawn(move || {
b.wait();
MemMap::<u64>::open(cfg)
})
};
let t2 = {
let b = Arc::clone(&barrier);
std::thread::spawn(move || {
b.wait();
MemMap::<u64>::open(cfg)
})
};
let r1 = t1.join().unwrap();
let r2 = t2.join().unwrap();
match (r1, r2) {
(Ok(m), Err(e)) | (Err(e), Ok(m)) => {
assert_eq!(e.domain, 0x08, "Error domain must be File (0x08)");
assert_eq!(
e.reason, 0x1C,
"Concurrent open should yield LCK (0x1C) from kernel flock: {:?}",
e
);
assert!(m.delete().is_ok());
}
(Ok(_), Ok(_)) => {
panic!("Exclusive advisory lock violation: both concurrent open calls succeeded");
}
(Err(e1), Err(e2)) => {
panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
}
}
}
#[test]
fn ok_reopen_after_drop_releases_lock() {
let (_dir, cfg) = tmp_cfg();
let m1 = MemMap::<u64>::new(cfg.clone()).unwrap();
m1.write(0, |v| *v = 0x1122_3344).unwrap();
drop(m1);
let m2 = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(m2.read(0, |v| *v).unwrap(), 0x1122_3344);
assert!(m2.delete().is_ok());
}
#[test]
fn ok_delete_and_recreate_cycle() {
let (_dir, cfg) = tmp_cfg();
for i in 0..4 {
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
let val = (i as u64 + 1) * 0xAA55;
mmap.write(0, |v| *v = val).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), val);
assert!(mmap.delete().is_ok());
assert!(!cfg.versioned_path().exists());
}
}
#[test]
fn ok_concurrent_distinct_slots_no_contention() {
let (_dir, mut cfg) = tmp_cfg();
const THREADS: usize = 8;
cfg.sync_interval = Some(Duration::from_secs(0x3C));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(THREADS));
let mut handles = Vec::with_capacity(THREADS);
for t in 0..THREADS {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x3E8 {
let val = (t as u64) << 0x20 | (i as u64);
let _ = mmap_clone.write(t, |slot| *slot = val).unwrap();
let read_val = mmap_clone.read(t, |v| *v).unwrap();
assert_eq!(read_val >> 0x20, t as u64);
}
}));
}
for h in handles {
h.join().unwrap();
}
for t in 0..THREADS {
let val = mmap.read(t, |v| *v).unwrap();
assert_eq!(val >> 0x20, t as u64);
}
}
#[test]
fn ok_stress_interleaved_reads_writes() {
let (_dir, mut cfg) = tmp_cfg();
const NUM_SLOTS: usize = 0x10;
const NUM_THREADS: usize = 8;
cfg.sync_interval = Some(Duration::from_millis(0x32));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(NUM_THREADS));
let mut handles = Vec::with_capacity(NUM_THREADS);
for t in 0..NUM_THREADS {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x5DC {
let slot = (t + i) % NUM_SLOTS;
if i % 2 == 0 {
let val = ((t + 1) as u64) << 0x20 | (i as u64);
let _ = mmap_clone.write(slot, |s| *s = val).unwrap();
} else {
mmap_clone.read(slot, |_| ()).unwrap();
}
}
}));
}
for h in handles {
h.join().unwrap();
}
assert!(mmap.inner.sync_internal().is_ok());
}
#[test]
fn ok_concurrent_writers_with_background_sync() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_millis(5));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(4));
let mut handles = Vec::new();
for t in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
let mut tickets = Vec::with_capacity(0x32);
for i in 0..0x32 {
let val = (t as u64) * 0x3E8 + i as u64;
tickets.push(mmap_clone.write(t, |slot| *slot = val).unwrap().0);
std::thread::sleep(Duration::from_millis(1));
}
for ticket in tickets {
let _ = ticket.wait().unwrap();
assert!(ticket.is_durable());
}
}));
}
for h in handles {
h.join().unwrap();
}
assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
}
#[test]
fn ok_concurrent_ticket_force_stress() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_secs(0x1E));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(4));
let mut handles = Vec::new();
for t in 0..4 {
let mmap_clone = Arc::clone(&mmap);
let barrier_clone = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
barrier_clone.wait();
for i in 0..0x14 {
let val = (t as u64) * 0x64 + i as u64;
let (ticket, _) = mmap_clone.write(t, |slot| *slot = val).unwrap();
let durable_epoch = ticket.force().unwrap();
assert!(ticket.is_durable());
assert!(durable_epoch >= ticket.epoch());
}
}));
}
for h in handles {
h.join().unwrap();
}
}
#[test]
fn ok_exact_boundary_indices() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert!(mmap.write(INIT_T - 1, |v| *v = 0xF00D).is_ok());
assert_eq!(mmap.read(INIT_T - 1, |v| *v).unwrap(), 0xF00D);
assert_eq!(mmap.write(INIT_T, |_| ()).unwrap_err().reason, err::HCF.reason);
assert_eq!(mmap.read(INIT_T, |_| ()).unwrap_err().reason, err::HCF.reason);
}
#[test]
fn ok_large_slot_count_initialization() {
let (_dir, mut cfg) = tmp_cfg();
const LARGE_COUNT: usize = 0x2800;
cfg.initial_t = LARGE_COUNT;
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert_eq!(mmap.length(), LARGE_COUNT * core::mem::size_of::<u64>());
assert!(mmap.write(0, |v| *v = 1).is_ok());
assert!(mmap.write(0x1388, |v| *v = 0x1388).is_ok());
assert!(mmap.write(LARGE_COUNT - 1, |v| *v = 0x27FF).is_ok());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 1);
assert_eq!(mmap.read(0x1388, |v| *v).unwrap(), 0x1388);
assert_eq!(mmap.read(LARGE_COUNT - 1, |v| *v).unwrap(), 0x27FF);
}
#[test]
fn ok_new_with_sync_interval_none() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = None;
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert!(mmap.bg_thread.is_none());
let (ticket, _) = mmap.write(0, |v| *v = 0x55AA).unwrap();
assert!(!ticket.is_durable());
mmap.sync().unwrap();
assert!(ticket.is_durable());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x55AA);
}
#[test]
fn ok_ticket_force_with_sync_interval_none() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = None;
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert!(mmap.bg_thread.is_none());
let (ticket, _) = mmap.write(1, |v| *v = 0x1122).unwrap();
assert!(!ticket.is_durable());
let epoch = ticket.force().unwrap();
assert_eq!(epoch, ticket.epoch());
assert!(ticket.is_durable());
assert!(mmap.durable_epoch() >= ticket.epoch());
}
#[test]
fn ok_drop_with_sync_interval_none() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = None;
{
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
assert!(mmap.bg_thread.is_none());
let (_ticket, _) = mmap.write(0, |v| *v = 0xCAFE_F00D).unwrap();
drop(mmap);
}
let reopened = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(reopened.read(0, |v| *v).unwrap(), 0xCAFE_F00D);
}
#[test]
fn ok_bg_worker_parks_when_idle_and_unparks_on_write() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_millis(0x0F));
let mmap = MemMap::<u64>::new(cfg).unwrap();
let start = std::time::Instant::now();
while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
std::thread::sleep(Duration::from_millis(5));
}
assert!(mmap.is_parked());
let (ticket, _) = mmap.write(0, |v| *v = 0x9999).unwrap();
assert!(!mmap.is_parked());
let epoch = ticket.wait().unwrap();
assert_eq!(epoch, ticket.epoch());
assert!(ticket.is_durable());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999);
let start = std::time::Instant::now();
while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
std::thread::sleep(Duration::from_millis(5));
}
assert!(mmap.is_parked());
}
#[test]
fn ok_batched_writes_while_active_remain_unparked() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_millis(0x1E));
let mmap = MemMap::<u64>::new(cfg).unwrap();
let start = std::time::Instant::now();
while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
std::thread::sleep(Duration::from_millis(5));
}
assert!(mmap.is_parked());
let (t1, _) = mmap.write(0, |v| *v = 0x1111).unwrap();
assert!(!mmap.is_parked());
let (t2, _) = mmap.write(1, |v| *v = 0x2222).unwrap();
let (t3, _) = mmap.write(2, |v| *v = 0x3333).unwrap();
assert!(!mmap.is_parked());
t1.wait().unwrap();
t2.wait().unwrap();
t3.wait().unwrap();
assert!(t1.is_durable());
assert!(t2.is_durable());
assert!(t3.is_durable());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x1111);
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x2222);
assert_eq!(mmap.read(2, |v| *v).unwrap(), 0x3333);
}
#[test]
fn ok_concurrent_stress_park_unpark_no_lost_wakeups() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_millis(5));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
const THREADS: usize = 4;
const WRITES_PER_THREAD: usize = 0x19;
let mut handles = Vec::with_capacity(THREADS);
for t in 0..THREADS {
let mmap_clone = Arc::clone(&mmap);
handles.push(std::thread::spawn(move || {
let mut tickets = Vec::with_capacity(WRITES_PER_THREAD);
for i in 0..WRITES_PER_THREAD {
let val = (t as u64) << 0x20 | (i as u64);
let slot = t;
let (ticket, _) = mmap_clone.write(slot, |s| *s = val).unwrap();
tickets.push(ticket);
if i % 5 == 0 {
std::thread::sleep(Duration::from_millis(0x0F));
}
}
for ticket in tickets {
ticket.wait().unwrap();
assert!(ticket.is_durable());
}
}));
}
for h in handles {
h.join().unwrap();
}
assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
}
#[test]
fn ok_spurious_wakeup_resumes_park_immediately() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_secs(0x0A));
let mmap = MemMap::<u64>::new(cfg).unwrap();
let start = std::time::Instant::now();
while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
std::thread::sleep(Duration::from_millis(5));
}
assert!(mmap.is_parked());
mmap.inner.sync_condvar.notify_one();
std::thread::sleep(Duration::from_millis(0x0A));
let start = std::time::Instant::now();
while !mmap.is_parked() && start.elapsed() < Duration::from_millis(0x1F4) {
std::thread::sleep(Duration::from_millis(5));
}
assert!(mmap.is_parked());
}
#[test]
fn ok_delete_cleans_up_file_when_mmap_already_unmapped() {
let (_dir, cfg) = tmp_cfg();
let path = cfg.versioned_path();
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert!(path.exists());
{
let mut guard = mmap.inner.mmap.write().unwrap();
let _ = guard.take();
}
assert!(mmap.delete().is_ok());
assert!(!path.exists());
}
mod memmap_grow {
use super::*;
#[test]
fn ok_grow_updates_length() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let initial_len = INIT_T * core::mem::size_of::<u64>();
let page_size = mmap.inner.page_size.as_usize();
assert_eq!(mmap.length(), initial_len);
mmap.grow(4).unwrap();
assert_eq!(mmap.length(), initial_len + 4 * page_size);
}
#[test]
fn ok_grow_zero_count() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let initial_len = mmap.length();
mmap.grow(0).unwrap();
assert_eq!(mmap.length(), initial_len);
}
#[test]
fn ok_grow_extends_with_zero() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let page_slots = mmap.inner.page_size.as_usize() / core::mem::size_of::<u64>();
mmap.grow(1).unwrap();
for idx in INIT_T..(INIT_T + page_slots) {
assert_eq!(mmap.read(idx, |v| *v).unwrap(), 0u64);
}
}
#[test]
fn ok_grow_preserves_existing_data() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let page_slots = mmap.inner.page_size.as_usize() / core::mem::size_of::<u64>();
for i in 0..INIT_T {
mmap.write(i, |v| *v = 0xCAFE_0000 + i as u64).unwrap();
}
mmap.grow(1).unwrap();
for i in 0..INIT_T {
assert_eq!(mmap.read(i, |v| *v).unwrap(), 0xCAFE_0000 + i as u64);
}
for i in INIT_T..(INIT_T + page_slots) {
mmap.write(i, |v| *v = 0xBEEF_0000 + i as u64).unwrap();
assert_eq!(mmap.read(i, |v| *v).unwrap(), 0xBEEF_0000 + i as u64);
}
}
#[test]
fn ok_grow_syncs_dirty_pages_when_curr_epoch_gt_durable() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = None;
let mmap = MemMap::<u64>::new(cfg).unwrap();
let (ticket, _) = mmap.write(1, |v| *v = 0x1234_5678u64).unwrap();
assert!(!ticket.is_durable());
assert!(mmap.current_epoch() > mmap.durable_epoch());
mmap.grow(2).unwrap();
assert!(ticket.is_durable());
assert!(mmap.durable_epoch() >= ticket.epoch());
}
#[test]
fn ok_grow_when_already_durable() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = None;
let mmap = MemMap::<u64>::new(cfg).unwrap();
let (ticket, _) = mmap.write(0, |v| *v = 0x9999u64).unwrap();
mmap.sync().unwrap();
assert!(ticket.is_durable());
assert_eq!(mmap.current_epoch(), mmap.durable_epoch());
mmap.grow(2).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999u64);
}
#[test]
fn err_grow_overflow_usize() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let err = mmap.grow(usize::MAX).unwrap_err();
assert_eq!(err.reason, err::GRW.reason);
}
#[test]
fn ok_grow_multiple_times() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let page_size = mmap.inner.page_size.as_usize();
mmap.write(0, |v| *v = 0x0Au64).unwrap();
mmap.grow(2).unwrap();
mmap.write(INIT_T, |v| *v = 0x14u64).unwrap();
mmap.grow(3).unwrap();
mmap.write(INIT_T + 2, |v| *v = 0x1Eu64).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x0Au64);
assert_eq!(mmap.read(INIT_T, |v| *v).unwrap(), 0x14u64);
assert_eq!(mmap.read(INIT_T + 2, |v| *v).unwrap(), 0x1Eu64);
assert_eq!(mmap.length(), (INIT_T * core::mem::size_of::<u64>()) + 5 * page_size);
}
#[test]
fn ok_concurrent_grow_and_reads_writes() {
let (_dir, mut cfg) = tmp_cfg();
cfg.sync_interval = Some(Duration::from_millis(0x0A));
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let page_size = mmap.inner.page_size.as_usize();
let num_writers = 4;
let mut handles = Vec::new();
for w in 0..num_writers {
let m = Arc::clone(&mmap);
handles.push(std::thread::spawn(move || {
for i in 0..0x32 {
let _ = m.write(w % INIT_T, |v| *v = (w * 0x3E8 + i) as u64);
m.read(w % INIT_T, |_| ()).unwrap();
}
}));
}
let m_grower = Arc::clone(&mmap);
let grower_handle = std::thread::spawn(move || {
for _ in 0..3 {
std::thread::sleep(Duration::from_millis(5));
m_grower.grow(2).unwrap();
}
});
for h in handles {
h.join().unwrap();
}
grower_handle.join().unwrap();
assert_eq!(mmap.length(), (INIT_T * core::mem::size_of::<u64>()) + 6 * page_size);
}
}
mod thread_failure_lifecycle {
use super::*;
#[test]
fn err_new_failure_paths_spawn_no_sync_thread() {
let (_dir, cfg) = tmp_cfg();
std::fs::write(cfg.versioned_path(), b"preexisting").unwrap();
let res = MemMap::<u64>::new(cfg);
assert!(res.is_err());
}
#[test]
fn err_open_failure_paths_spawn_no_sync_thread() {
let (_dir, mut cfg) = tmp_cfg();
cfg.path = std::path::PathBuf::from("/non/existent/path/memmap.bin");
let res = MemMap::<u64>::open(cfg);
assert!(res.is_err());
}
#[test]
fn ok_thread_shutdown_on_panic_unwinding() {
let (_dir, cfg) = tmp_cfg();
let thread_inner = Arc::new(atomic::AtomicBool::new(false));
let thread_inner_clone = Arc::clone(&thread_inner);
let handle = std::thread::spawn(move || {
let mmap = MemMap::<u64>::new(cfg).unwrap();
let _inner = Arc::clone(&mmap.inner);
thread_inner_clone.store(true, atomic::Ordering::Release);
panic!("intentional test panic to trigger Drop unwinding");
});
assert!(handle.join().is_err());
assert!(thread_inner.load(atomic::Ordering::Acquire));
}
#[test]
fn ok_thread_shutdown_in_delete_even_if_already_unmapped() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
{
let mut guard = mmap.inner.mmap.write().unwrap();
let _ = guard.take();
}
assert!(mmap.delete().is_ok());
}
#[test]
fn ok_grow_failure_keeps_thread_healthy() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert!(mmap.grow(usize::MAX).is_err());
let (ticket, _) = mmap.write(0, |v| *v = 0x2Au64).unwrap();
mmap.sync().unwrap();
assert!(ticket.is_durable());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x2Au64);
}
}
mod memmap_transactions {
use super::*;
#[test]
fn ok_single_slot_cas_success_and_conflict() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
let t1 = mmap.compare_and_write(&0, &0x64, 0).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x64);
assert_eq!(mmap.current_epoch(), 1);
let err = mmap.compare_and_write(&0, &0xC8, 0).unwrap_err();
assert_eq!(err.reason, err::CAS.reason);
assert!(err.context.contains("slot data has changed"));
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x64);
assert_eq!(mmap.current_epoch(), 1);
let t2 = mmap.compare_and_write(&0x64, &0xC8, 0).unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xC8);
assert_eq!(mmap.current_epoch(), 2);
assert!(t2.epoch() > t1.epoch());
}
#[test]
fn ok_tx_multiple_writes_single_epoch() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let epoch_before = mmap.current_epoch();
let mut tx = mmap.new_tx();
assert!(tx.is_empty());
tx.write(&0, &0x6F, 0).unwrap();
tx.write(&0, &0xDE, 1).unwrap();
tx.write(&0, &0x14D, 2).unwrap();
assert_eq!(tx.len(), 3);
let ticket = tx.commit().unwrap();
let epoch_after = mmap.current_epoch();
assert_eq!(epoch_after, epoch_before + 1);
assert_eq!(ticket.epoch(), epoch_after);
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x6F);
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0xDE);
assert_eq!(mmap.read(2, |v| *v).unwrap(), 0x14D);
ticket.wait().unwrap();
assert!(ticket.is_durable());
}
#[test]
fn err_tx_cas_conflict_aborts_all_with_zero_side_effects() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
mmap.write(1, |v| *v = 0x3E7).unwrap();
let epoch_before = mmap.current_epoch();
let mut tx = mmap.new_tx();
tx.write(&0, &0x0A, 0).unwrap(); tx.write(&0, &0x14, 1).unwrap(); tx.write(&0, &0x1E, 2).unwrap();
let err = tx.commit().unwrap_err();
assert_eq!(err.reason, err::CAS.reason);
assert!(err.context.contains("state conflict at index 1"));
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x3E7);
assert_eq!(mmap.read(2, |v| *v).unwrap(), 0);
assert_eq!(mmap.current_epoch(), epoch_before);
}
#[test]
fn ok_tx_retry_after_conflict() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
mmap.write(1, |v| *v = 0x2A).unwrap();
let mut tx1 = mmap.new_tx();
tx1.write(&0, &0x64, 0).unwrap();
tx1.write(&0, &0xC8, 1).unwrap();
assert_eq!(tx1.commit().unwrap_err().reason, err::CAS.reason);
let curr_val = mmap.read(1, |v| *v).unwrap();
assert_eq!(curr_val, 0x2A);
let mut tx2 = mmap.new_tx();
tx2.write(&0, &0x64, 0).unwrap();
tx2.write(&0x2A, &0xC8, 1).unwrap();
let ticket = tx2.commit().unwrap();
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x64);
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0xC8);
ticket.wait().unwrap();
assert!(ticket.is_durable());
}
#[test]
fn err_tx_out_of_order() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let mut tx = mmap.new_tx();
tx.write(&0, &0x0A, 2).unwrap();
let err = tx.write(&0, &0x14, 1).unwrap_err();
assert_eq!(err.reason, err::HCF.reason);
assert!(err.context.contains("strictly ordered"));
}
#[test]
fn err_tx_duplicate_index() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let mut tx = mmap.new_tx();
tx.write(&0, &0x0A, 1).unwrap();
let err = tx.write(&0, &0x14, 1).unwrap_err();
assert_eq!(err.reason, err::HCF.reason);
assert!(err.context.contains("strictly ordered"));
}
#[test]
fn err_tx_out_of_bounds() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let mut tx = mmap.new_tx();
let err = tx.write(&0, &0x0A, INIT_T).unwrap_err();
assert_eq!(err.reason, err::HCF.reason);
}
#[test]
fn err_tx_empty_commit() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let tx = mmap.new_tx();
let err = tx.commit().unwrap_err();
assert_eq!(err.reason, err::INV.reason);
assert!(err.context.contains("empty transaction"));
}
#[test]
fn ok_tx_discard_on_drop() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let epoch_before = mmap.current_epoch();
{
let mut tx = mmap.new_tx();
tx.write(&0, &0x309, 0).unwrap();
tx.write(&0, &0x378, 1).unwrap();
}
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
assert_eq!(mmap.read(1, |v| *v).unwrap(), 0);
assert_eq!(mmap.current_epoch(), epoch_before);
}
#[test]
fn ok_tx_concurrent_non_overlapping() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let mut handles = Vec::new();
for thread_id in 0..4u64 {
let m = mmap.clone();
handles.push(std::thread::spawn(move || {
let base = (thread_id as usize) * 4;
let mut tx = m.new_tx();
tx.write(&0, &(thread_id * 0x0A + 1), base).unwrap();
tx.write(&0, &(thread_id * 0x0A + 2), base + 1).unwrap();
tx.write(&0, &(thread_id * 0x0A + 3), base + 2).unwrap();
tx.write(&0, &(thread_id * 0x0A + 4), base + 3).unwrap();
let ticket = tx.commit().unwrap();
ticket.wait().unwrap();
}));
}
for h in handles {
h.join().unwrap();
}
for thread_id in 0..4u64 {
let base = (thread_id as usize) * 4;
assert_eq!(mmap.read(base, |v| *v).unwrap(), thread_id * 0x0A + 1);
assert_eq!(mmap.read(base + 1, |v| *v).unwrap(), thread_id * 0x0A + 2);
assert_eq!(mmap.read(base + 2, |v| *v).unwrap(), thread_id * 0x0A + 3);
assert_eq!(mmap.read(base + 3, |v| *v).unwrap(), thread_id * 0x0A + 4);
}
}
#[test]
fn ok_tx_durability_across_reopen() {
let (_dir, cfg) = tmp_cfg();
{
let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
let mut tx = mmap.new_tx();
tx.write(&0, &0xAAAA_BBBB, 0).unwrap();
tx.write(&0, &0xCCCC_DDDD, 1).unwrap();
let ticket = tx.commit().unwrap();
ticket.wait().unwrap();
assert!(ticket.is_durable());
}
{
let reopened = MemMap::<u64>::open(cfg).unwrap();
assert_eq!(reopened.read(0, |v| *v).unwrap(), 0xAAAA_BBBB);
assert_eq!(reopened.read(1, |v| *v).unwrap(), 0xCCCC_DDDD);
}
}
#[test]
fn ok_tx_unparks_idle_bg_worker() {
let (_dir, cfg) = tmp_cfg();
let mmap = MemMap::<u64>::new(cfg).unwrap();
let mut parked = false;
for _ in 0..0x64 {
if mmap.is_parked() {
parked = true;
break;
}
std::thread::sleep(Duration::from_millis(0x0A));
}
assert!(parked);
let mut tx = mmap.new_tx();
tx.write(&0, &0x22B, 0).unwrap();
let ticket = tx.commit().unwrap();
ticket.wait().unwrap();
assert!(ticket.is_durable());
assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x22B);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C)]
struct TxNode {
id: u64,
flags: u64,
keys: [u64; 2],
}
#[test]
fn ok_concurrent_tx_read_exact_and_write_exact_same_slots() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(6));
let mut handles = Vec::new();
for t in 0..2 {
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for i in 0..0xC8 {
let val = ((t + 1) as u64) << 0x20 | (i as u64);
let _ = m
.write_exact(0, |node: &mut TxNode| {
node.id = val;
node.flags = val;
node.keys = [val, val];
})
.unwrap();
}
}));
}
for t in 2..4 {
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for i in 0..0xC8 {
let new_val = ((t + 1) as u64) << 0x20 | (i as u64);
let current_node = m.read_exact(0, |n: &TxNode| *n).unwrap();
let mut tx = m.new_tx();
if tx.write(¤t_node.id, &new_val, 0).is_ok()
&& tx.write(¤t_node.flags, &new_val, 1).is_ok()
&& tx.write(¤t_node.keys[0], &new_val, 2).is_ok()
&& tx.write(¤t_node.keys[1], &new_val, 3).is_ok()
{
let _ = tx.commit();
}
}
}));
}
for _ in 0..2 {
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for _ in 0..0x190 {
m.read_exact(0, |node: &TxNode| {
assert_eq!(node.id, node.flags);
assert_eq!(node.id, node.keys[0]);
assert_eq!(node.id, node.keys[1]);
})
.unwrap();
}
}));
}
for h in handles {
h.join().unwrap();
}
let final_node = mmap.read_exact(0, |n: &TxNode| *n).unwrap();
assert_eq!(final_node.id, final_node.flags);
assert_eq!(final_node.id, final_node.keys[0]);
assert_eq!(final_node.id, final_node.keys[1]);
}
#[test]
fn ok_tx_retry_loop_against_concurrent_write_exact_and_read_exact() {
let (_dir, cfg) = tmp_cfg();
let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
let barrier = Arc::new(std::sync::Barrier::new(4));
let mut handles = Vec::new();
{
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for i in 0..0x64 {
let _ = m
.write_exact(0, |pair: &mut [u64; 2]| {
pair[0] = i as u64;
pair[1] = i as u64;
})
.unwrap();
}
}));
}
{
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
let mut successful_commits = 0;
while successful_commits < 10 {
let current = m.read_exact(0, |pair: &[u64; 2]| *pair).unwrap();
let new_val = 0x8000_0000 + successful_commits;
let mut tx = m.new_tx();
if tx.write(¤t[0], &new_val, 0).is_ok()
&& tx.write(¤t[1], &new_val, 1).is_ok()
&& tx.commit().is_ok()
{
successful_commits += 1;
}
}
}));
}
for _ in 0..2 {
let m = Arc::clone(&mmap);
let b = Arc::clone(&barrier);
handles.push(std::thread::spawn(move || {
b.wait();
for _ in 0..0xC8 {
m.read_exact(0, |pair: &[u64; 2]| {
assert_eq!(pair[0], pair[1]);
})
.unwrap();
}
}));
}
for h in handles {
h.join().unwrap();
}
mmap.read_exact(0, |pair: &[u64; 2]| {
assert_eq!(pair[0], pair[1]);
})
.unwrap();
}
}
mod memmap_page_size {
use super::*;
#[test]
fn ok_page_size_returns_valid_power_of_two() {
let page_size = MemMap::<u64>::page_size().unwrap();
assert!(page_size.as_usize() >= 0x1000);
assert!(page_size.is_power_of_two());
assert_eq!(page_size.as_usize(), page_size.as_u32() as usize);
let matched_bytes = match page_size {
PageSize::K4 => 4096,
PageSize::K16 => 16384,
PageSize::K64 => 65536,
};
assert_eq!(matched_bytes, page_size.as_usize());
}
#[test]
fn ok_page_size_consistency() {
let ps1 = MemMap::<u64>::page_size().unwrap();
let ps2 = MemMap::<u64>::page_size().unwrap();
let ps3 = MemMap::<ValidT>::page_size().unwrap();
assert_eq!(ps1, ps2);
assert_eq!(ps1, ps3);
}
#[test]
fn ok_page_size_across_threads() {
let expected = MemMap::<u64>::page_size().unwrap();
let mut handles = Vec::new();
for _ in 0..8 {
handles.push(std::thread::spawn(move || {
let ps = MemMap::<u64>::page_size().unwrap();
assert_eq!(ps, expected);
}));
}
for h in handles {
h.join().unwrap();
}
}
}
}