1use crate::{
32 ack::{AckTicket, Completion, SyncTrigger, TEpoch},
33 error::{BindModule, ErrCode, FrozenError, FrozenResult},
34 file::{File, FileCfg},
35 hints,
36};
37use core::marker::PhantomData;
38use std::sync::{Arc, Condvar, Mutex, RwLock, atomic};
39
40mod interface;
41
42#[cfg(any(target_os = "linux", target_os = "macos"))]
43mod posix;
44
45#[cfg(target_os = "windows")]
46mod windows;
47
48pub(in crate::memmap) use interface::MemMapInterface;
49
50#[cfg(any(target_os = "linux", target_os = "macos"))]
51pub(in crate::memmap) type PlatformMemMap = posix::POSIXMemMap;
52
53#[cfg(target_os = "windows")]
54pub(in crate::memmap) type PlatformMemMap = windows::WINMemMap;
55
56pub(in crate::memmap) mod err {
58 use super::{ErrCode, FrozenError, FrozenResult};
59
60 const ERRDOMAIN: u8 = 0x10;
62
63 pub const HCF: ErrCode = ErrCode::new(0x02, "hault and catch fire");
65
66 pub const UNK: ErrCode = ErrCode::new(0x04, "unknown error");
68
69 pub const NMM: ErrCode = ErrCode::new(0x06, "not enough memory available on the device");
71
72 pub const SYN: ErrCode = ErrCode::new(0x08, "failed to sync/flush data to storage device");
74
75 pub const PRM: ErrCode = ErrCode::new(0x0A, "missing permissions for IO");
77
78 pub const FXE: ErrCode = ErrCode::new(0x10, "unable to spawn flush_tx");
80
81 pub const ZRO: ErrCode = ErrCode::new(0x0C, "type T must not be zero sized");
83
84 pub const DRP: ErrCode = ErrCode::new(0x12, "type T must not implement `Drop`");
86
87 pub const ALN: ErrCode = ErrCode::new(0x14, "type T must be 8-bytes aligned");
89
90 pub const SZE: ErrCode = ErrCode::new(0x16, "`size_of::<T>()` must be multiple of 8 bytes");
92
93 pub const INV: ErrCode =
96 ErrCode::new(0x18, "`size_of::<T>()` must not exceed system page size");
97
98 pub const DIR: ErrCode = ErrCode::new(0x1A, "directory does not exist");
100
101 pub const PTH: ErrCode = ErrCode::new(0x1C, "invalid path to file");
103
104 pub const EXS: ErrCode = ErrCode::new(0x1E, "file already exists");
106
107 pub const NEX: ErrCode = ErrCode::new(0x20, "file does not exist");
109
110 pub const GRW: ErrCode = ErrCode::new(0x22, "unable to grow memory-mapped file");
112
113 pub const CAS: ErrCode = ErrCode::new(0x24, "compare and swap check failed");
115
116 #[inline]
117 pub(in crate::memmap) fn raw_error<R, E: std::fmt::Display>(
118 code: ErrCode,
119 error: E,
120 ) -> FrozenResult<R> {
121 let err = FrozenError::unbound_raw(ERRDOMAIN, code, error);
122 Err(err)
123 }
124
125 #[inline]
126 pub(in crate::memmap) fn default_error<R>(code: ErrCode) -> FrozenResult<R> {
127 let err = FrozenError::unbound(ERRDOMAIN, code, "");
128 Err(err)
129 }
130
131 #[inline]
132 pub(in crate::memmap) fn make_raw_error<E: std::fmt::Display>(
133 code: ErrCode,
134 error: E,
135 ) -> FrozenError {
136 FrozenError::unbound_raw(ERRDOMAIN, code, error)
137 }
138}
139
140#[derive(Debug, Clone)]
161pub struct MemMapCfg {
162 pub module_id: u8,
164
165 pub path: std::path::PathBuf,
174
175 pub version: u32,
184
185 pub initial_t: usize,
189
190 pub sync_interval: Option<std::time::Duration>,
196}
197
198impl MemMapCfg {
199 #[inline]
203 fn versioned_path(&self) -> std::path::PathBuf {
204 let parent = self.path.parent();
205 let stem = self.path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
206 let extension = self.path.extension().and_then(|e| e.to_str());
207
208 let versioned_filename = match extension {
209 Some(ext) if !ext.is_empty() => format!("{}_{}.{}", stem, self.version, ext),
210 _ => format!("{}_{}", stem, self.version),
211 };
212
213 match parent {
214 Some(p) if !p.as_os_str().is_empty() => p.join(versioned_filename),
215 _ => std::path::PathBuf::from(versioned_filename),
216 }
217 }
218}
219
220const WRITER_BIT: u32 = 0x8000_0000;
221const L1_CONTENTION: usize = 0x10;
222const L2_CONTENTION: usize = 0x20;
223
224#[derive(Debug)]
226struct SlotLocks {
227 locks: Box<[atomic::AtomicU32]>,
228}
229
230impl SlotLocks {
231 fn new(total_slots: usize) -> Self {
232 let mut slots = Vec::with_capacity(total_slots);
233 for _ in 0..total_slots {
234 slots.push(atomic::AtomicU32::new(0));
235 }
236
237 Self { locks: slots.into_boxed_slice() }
238 }
239
240 fn grow(&mut self, additional_slots: usize) {
241 if additional_slots == 0 {
242 return;
243 }
244
245 let total = self.locks.len() + additional_slots;
246 let mut slots = Vec::with_capacity(total);
247 for val in self.locks.iter() {
248 slots.push(atomic::AtomicU32::new(val.load(atomic::Ordering::Relaxed)));
249 }
250 for _ in 0..additional_slots {
251 slots.push(atomic::AtomicU32::new(0));
252 }
253
254 self.locks = slots.into_boxed_slice();
255 }
256
257 #[inline(always)]
258 fn read_lock(&self, index: usize) -> SlotReadGuard<'_> {
259 let val = &self.locks[index];
260 let mut spins = 0;
261
262 loop {
263 let state = val.load(atomic::Ordering::Relaxed);
264 if hints::likely(state & WRITER_BIT == 0)
265 && val
266 .compare_exchange_weak(
267 state,
268 state + 1,
269 atomic::Ordering::Acquire,
270 atomic::Ordering::Relaxed,
271 )
272 .is_ok()
273 {
274 return SlotReadGuard(val);
275 }
276
277 if hints::likely(spins < L1_CONTENTION) {
278 std::hint::spin_loop();
279 } else if spins < L2_CONTENTION {
280 std::thread::yield_now();
281 } else {
282 let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
283 std::thread::sleep(std::time::Duration::from_nanos(ns));
284 }
285
286 spins += 1;
287 }
288 }
289
290 #[inline(always)]
291 fn write_lock(&self, index: usize) -> SlotWriteGuard<'_> {
292 let val = &self.locks[index];
293 let mut spins = 0;
294
295 loop {
296 if val
297 .compare_exchange_weak(
298 0,
299 WRITER_BIT,
300 atomic::Ordering::Acquire,
301 atomic::Ordering::Relaxed,
302 )
303 .is_ok()
304 {
305 return SlotWriteGuard(val);
306 }
307
308 if hints::likely(spins < L1_CONTENTION) {
309 std::hint::spin_loop();
310 } else if spins < L2_CONTENTION {
311 std::thread::yield_now();
312 } else {
313 let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
314 std::thread::sleep(std::time::Duration::from_nanos(ns));
315 }
316
317 spins += 1;
318 }
319 }
320}
321
322struct SlotReadGuard<'a>(&'a atomic::AtomicU32);
324
325impl Drop for SlotReadGuard<'_> {
326 #[inline(always)]
327 fn drop(&mut self) {
328 self.0.fetch_sub(1, atomic::Ordering::Release);
329 }
330}
331
332struct SlotWriteGuard<'a>(&'a atomic::AtomicU32);
334
335impl Drop for SlotWriteGuard<'_> {
336 #[inline(always)]
337 fn drop(&mut self) {
338 self.0.store(0, atomic::Ordering::Release);
339 }
340}
341
342#[derive(Debug)]
344struct MemMapInner {
345 cfg: MemMapCfg,
346 mmap: RwLock<Option<PlatformMemMap>>,
347 slot_locks: RwLock<SlotLocks>,
348 length: atomic::AtomicUsize,
349 completion: Arc<Completion>,
350 sync_condvar: Condvar,
351 sync_mutex: Mutex<bool>,
352 shutdown: atomic::AtomicBool,
353 is_parked: atomic::AtomicBool,
354}
355
356impl SyncTrigger for MemMapInner {
357 fn trigger_sync(&self) -> FrozenResult<()> {
358 if self.cfg.sync_interval.is_some() {
359 {
360 let mut guard = self.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
361 *guard = true;
362 }
363
364 self.sync_condvar.notify_one();
365 Ok(())
366 } else {
367 self.sync_internal()
368 }
369 }
370}
371
372impl MemMapInner {
373 fn sync_mmap_if_needed(&self, mmap: &PlatformMemMap, sync_len: usize) -> FrozenResult<()> {
375 let target_epoch = self.completion.read_current_epoch();
376 let durable_epoch = self.completion.read_durable_epoch();
377
378 if target_epoch <= durable_epoch {
379 return Ok(());
380 }
381
382 let sync_res = mmap.sync(sync_len);
383 match sync_res {
384 Ok(()) => {
385 self.completion.mark_epoch_as_durable(target_epoch);
386 self.completion.del_err();
387 self.completion.notify_all_listeners();
388
389 Ok(())
390 }
391 Err(e) => {
392 let e = e.with_module(self.cfg.module_id);
393 self.completion.set_err(e.clone());
394 self.completion.notify_all_listeners();
395
396 Err(e)
397 }
398 }
399 }
400
401 fn sync_internal(&self) -> FrozenResult<()> {
402 let guard = self.mmap.read().unwrap_or_else(|e| e.into_inner());
403 let mmap = match guard.as_ref() {
404 Some(m) => m,
405 None => return Ok(()),
406 };
407
408 let length = self.length.load(atomic::Ordering::Acquire);
409 self.sync_mmap_if_needed(mmap, length)
410 }
411}
412
413pub struct MemMap<T: Sized + Clone> {
459 cfg: MemMapCfg,
460 file: Option<File>,
461 inner: Arc<MemMapInner>,
462 bg_thread: Option<std::thread::JoinHandle<()>>,
463 _marker: PhantomData<T>,
464}
465
466impl<T> core::fmt::Debug for MemMap<T>
467where
468 T: Sized + Clone,
469{
470 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
471 f.debug_struct("MemMap").field("cfg", &self.cfg).field("inner", &self.inner).finish()
472 }
473}
474
475unsafe impl<T> Send for MemMap<T> where T: Sized + Clone {}
476unsafe impl<T> Sync for MemMap<T> where T: Sized + Clone {}
477
478impl<T> MemMap<T>
479where
480 T: Sized + Clone,
481{
482 fn validate_t(mid: u8) -> FrozenResult<()> {
483 if core::mem::needs_drop::<T>() {
484 return err::default_error(err::DRP).with_module(mid);
485 }
486
487 let size = core::mem::size_of::<T>();
488 if size == 0 {
489 return err::default_error(err::ZRO).with_module(mid);
490 }
491
492 if size % 8 != 0 {
493 return err::default_error(err::SZE).with_module(mid);
494 }
495
496 let align = core::mem::align_of::<T>();
497 if align != 8 {
498 return err::default_error(err::ALN).with_module(mid);
499 }
500
501 let page_size = PlatformMemMap::page_size().with_module(mid)?;
502 if size > page_size {
503 return err::default_error(err::INV).with_module(mid);
504 }
505
506 Ok(())
507 }
508
509 fn validate_path(cfg: &MemMapCfg) -> FrozenResult<()> {
510 let mid = cfg.module_id;
511
512 if cfg.path.as_os_str().is_empty() {
513 return err::default_error(err::PTH).with_module(mid);
514 }
515
516 let stem = cfg.path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
517 if stem.is_empty() {
518 return err::default_error(err::PTH).with_module(mid);
519 }
520
521 if cfg.path.exists() && cfg.path.is_dir() {
522 return err::default_error(err::PTH).with_module(mid);
523 }
524
525 if let Some(parent) = cfg.path.parent() {
526 if !parent.as_os_str().is_empty() && (!parent.exists() || !parent.is_dir()) {
527 return err::default_error(err::DIR).with_module(mid);
528 }
529 }
530
531 Ok(())
532 }
533
534 fn file_cfg(cfg: &MemMapCfg) -> FileCfg {
535 FileCfg {
536 module_id: cfg.module_id,
537 path: cfg.versioned_path(),
538 buffer_size: core::mem::size_of::<T>(),
539 initial_available_buffers: cfg.initial_t,
540 sync_interval: None,
541 }
542 }
543
544 fn from_components(cfg: MemMapCfg, mmap: PlatformMemMap, file: File) -> FrozenResult<Self> {
545 let length = mmap.length();
546 let completion = Arc::new(Completion::default());
547 let sync_interval = cfg.sync_interval;
548 let total_slots = length / core::mem::size_of::<T>();
549 let slot_locks = SlotLocks::new(total_slots);
550
551 let inner = Arc::new(MemMapInner {
552 cfg: cfg.clone(),
553 mmap: RwLock::new(Some(mmap)),
554 slot_locks: RwLock::new(slot_locks),
555 length: atomic::AtomicUsize::new(length),
556 completion: completion.clone(),
557 sync_condvar: Condvar::new(),
558 sync_mutex: Mutex::new(false),
559 shutdown: atomic::AtomicBool::new(false),
560 is_parked: atomic::AtomicBool::new(false),
561 });
562
563 let weak_inner: std::sync::Weak<dyn SyncTrigger> =
564 Arc::downgrade(&inner) as std::sync::Weak<dyn SyncTrigger>;
565 completion.set_sync_trigger(weak_inner);
566
567 let bg_thread = if let Some(interval) = sync_interval {
568 let worker_inner = Arc::clone(&inner);
569 let handle = std::thread::Builder::new().name("frozen-mmap-sync-worker".into()).spawn(
570 move || {
571 let mut guard =
572 worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
573
574 while !worker_inner.shutdown.load(atomic::Ordering::Acquire) {
575 if *guard {
576 *guard = false;
577 drop(guard);
578
579 let _ = worker_inner.sync_internal();
580 guard =
581 worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
582 continue;
583 }
584
585 let target_epoch = worker_inner.completion.read_current_epoch();
586 let durable_epoch = worker_inner.completion.read_durable_epoch();
587
588 if target_epoch == durable_epoch {
589 worker_inner.is_parked.store(true, atomic::Ordering::Release);
590
591 if worker_inner.completion.read_current_epoch() != durable_epoch {
592 worker_inner.is_parked.store(false, atomic::Ordering::Release);
593 continue;
594 }
595
596 let new_guard = worker_inner
597 .sync_condvar
598 .wait(guard)
599 .unwrap_or_else(|e| e.into_inner());
600 guard = new_guard;
601 worker_inner.is_parked.store(false, atomic::Ordering::Release);
602
603 if worker_inner.shutdown.load(atomic::Ordering::Acquire) || *guard {
604 continue;
605 }
606
607 if worker_inner.completion.read_current_epoch()
608 == worker_inner.completion.read_durable_epoch()
609 {
610 continue;
611 }
612
613 let (new_guard, _) = worker_inner
614 .sync_condvar
615 .wait_timeout(guard, interval)
616 .unwrap_or_else(|e| e.into_inner());
617 guard = new_guard;
618 } else {
619 worker_inner.is_parked.store(false, atomic::Ordering::Release);
620
621 let (new_guard, _) = worker_inner
622 .sync_condvar
623 .wait_timeout(guard, interval)
624 .unwrap_or_else(|e| e.into_inner());
625 guard = new_guard;
626 }
627
628 if worker_inner.shutdown.load(atomic::Ordering::Acquire) {
629 break;
630 }
631
632 *guard = false;
633 drop(guard);
634
635 let _ = worker_inner.sync_internal();
636 guard = worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
637 }
638 },
639 );
640
641 let handle = match handle {
642 Ok(h) => h,
643 Err(spawn_err) => {
644 let mut mmap_guard = inner.mmap.write().unwrap_or_else(|e| e.into_inner());
645 let mut err = err::make_raw_error(err::FXE, spawn_err);
646 if let Some(m) = mmap_guard.take() {
647 if let Err(unmap_err) = m.unmap() {
648 err.add_suppressed(unmap_err);
649 }
650 }
651 return Err(err.with_module(cfg.module_id));
652 }
653 };
654
655 Some(handle)
656 } else {
657 None
658 };
659
660 Ok(Self { cfg, file: Some(file), inner, bg_thread, _marker: PhantomData })
661 }
662
663 pub fn new(cfg: MemMapCfg) -> FrozenResult<Self> {
701 Self::validate_t(cfg.module_id)?;
702 Self::validate_path(&cfg)?;
703
704 let vpath = cfg.versioned_path();
705 if vpath.exists() {
706 return err::default_error(err::EXS).with_module(cfg.module_id);
707 }
708
709 let file_cfg = Self::file_cfg(&cfg);
710 let file = match File::new(file_cfg) {
711 Ok(f) => f,
712 Err(e) if e.reason == err::EXS.reason => {
713 return err::default_error(err::EXS).with_module(cfg.module_id);
714 }
715 Err(e) => return Err(e),
716 };
717
718 let mmap = match PlatformMemMap::new(file.fd(), file.length()) {
719 Ok(m) => m,
720 Err(mut e) => {
721 if let Err(del_err) = file.delete() {
722 e.add_suppressed(del_err);
723 }
724
725 return Err(e.with_module(cfg.module_id));
726 }
727 };
728
729 Self::from_components(cfg, mmap, file)
730 }
731
732 pub fn open(cfg: MemMapCfg) -> FrozenResult<Self> {
769 Self::validate_t(cfg.module_id)?;
770 Self::validate_path(&cfg)?;
771
772 let vpath = cfg.versioned_path();
773 if !vpath.exists() {
774 return err::default_error(err::NEX).with_module(cfg.module_id);
775 }
776 if vpath.is_dir() {
777 return err::default_error(err::PTH).with_module(cfg.module_id);
778 }
779
780 let file_cfg = Self::file_cfg(&cfg);
781 let file = File::open(file_cfg)?;
782
783 let mmap = PlatformMemMap::new(file.fd(), file.length()).with_module(cfg.module_id)?;
784
785 Self::from_components(cfg, mmap, file)
786 }
787
788 pub fn open_or_create(cfg: MemMapCfg) -> FrozenResult<Self> {
825 Self::validate_t(cfg.module_id)?;
826 Self::validate_path(&cfg)?;
827
828 let vpath = cfg.versioned_path();
829 if vpath.exists() && vpath.is_dir() {
830 return err::default_error(err::PTH).with_module(cfg.module_id);
831 }
832
833 let file_cfg = Self::file_cfg(&cfg);
834 let file = File::open_or_create(file_cfg)?;
835
836 let mmap = PlatformMemMap::new(file.fd(), file.length()).with_module(cfg.module_id)?;
837
838 Self::from_components(cfg, mmap, file)
839 }
840
841 #[inline]
853 pub fn page_size() -> FrozenResult<usize> {
854 PlatformMemMap::page_size()
855 }
856
857 #[inline]
880 pub fn length(&self) -> usize {
881 self.inner.length.load(atomic::Ordering::Acquire)
882 }
883
884 pub fn grow(&self, count: usize) -> FrozenResult<()> {
933 let mid = self.cfg.module_id;
934 if count == 0 {
935 return Ok(());
936 }
937
938 let size = core::mem::size_of::<T>();
939 let len_to_add = match count.checked_mul(size) {
940 Some(len) => len,
941 None => {
942 return err::raw_error(err::GRW, "file growth calculation overflowed usize")
943 .with_module(mid);
944 }
945 };
946
947 let file = match self.file.as_ref() {
948 Some(f) => f,
949 None => return err::default_error(err::INV).with_module(mid),
950 };
951
952 let mut slot_locks_guard = self.inner.slot_locks.write().unwrap_or_else(|e| e.into_inner());
954
955 let mut mmap_guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
957
958 let mmap = match mmap_guard.as_ref() {
959 Some(m) => m,
960 None => return err::default_error(err::INV).with_module(mid),
961 };
962
963 let curr_len = self.inner.length.load(atomic::Ordering::Acquire);
964 let new_length = match curr_len.checked_add(len_to_add) {
965 Some(len) => len,
966 None => {
967 return err::raw_error(err::GRW, "file growth calculation overflowed usize")
968 .with_module(mid);
969 }
970 };
971
972 self.inner.sync_mmap_if_needed(mmap, curr_len)?;
974
975 let old_mmap = match mmap_guard.take() {
976 Some(m) => m,
977 None => return err::default_error(err::INV).with_module(mid),
978 };
979
980 if let Err(e) = old_mmap.unmap().with_module(mid) {
981 if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
982 *mmap_guard = Some(restored);
983 }
984 return Err(e);
985 }
986
987 if let Err(e) = file.grow(count).with_module(mid) {
988 if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
989 *mmap_guard = Some(restored);
990 }
991 return Err(e);
992 }
993
994 if let Err(e) = file.sync().with_module(mid) {
996 if let Ok(restored) = PlatformMemMap::new(file.fd(), new_length).with_module(mid) {
997 *mmap_guard = Some(restored);
998 self.inner.length.store(new_length, atomic::Ordering::Release);
999 slot_locks_guard.grow(count);
1000 }
1001 return Err(e);
1002 }
1003
1004 let new_mmap = match PlatformMemMap::new(file.fd(), new_length).with_module(mid) {
1005 Ok(m) => m,
1006 Err(e) => {
1007 if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
1008 *mmap_guard = Some(restored);
1009 }
1010 return Err(e);
1011 }
1012 };
1013
1014 *mmap_guard = Some(new_mmap);
1015 slot_locks_guard.grow(count);
1016 self.inner.length.store(new_length, atomic::Ordering::Release);
1017
1018 self.inner.completion.increment_current_epoch();
1019
1020 Ok(())
1021 }
1022
1023 #[inline]
1045 pub fn cfg(&self) -> &MemMapCfg {
1046 &self.cfg
1047 }
1048
1049 #[inline]
1074 pub fn current_epoch(&self) -> TEpoch {
1075 self.inner.completion.read_current_epoch()
1076 }
1077
1078 #[inline]
1101 pub fn durable_epoch(&self) -> TEpoch {
1102 self.inner.completion.read_durable_epoch()
1103 }
1104
1105 #[inline]
1136 pub fn sync(&self) -> FrozenResult<()> {
1137 self.inner.sync_internal()
1138 }
1139
1140 #[inline(always)]
1167 pub fn read(&self, index: usize) -> FrozenResult<T> {
1168 let mid = self.cfg.module_id;
1169 let size = core::mem::size_of::<T>();
1170
1171 let offset = match index.checked_mul(size) {
1172 Some(off) => off,
1173 None => return err::default_error(err::INV).with_module(mid),
1174 };
1175
1176 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1177
1178 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1179 return err::default_error(err::HCF).with_module(mid);
1180 }
1181
1182 let _slot_guard = slot_locks_guard.read_lock(index);
1183
1184 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1185 let mmap = match guard.as_ref() {
1186 Some(m) => m,
1187 None => return err::default_error(err::INV).with_module(mid),
1188 };
1189
1190 let ptr = unsafe { mmap.as_ptr::<T>(offset) };
1191 let val = unsafe { ptr.read() };
1192 drop(guard);
1193 drop(_slot_guard);
1194 drop(slot_locks_guard);
1195 Ok(val)
1196 }
1197
1198 #[inline(always)]
1227 pub fn write(&self, val: &T, index: usize) -> FrozenResult<AckTicket> {
1228 let mid = self.cfg.module_id;
1229 let size = core::mem::size_of::<T>();
1230
1231 let offset = match index.checked_mul(size) {
1232 Some(off) => off,
1233 None => return err::default_error(err::INV).with_module(mid),
1234 };
1235
1236 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1237
1238 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1239 return err::default_error(err::HCF).with_module(mid);
1240 }
1241
1242 let _slot_guard = slot_locks_guard.write_lock(index);
1243
1244 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1245 let mmap = match guard.as_ref() {
1246 Some(m) => m,
1247 None => return err::default_error(err::INV).with_module(mid),
1248 };
1249
1250 let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1251 unsafe { core::ptr::write(ptr, val.clone()) };
1252 drop(guard);
1253 drop(_slot_guard);
1254 drop(slot_locks_guard);
1255
1256 let epoch = self.inner.completion.increment_current_epoch();
1257
1258 if self.inner.is_parked.load(atomic::Ordering::Acquire) {
1259 let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1260 if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1261 self.inner.sync_condvar.notify_one();
1262 }
1263 }
1264
1265 Ok(AckTicket::new(epoch, self.inner.completion.clone()))
1266 }
1267
1268 #[cfg(test)]
1269 pub(crate) fn is_parked(&self) -> bool {
1270 self.inner.is_parked.load(atomic::Ordering::Acquire)
1271 }
1272
1273 pub fn delete(mut self) -> FrozenResult<()> {
1310 let mid = self.cfg.module_id;
1311 self.inner.shutdown.store(true, atomic::Ordering::Release);
1312 {
1313 let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1314 *guard = true;
1315 }
1316 self.inner.sync_condvar.notify_all();
1317
1318 if let Some(handle) = self.bg_thread.take() {
1319 let _ = handle.join();
1320 }
1321
1322 let mut guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
1323 let unmap_res =
1324 if let Some(mmap) = guard.take() { mmap.unmap().with_module(mid) } else { Ok(()) };
1325
1326 let file = match self.file.take() {
1327 Some(f) => f,
1328 None => {
1329 if let Err(e) = unmap_res {
1330 return Err(e);
1331 }
1332 return err::default_error(err::INV).with_module(mid);
1333 }
1334 };
1335
1336 let delete_res = file.delete().with_module(mid);
1337
1338 match (unmap_res, delete_res) {
1339 (Err(mut unmap_err), Err(del_err)) => {
1340 unmap_err.add_suppressed(del_err);
1341 Err(unmap_err)
1342 }
1343 (Err(unmap_err), Ok(())) => Err(unmap_err),
1344 (Ok(()), Err(del_err)) => Err(del_err),
1345 (Ok(()), Ok(())) => Ok(()),
1346 }
1347 }
1348}
1349
1350impl<T> MemMap<T>
1351where
1352 T: Sized + Clone + PartialEq,
1353{
1354 pub fn compare_and_write(
1398 &self,
1399 expected: &T,
1400 val: &T,
1401 index: usize,
1402 ) -> FrozenResult<AckTicket> {
1403 let mid = self.cfg.module_id;
1404 let size = core::mem::size_of::<T>();
1405
1406 let offset = match index.checked_mul(size) {
1407 Some(off) => off,
1408 None => return err::default_error(err::INV).with_module(mid),
1409 };
1410
1411 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1412
1413 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1414 return err::default_error(err::HCF).with_module(mid);
1415 }
1416
1417 let _slot_guard = slot_locks_guard.write_lock(index);
1418
1419 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1420 let mmap = match guard.as_ref() {
1421 Some(m) => m,
1422 None => return err::default_error(err::INV).with_module(mid),
1423 };
1424
1425 let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1426 let current = unsafe { ptr.read() };
1427 if current != *expected {
1428 drop(guard);
1429 drop(_slot_guard);
1430 drop(slot_locks_guard);
1431 return err::raw_error(
1432 err::CAS,
1433 format!("state conflict at index {}: slot data has changed", index),
1434 )
1435 .with_module(mid);
1436 }
1437
1438 unsafe { core::ptr::write(ptr, val.clone()) };
1439 drop(guard);
1440 drop(_slot_guard);
1441 drop(slot_locks_guard);
1442
1443 let epoch = self.inner.completion.increment_current_epoch();
1444
1445 if self.inner.is_parked.load(atomic::Ordering::Acquire) {
1446 let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1447 if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1448 self.inner.sync_condvar.notify_one();
1449 }
1450 }
1451
1452 Ok(AckTicket::new(epoch, self.inner.completion.clone()))
1453 }
1454
1455 #[inline]
1495 pub fn new_tx(&self) -> MemMapTransaction<'_, T> {
1496 MemMapTransaction { memmap: self, ops: Vec::new() }
1497 }
1498}
1499
1500#[derive(Debug)]
1535pub struct MemMapTransaction<'a, T: Sized + Clone + PartialEq> {
1536 memmap: &'a MemMap<T>,
1537 ops: Vec<(usize, T, T)>,
1538}
1539
1540impl<'a, T> MemMapTransaction<'a, T>
1541where
1542 T: Sized + Clone + PartialEq,
1543{
1544 #[inline(always)]
1578 pub fn write(&mut self, expected: &T, new: &T, index: usize) -> FrozenResult<&mut Self> {
1579 let mid = self.memmap.cfg.module_id;
1580 if let Some(err) = self.memmap.inner.completion.get_err() {
1581 return Err(err);
1582 }
1583
1584 if let Some((last_idx, _, _)) = self.ops.last() {
1585 if index <= *last_idx {
1586 return err::raw_error(
1587 err::HCF,
1588 "tx writes must be strictly ordered, with no more then single ops on given index",
1589 )
1590 .with_module(mid);
1591 }
1592 }
1593
1594 let size = core::mem::size_of::<T>();
1595 let offset = match index.checked_mul(size) {
1596 Some(off) => off,
1597 None => return err::default_error(err::INV).with_module(mid),
1598 };
1599
1600 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.memmap.length())) {
1601 return err::default_error(err::HCF).with_module(mid);
1602 }
1603
1604 self.ops.push((index, expected.clone(), new.clone()));
1605 Ok(self)
1606 }
1607
1608 pub fn commit(self) -> FrozenResult<AckTicket> {
1642 let mid = self.memmap.cfg.module_id;
1643 if self.ops.is_empty() {
1644 return err::raw_error(err::INV, "empty transaction").with_module(mid);
1645 }
1646
1647 if let Some(err) = self.memmap.inner.completion.get_err() {
1648 return Err(err);
1649 }
1650
1651 let size = core::mem::size_of::<T>();
1652 let slot_locks_guard =
1653 self.memmap.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1654 let current_length = self.memmap.length();
1655
1656 for (index, _, _) in &self.ops {
1658 let offset = match index.checked_mul(size) {
1659 Some(off) => off,
1660 None => return err::default_error(err::INV).with_module(mid),
1661 };
1662 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > current_length)) {
1663 return err::default_error(err::HCF).with_module(mid);
1664 }
1665 }
1666
1667 let mut slot_guards = Vec::with_capacity(self.ops.len());
1669 for (index, _, _) in &self.ops {
1670 slot_guards.push(slot_locks_guard.write_lock(*index));
1671 }
1672
1673 let guard = self.memmap.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1674 let mmap = match guard.as_ref() {
1675 Some(m) => m,
1676 None => return err::default_error(err::INV).with_module(mid),
1677 };
1678
1679 for (index, expected, _) in &self.ops {
1681 let offset = *index * size;
1682 let ptr = unsafe { mmap.as_ptr::<T>(offset) };
1683 let current = unsafe { ptr.read() };
1684 if current != *expected {
1685 drop(guard);
1686 drop(slot_guards);
1687 drop(slot_locks_guard);
1688 return err::raw_error(
1689 err::CAS,
1690 format!("state conflict at index {}: slot data has changed", index),
1691 )
1692 .with_module(mid);
1693 }
1694 }
1695
1696 for (index, _, new) in self.ops {
1698 let offset = index * size;
1699 let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1700 unsafe { core::ptr::write(ptr, new) };
1701 }
1702
1703 drop(guard);
1704 drop(slot_guards);
1705 drop(slot_locks_guard);
1706
1707 let epoch = self.memmap.inner.completion.increment_current_epoch();
1708
1709 if self.memmap.inner.is_parked.load(atomic::Ordering::Acquire) {
1710 let _guard = self.memmap.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1711 if self.memmap.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1712 self.memmap.inner.sync_condvar.notify_one();
1713 }
1714 }
1715
1716 Ok(AckTicket::new(epoch, self.memmap.inner.completion.clone()))
1717 }
1718
1719 #[inline]
1742 pub fn len(&self) -> usize {
1743 self.ops.len()
1744 }
1745
1746 #[inline]
1767 pub fn is_empty(&self) -> bool {
1768 self.ops.is_empty()
1769 }
1770}
1771
1772impl<T> Drop for MemMap<T>
1773where
1774 T: Sized + Clone,
1775{
1776 fn drop(&mut self) {
1777 self.inner.shutdown.store(true, atomic::Ordering::Release);
1778 {
1779 let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1780 *guard = true;
1781 }
1782 self.inner.sync_condvar.notify_all();
1783
1784 if let Some(handle) = self.bg_thread.take() {
1785 let _ = handle.join();
1786 }
1787
1788 let _ = self.inner.sync_internal();
1789 }
1790}
1791
1792#[cfg(test)]
1793mod tests {
1794 use super::*;
1795 use std::time::Duration;
1796
1797 const MID: u8 = 0;
1798 const INIT_T: usize = 4;
1799
1800 #[derive(Debug, Clone)]
1801 #[repr(C, align(8))]
1802 struct ValidT {
1803 a: u64,
1804 b: u64,
1805 }
1806
1807 #[derive(Debug, Clone)]
1808 #[repr(C)]
1809 struct BadAlignT {
1810 a: u32,
1811 b: u32,
1812 }
1813
1814 #[derive(Debug, Clone)]
1815 #[repr(C)]
1816 struct BadSizeT {
1817 a: [u8; 0x0C],
1818 }
1819
1820 #[derive(Debug, Clone)]
1821 #[repr(C, align(8))]
1822 struct DropT(u64);
1823
1824 impl Drop for DropT {
1825 fn drop(&mut self) {}
1826 }
1827
1828 #[derive(Debug, Clone)]
1829 #[repr(C, align(8))]
1830 struct ZstT;
1831
1832 #[derive(Debug, Clone)]
1833 #[repr(C, align(8))]
1834 struct OversizedT {
1835 data: [u8; 0x20000],
1836 }
1837
1838 fn tmp_cfg() -> (tempfile::TempDir, MemMapCfg) {
1839 let dir = tempfile::tempdir().unwrap();
1840 let path = dir.path().join("tmp_memmap");
1841 let cfg = MemMapCfg {
1842 module_id: MID,
1843 path,
1844 version: 1,
1845 initial_t: INIT_T,
1846 sync_interval: Some(Duration::from_millis(0x64)),
1847 };
1848
1849 (dir, cfg)
1850 }
1851
1852 #[test]
1853 fn ok_validate_t() {
1854 assert!(MemMap::<u64>::validate_t(MID).is_ok());
1855 assert!(MemMap::<ValidT>::validate_t(MID).is_ok());
1856 }
1857
1858 #[test]
1859 fn err_validate_t_invariants() {
1860 assert_eq!(MemMap::<DropT>::validate_t(MID).unwrap_err().reason, err::DRP.reason);
1861 assert_eq!(MemMap::<BadAlignT>::validate_t(MID).unwrap_err().reason, err::ALN.reason);
1862 assert_eq!(MemMap::<ZstT>::validate_t(MID).unwrap_err().reason, err::ZRO.reason);
1863 assert_eq!(MemMap::<BadSizeT>::validate_t(MID).unwrap_err().reason, err::SZE.reason);
1864 assert_eq!(MemMap::<OversizedT>::validate_t(MID).unwrap_err().reason, err::INV.reason);
1865 }
1866
1867 #[test]
1868 fn ok_new_sets_file_sync_interval_none() {
1869 let (_dir, cfg) = tmp_cfg();
1870
1871 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
1872 assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
1873 assert_eq!(mmap.cfg().sync_interval, Some(Duration::from_millis(0x64)));
1874 }
1875
1876 #[test]
1877 fn err_new_when_file_exists() {
1878 let (_dir, cfg) = tmp_cfg();
1879 let _mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
1880 let err = MemMap::<u64>::new(cfg).unwrap_err();
1881 assert_eq!(err.reason, err::EXS.reason);
1882 }
1883
1884 #[test]
1885 fn err_new_when_t_invalid() {
1886 let (_dir, cfg) = tmp_cfg();
1887 assert_eq!(MemMap::<DropT>::new(cfg.clone()).unwrap_err().reason, err::DRP.reason);
1888 assert!(!cfg.versioned_path().exists());
1889 }
1890
1891 #[test]
1892 fn ok_open() {
1893 let (_dir, cfg) = tmp_cfg();
1894 {
1895 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
1896 assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
1897 }
1898
1899 let opened = MemMap::<u64>::open(cfg).unwrap();
1900 assert_eq!(opened.length(), core::mem::size_of::<u64>() * INIT_T);
1901 }
1902
1903 #[test]
1904 fn err_open_when_file_missing() {
1905 let (_dir, cfg) = tmp_cfg();
1906 let err = MemMap::<u64>::open(cfg).unwrap_err();
1907 assert_eq!(err.reason, err::NEX.reason);
1908 }
1909
1910 #[test]
1911 fn ok_open_or_create() {
1912 let (_dir, cfg) = tmp_cfg();
1913 {
1914 let mmap = MemMap::<u64>::open_or_create(cfg.clone()).unwrap();
1915 assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
1916 }
1917
1918 let reopened = MemMap::<u64>::open_or_create(cfg).unwrap();
1919 assert_eq!(reopened.length(), core::mem::size_of::<u64>() * INIT_T);
1920 }
1921
1922 #[test]
1923 fn err_path_when_dir_missing() {
1924 let dir = tempfile::tempdir().unwrap();
1925 let non_existent_dir = dir.path().join("missing_dir_xyz");
1926 let cfg = MemMapCfg {
1927 module_id: MID,
1928 path: non_existent_dir.join("file.bin"),
1929 version: 1,
1930 initial_t: INIT_T,
1931 sync_interval: Some(Duration::from_millis(0x64)),
1932 };
1933
1934 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
1935 assert_eq!(err_new.reason, err::DIR.reason);
1936
1937 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
1938 assert_eq!(err_open.reason, err::DIR.reason);
1939
1940 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
1941 assert_eq!(err_open_or_create.reason, err::DIR.reason);
1942 }
1943
1944 #[test]
1945 fn err_path_when_parent_is_not_dir() {
1946 let dir = tempfile::tempdir().unwrap();
1947 let file_path = dir.path().join("regular_file");
1948 std::fs::write(&file_path, b"test").unwrap();
1949
1950 let cfg = MemMapCfg {
1951 module_id: MID,
1952 path: file_path.join("subfile.bin"),
1953 version: 1,
1954 initial_t: INIT_T,
1955 sync_interval: Some(Duration::from_millis(0x64)),
1956 };
1957
1958 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
1959 assert_eq!(err_new.reason, err::DIR.reason);
1960
1961 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
1962 assert_eq!(err_open.reason, err::DIR.reason);
1963
1964 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
1965 assert_eq!(err_open_or_create.reason, err::DIR.reason);
1966 }
1967
1968 #[test]
1969 fn err_path_when_empty() {
1970 let cfg = MemMapCfg {
1971 module_id: MID,
1972 path: std::path::PathBuf::from(""),
1973 version: 1,
1974 initial_t: INIT_T,
1975 sync_interval: Some(Duration::from_millis(0x64)),
1976 };
1977
1978 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
1979 assert_eq!(err_new.reason, err::PTH.reason);
1980
1981 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
1982 assert_eq!(err_open.reason, err::PTH.reason);
1983
1984 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
1985 assert_eq!(err_open_or_create.reason, err::PTH.reason);
1986 }
1987
1988 #[test]
1989 fn err_path_when_is_directory() {
1990 let dir = tempfile::tempdir().unwrap();
1991 let cfg = MemMapCfg {
1992 module_id: MID,
1993 path: dir.path().to_path_buf(),
1994 version: 1,
1995 initial_t: INIT_T,
1996 sync_interval: Some(Duration::from_millis(0x64)),
1997 };
1998
1999 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2000 assert_eq!(err_new.reason, err::PTH.reason);
2001
2002 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2003 assert_eq!(err_open.reason, err::PTH.reason);
2004
2005 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2006 assert_eq!(err_open_or_create.reason, err::PTH.reason);
2007 }
2008
2009 #[test]
2010 fn err_path_when_no_filename() {
2011 let cfg = MemMapCfg {
2012 module_id: MID,
2013 path: std::path::PathBuf::from("/"),
2014 version: 1,
2015 initial_t: INIT_T,
2016 sync_interval: Some(Duration::from_millis(0x64)),
2017 };
2018
2019 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2020 assert_eq!(err_new.reason, err::PTH.reason);
2021
2022 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2023 assert_eq!(err_open.reason, err::PTH.reason);
2024
2025 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2026 assert_eq!(err_open_or_create.reason, err::PTH.reason);
2027 }
2028
2029 #[test]
2030 fn ok_delete() {
2031 let (_dir, cfg) = tmp_cfg();
2032 let path = cfg.versioned_path();
2033 let mmap = MemMap::<u64>::new(cfg).unwrap();
2034 assert!(path.exists());
2035 assert!(mmap.delete().is_ok());
2036 assert!(!path.exists());
2037 }
2038
2039 #[test]
2040 fn ok_versioned_path_resolution() {
2041 let cfg = MemMapCfg {
2042 module_id: MID,
2043 path: std::path::PathBuf::from("data.bin"),
2044 version: 1,
2045 initial_t: 2,
2046 sync_interval: Some(Duration::from_millis(0x64)),
2047 };
2048 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("data_1.bin"));
2049
2050 let cfg = MemMapCfg {
2051 module_id: MID,
2052 path: std::path::PathBuf::from("datastore"),
2053 version: 2,
2054 initial_t: 2,
2055 sync_interval: Some(Duration::from_millis(0x64)),
2056 };
2057 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("datastore_2"));
2058
2059 let cfg = MemMapCfg {
2060 module_id: MID,
2061 path: std::path::PathBuf::from("archive.tar.gz"),
2062 version: 3,
2063 initial_t: 2,
2064 sync_interval: Some(Duration::from_millis(0x64)),
2065 };
2066 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("archive.tar_3.gz"));
2067
2068 let cfg = MemMapCfg {
2069 module_id: MID,
2070 path: std::path::PathBuf::from("/var/data/store.db"),
2071 version: 0x0A,
2072 initial_t: 2,
2073 sync_interval: Some(Duration::from_millis(0x64)),
2074 };
2075 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("/var/data/store_10.db"));
2076
2077 let cfg = MemMapCfg {
2078 module_id: MID,
2079 path: std::path::PathBuf::from("dir/sub/file.dat"),
2080 version: 0,
2081 initial_t: 2,
2082 sync_interval: Some(Duration::from_millis(0x64)),
2083 };
2084 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("dir/sub/file_0.dat"));
2085
2086 let cfg = MemMapCfg {
2087 module_id: MID,
2088 path: std::path::PathBuf::from(".env"),
2089 version: 5,
2090 initial_t: 2,
2091 sync_interval: Some(Duration::from_millis(0x64)),
2092 };
2093 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from(".env_5"));
2094
2095 let cfg = MemMapCfg {
2096 module_id: MID,
2097 path: std::path::PathBuf::from(".config.json"),
2098 version: 1,
2099 initial_t: 2,
2100 sync_interval: Some(Duration::from_millis(0x64)),
2101 };
2102 assert_eq!(cfg.versioned_path(), std::path::PathBuf::from(".config_1.json"));
2103 }
2104
2105 #[test]
2106 fn ok_versioned_path_and_multiversion() {
2107 let dir = tempfile::tempdir().unwrap();
2108 let base_path = dir.path().join("store.dat");
2109
2110 let cfg_v1 = MemMapCfg {
2111 module_id: MID,
2112 path: base_path.clone(),
2113 version: 1,
2114 initial_t: 2,
2115 sync_interval: Some(Duration::from_millis(0x64)),
2116 };
2117 assert_eq!(cfg_v1.versioned_path(), dir.path().join("store_1.dat"));
2118
2119 let cfg_v2 = MemMapCfg {
2120 module_id: MID,
2121 path: base_path.clone(),
2122 version: 2,
2123 initial_t: 4,
2124 sync_interval: Some(Duration::from_millis(0x64)),
2125 };
2126 assert_eq!(cfg_v2.versioned_path(), dir.path().join("store_2.dat"));
2127
2128 let mmap_v1 = MemMap::<u64>::new(cfg_v1.clone()).unwrap();
2130 let mmap_v2 = MemMap::<ValidT>::new(cfg_v2.clone()).unwrap();
2131
2132 assert!(cfg_v1.versioned_path().exists());
2133 assert!(cfg_v2.versioned_path().exists());
2134 assert_eq!(mmap_v1.length(), core::mem::size_of::<u64>() * 2);
2135 assert_eq!(mmap_v2.length(), core::mem::size_of::<ValidT>() * 4);
2136
2137 assert!(mmap_v1.delete().is_ok());
2138 assert!(!cfg_v1.versioned_path().exists());
2139 assert!(cfg_v2.versioned_path().exists());
2140
2141 assert!(mmap_v2.delete().is_ok());
2142 assert!(!cfg_v2.versioned_path().exists());
2143 }
2144
2145 #[test]
2146 fn ok_write_read_single() {
2147 let (_dir, cfg) = tmp_cfg();
2148 let mmap = MemMap::<u64>::new(cfg).unwrap();
2149
2150 assert_eq!(mmap.read(0).unwrap(), 0);
2151 let ticket = mmap.write(&0xDEAD_BEEF_CAFE_BABE, 0).unwrap();
2152 assert_eq!(mmap.read(0).unwrap(), 0xDEAD_BEEF_CAFE_BABE);
2153 assert_eq!(ticket.epoch(), 1);
2154 }
2155
2156 #[test]
2157 fn ok_write_read_multiple_indices() {
2158 let (_dir, cfg) = tmp_cfg();
2159 let mmap = MemMap::<u64>::new(cfg).unwrap();
2160
2161 for i in 0..INIT_T {
2162 let val = (i as u64 + 1) * 0x1111_2222_3333_4444;
2163 mmap.write(&val, i).unwrap();
2164 }
2165
2166 for i in 0..INIT_T {
2167 let expected = (i as u64 + 1) * 0x1111_2222_3333_4444;
2168 assert_eq!(mmap.read(i).unwrap(), expected);
2169 }
2170 }
2171
2172 #[test]
2173 fn ok_write_read_struct() {
2174 let (_dir, cfg) = tmp_cfg();
2175 let mmap = MemMap::<ValidT>::new(cfg).unwrap();
2176
2177 let val0 = ValidT { a: 0x0A, b: 0x14 };
2178 let val1 = ValidT { a: 0x64, b: 0xC8 };
2179
2180 mmap.write(&val0, 0).unwrap();
2181 mmap.write(&val1, 1).unwrap();
2182
2183 let r0 = mmap.read(0).unwrap();
2184 let r1 = mmap.read(1).unwrap();
2185
2186 assert_eq!(r0.a, 0x0A);
2187 assert_eq!(r0.b, 0x14);
2188 assert_eq!(r1.a, 0x64);
2189 assert_eq!(r1.b, 0xC8);
2190 }
2191
2192 #[test]
2193 fn err_write_read_out_of_bounds() {
2194 let (_dir, cfg) = tmp_cfg();
2195 let mmap = MemMap::<u64>::new(cfg).unwrap();
2196
2197 let err_r = mmap.read(INIT_T).unwrap_err();
2198 assert_eq!(err_r.reason, err::HCF.reason);
2199
2200 let err_w = mmap.write(&0x3039, INIT_T).unwrap_err();
2201 assert_eq!(err_w.reason, err::HCF.reason);
2202
2203 let err_r_large = mmap.read(0x64).unwrap_err();
2204 assert_eq!(err_r_large.reason, err::HCF.reason);
2205
2206 let err_w_large = mmap.write(&0x3039, 0x64).unwrap_err();
2207 assert_eq!(err_w_large.reason, err::HCF.reason);
2208 }
2209
2210 #[test]
2211 fn err_write_read_overflow_index() {
2212 let (_dir, cfg) = tmp_cfg();
2213 let mmap = MemMap::<u64>::new(cfg).unwrap();
2214
2215 let err_r = mmap.read(usize::MAX).unwrap_err();
2216 assert_eq!(err_r.reason, err::INV.reason);
2217
2218 let err_w = mmap.write(&0x3039, usize::MAX).unwrap_err();
2219 assert_eq!(err_w.reason, err::INV.reason);
2220 }
2221
2222 #[test]
2223 fn ok_durability_across_reopen() {
2224 let (_dir, cfg) = tmp_cfg();
2225 {
2226 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2227 mmap.write(&0xCAFE_BABE_1234_5678, 1).unwrap();
2228 mmap.inner.sync_internal().unwrap();
2229 }
2230
2231 let reopened = MemMap::<u64>::open(cfg).unwrap();
2232 assert_eq!(reopened.read(1).unwrap(), 0xCAFE_BABE_1234_5678);
2233 }
2234
2235 #[test]
2236 fn ok_background_sync_advances_epoch() {
2237 let (_dir, mut cfg) = tmp_cfg();
2238 cfg.sync_interval = Some(Duration::from_millis(0x19));
2239
2240 let mmap = MemMap::<u64>::new(cfg).unwrap();
2241 let ticket = mmap.write(&0x1234, 0).unwrap();
2242 let epoch = ticket.wait().unwrap();
2243
2244 assert_eq!(epoch, ticket.epoch());
2245 assert!(ticket.is_durable());
2246 assert!(mmap.durable_epoch() >= ticket.epoch());
2247 }
2248
2249 #[test]
2250 fn ok_ticket_force_wakes_bg_worker() {
2251 let (_dir, mut cfg) = tmp_cfg();
2252 cfg.sync_interval = Some(Duration::from_secs(0x0A));
2254
2255 let mmap = MemMap::<u64>::new(cfg).unwrap();
2256 let ticket = mmap.write(&0xABCD, 0).unwrap();
2257 assert!(!ticket.is_durable());
2258
2259 let start = std::time::Instant::now();
2260 let epoch = ticket.force().unwrap();
2261 let elapsed = start.elapsed();
2262
2263 assert_eq!(epoch, ticket.epoch());
2264 assert!(ticket.is_durable());
2265 assert!(elapsed < Duration::from_secs(2));
2266 }
2267
2268 #[test]
2269 fn ok_drop_persists_unflushed_writes() {
2270 let (_dir, mut cfg) = tmp_cfg();
2271 cfg.sync_interval = Some(Duration::from_secs(0x1E));
2272
2273 {
2274 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2275 let _ticket = mmap.write(&0xBEEF_CAFE, 0).unwrap();
2276 drop(mmap);
2277 }
2278
2279 let reopened = MemMap::<u64>::open(cfg).unwrap();
2280 assert_eq!(reopened.read(0).unwrap(), 0xBEEF_CAFE);
2281 }
2282
2283 #[test]
2284 fn ok_concurrent_slot_locking() {
2285 let (_dir, mut cfg) = tmp_cfg();
2286 cfg.initial_t = 8;
2287 cfg.sync_interval = Some(Duration::from_secs(0x3C));
2288
2289 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2290 let barrier = Arc::new(std::sync::Barrier::new(8));
2291 let mut handles = Vec::new();
2292
2293 for t in 0..4 {
2295 let mmap_clone = Arc::clone(&mmap);
2296 let barrier_clone = Arc::clone(&barrier);
2297 handles.push(std::thread::spawn(move || {
2298 barrier_clone.wait();
2299 for i in 0..0x1F4 {
2300 let val = ((t + 1) as u64) << 0x20 | (i as u64);
2301 let _ = mmap_clone.write(&val, 0).unwrap();
2302 }
2303 }));
2304 }
2305
2306 for _ in 0..4 {
2308 let mmap_clone = Arc::clone(&mmap);
2309 let barrier_clone = Arc::clone(&barrier);
2310 handles.push(std::thread::spawn(move || {
2311 barrier_clone.wait();
2312 for _ in 0..0x3E8 {
2313 let val = mmap_clone.read(0).unwrap();
2314 let thread_id = (val >> 0x20) as usize;
2315 assert!(thread_id <= 4, "read value must be valid without torn reads");
2316 }
2317 }));
2318 }
2319
2320 for h in handles {
2321 h.join().unwrap();
2322 }
2323
2324 let barrier2 = Arc::new(std::sync::Barrier::new(4));
2326 let mut read_handles = Vec::new();
2327 for _ in 0..4 {
2328 let mmap_clone = Arc::clone(&mmap);
2329 let barrier_clone = Arc::clone(&barrier2);
2330 read_handles.push(std::thread::spawn(move || {
2331 barrier_clone.wait();
2332 for _ in 0..0x1F4 {
2333 let _ = mmap_clone.read(0).unwrap();
2334 }
2335 }));
2336 }
2337
2338 for h in read_handles {
2339 h.join().unwrap();
2340 }
2341 }
2342
2343 #[test]
2344 fn err_concurrent_new_same_path() {
2345 let (_dir, cfg) = tmp_cfg();
2346 let barrier = Arc::new(std::sync::Barrier::new(2));
2347
2348 let t1 = {
2349 let cfg = cfg.clone();
2350 let b = Arc::clone(&barrier);
2351 std::thread::spawn(move || {
2352 b.wait();
2353 MemMap::<u64>::new(cfg)
2354 })
2355 };
2356
2357 let t2 = {
2358 let b = Arc::clone(&barrier);
2359 std::thread::spawn(move || {
2360 b.wait();
2361 MemMap::<u64>::new(cfg)
2362 })
2363 };
2364
2365 let r1 = t1.join().unwrap();
2366 let r2 = t2.join().unwrap();
2367
2368 match (r1, r2) {
2369 (Ok(m), Err(e)) | (Err(e), Ok(m)) => {
2370 assert_eq!(e.reason, err::EXS.reason);
2371 assert!(m.delete().is_ok());
2372 }
2373 (Ok(_), Ok(_)) => {
2374 panic!("Concurrent MemMap::new on identical path must not both succeed");
2375 }
2376 (Err(e1), Err(e2)) => {
2377 panic!("Unexpected error on both threads: {:?}, {:?}", e1, e2);
2378 }
2379 }
2380 }
2381
2382 #[test]
2383 fn err_concurrent_open_same_path() {
2384 let (_dir, cfg) = tmp_cfg();
2385 {
2386 let m = MemMap::<u64>::new(cfg.clone()).unwrap();
2387 assert_eq!(m.length(), core::mem::size_of::<u64>() * INIT_T);
2388 }
2389
2390 let barrier = Arc::new(std::sync::Barrier::new(2));
2391
2392 let t1 = {
2393 let cfg = cfg.clone();
2394 let b = Arc::clone(&barrier);
2395 std::thread::spawn(move || {
2396 b.wait();
2397 MemMap::<u64>::open(cfg)
2398 })
2399 };
2400
2401 let t2 = {
2402 let b = Arc::clone(&barrier);
2403 std::thread::spawn(move || {
2404 b.wait();
2405 MemMap::<u64>::open(cfg)
2406 })
2407 };
2408
2409 let r1 = t1.join().unwrap();
2410 let r2 = t2.join().unwrap();
2411
2412 match (r1, r2) {
2413 (Ok(m), Err(e)) | (Err(e), Ok(m)) => {
2414 assert_eq!(e.domain, 0x08, "Error domain must be File (0x08)");
2415 assert_eq!(
2416 e.reason, 0x1C,
2417 "Concurrent open should yield LCK (0x1C) from kernel flock: {:?}",
2418 e
2419 );
2420 assert!(m.delete().is_ok());
2421 }
2422 (Ok(_), Ok(_)) => {
2423 panic!("Exclusive advisory lock violation: both concurrent open calls succeeded");
2424 }
2425 (Err(e1), Err(e2)) => {
2426 panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
2427 }
2428 }
2429 }
2430
2431 #[test]
2432 fn ok_reopen_after_drop_releases_lock() {
2433 let (_dir, cfg) = tmp_cfg();
2434 let m1 = MemMap::<u64>::new(cfg.clone()).unwrap();
2435 m1.write(&0x1122_3344, 0).unwrap();
2436 drop(m1);
2437
2438 let m2 = MemMap::<u64>::open(cfg).unwrap();
2439 assert_eq!(m2.read(0).unwrap(), 0x1122_3344);
2440 assert!(m2.delete().is_ok());
2441 }
2442
2443 #[test]
2444 fn ok_delete_and_recreate_cycle() {
2445 let (_dir, cfg) = tmp_cfg();
2446 for i in 0..4 {
2447 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2448 let val = (i as u64 + 1) * 0xAA55;
2449 mmap.write(&val, 0).unwrap();
2450 assert_eq!(mmap.read(0).unwrap(), val);
2451 assert!(mmap.delete().is_ok());
2452 assert!(!cfg.versioned_path().exists());
2453 }
2454 }
2455
2456 #[test]
2457 fn ok_concurrent_distinct_slots_no_contention() {
2458 let (_dir, mut cfg) = tmp_cfg();
2459 const THREADS: usize = 8;
2460 cfg.initial_t = THREADS;
2461 cfg.sync_interval = Some(Duration::from_secs(0x3C));
2462
2463 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2464 let barrier = Arc::new(std::sync::Barrier::new(THREADS));
2465 let mut handles = Vec::with_capacity(THREADS);
2466
2467 for t in 0..THREADS {
2468 let mmap_clone = Arc::clone(&mmap);
2469 let barrier_clone = Arc::clone(&barrier);
2470 handles.push(std::thread::spawn(move || {
2471 barrier_clone.wait();
2472 for i in 0..0x3E8 {
2473 let val = (t as u64) << 0x20 | (i as u64);
2474 let _ = mmap_clone.write(&val, t).unwrap();
2475 let read_val = mmap_clone.read(t).unwrap();
2476 assert_eq!(read_val >> 0x20, t as u64);
2477 }
2478 }));
2479 }
2480
2481 for h in handles {
2482 h.join().unwrap();
2483 }
2484
2485 for t in 0..THREADS {
2486 let val = mmap.read(t).unwrap();
2487 assert_eq!(val >> 0x20, t as u64);
2488 }
2489 }
2490
2491 #[test]
2492 fn ok_stress_interleaved_reads_writes() {
2493 let (_dir, mut cfg) = tmp_cfg();
2494 const NUM_SLOTS: usize = 0x10;
2495 const NUM_THREADS: usize = 8;
2496 cfg.initial_t = NUM_SLOTS;
2497 cfg.sync_interval = Some(Duration::from_millis(0x32));
2498
2499 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2500 let barrier = Arc::new(std::sync::Barrier::new(NUM_THREADS));
2501 let mut handles = Vec::with_capacity(NUM_THREADS);
2502
2503 for t in 0..NUM_THREADS {
2504 let mmap_clone = Arc::clone(&mmap);
2505 let barrier_clone = Arc::clone(&barrier);
2506 handles.push(std::thread::spawn(move || {
2507 barrier_clone.wait();
2508 for i in 0..0x5DC {
2509 let slot = (t + i) % NUM_SLOTS;
2510 if i % 2 == 0 {
2511 let val = ((t + 1) as u64) << 0x20 | (i as u64);
2512 let _ = mmap_clone.write(&val, slot).unwrap();
2513 } else {
2514 let _ = mmap_clone.read(slot).unwrap();
2515 }
2516 }
2517 }));
2518 }
2519
2520 for h in handles {
2521 h.join().unwrap();
2522 }
2523
2524 assert!(mmap.inner.sync_internal().is_ok());
2525 }
2526
2527 #[test]
2528 fn ok_concurrent_writers_with_background_sync() {
2529 let (_dir, mut cfg) = tmp_cfg();
2530 cfg.initial_t = 8;
2531 cfg.sync_interval = Some(Duration::from_millis(5));
2532
2533 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2534 let barrier = Arc::new(std::sync::Barrier::new(4));
2535 let mut handles = Vec::new();
2536
2537 for t in 0..4 {
2538 let mmap_clone = Arc::clone(&mmap);
2539 let barrier_clone = Arc::clone(&barrier);
2540 handles.push(std::thread::spawn(move || {
2541 barrier_clone.wait();
2542 let mut tickets = Vec::with_capacity(0x32);
2543 for i in 0..0x32 {
2544 let val = (t as u64) * 0x3E8 + i as u64;
2545 tickets.push(mmap_clone.write(&val, t).unwrap());
2546 std::thread::sleep(Duration::from_millis(1));
2547 }
2548 for ticket in tickets {
2549 let _ = ticket.wait().unwrap();
2550 assert!(ticket.is_durable());
2551 }
2552 }));
2553 }
2554
2555 for h in handles {
2556 h.join().unwrap();
2557 }
2558
2559 assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
2560 }
2561
2562 #[test]
2563 fn ok_concurrent_ticket_force_stress() {
2564 let (_dir, mut cfg) = tmp_cfg();
2565 cfg.initial_t = 8;
2566 cfg.sync_interval = Some(Duration::from_secs(0x1E));
2567
2568 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2569 let barrier = Arc::new(std::sync::Barrier::new(4));
2570 let mut handles = Vec::new();
2571
2572 for t in 0..4 {
2573 let mmap_clone = Arc::clone(&mmap);
2574 let barrier_clone = Arc::clone(&barrier);
2575 handles.push(std::thread::spawn(move || {
2576 barrier_clone.wait();
2577 for i in 0..0x14 {
2578 let val = (t as u64) * 0x64 + i as u64;
2579 let ticket = mmap_clone.write(&val, t).unwrap();
2580 let durable_epoch = ticket.force().unwrap();
2581 assert!(ticket.is_durable());
2582 assert!(durable_epoch >= ticket.epoch());
2583 }
2584 }));
2585 }
2586
2587 for h in handles {
2588 h.join().unwrap();
2589 }
2590 }
2591
2592 #[test]
2593 fn ok_exact_boundary_indices() {
2594 let (_dir, cfg) = tmp_cfg();
2595 let mmap = MemMap::<u64>::new(cfg).unwrap();
2596
2597 assert!(mmap.write(&0xF00D, INIT_T - 1).is_ok());
2598 assert_eq!(mmap.read(INIT_T - 1).unwrap(), 0xF00D);
2599
2600 assert_eq!(mmap.write(&0xBAD, INIT_T).unwrap_err().reason, err::HCF.reason);
2601 assert_eq!(mmap.read(INIT_T).unwrap_err().reason, err::HCF.reason);
2602 }
2603
2604 #[test]
2605 fn ok_large_slot_count_initialization() {
2606 let (_dir, mut cfg) = tmp_cfg();
2607 const LARGE_COUNT: usize = 0x2710;
2608 cfg.initial_t = LARGE_COUNT;
2609
2610 let mmap = MemMap::<u64>::new(cfg).unwrap();
2611 assert_eq!(mmap.length(), LARGE_COUNT * core::mem::size_of::<u64>());
2612
2613 assert!(mmap.write(&1, 0).is_ok());
2614 assert!(mmap.write(&0x1388, 0x1388).is_ok());
2615 assert!(mmap.write(&0x270F, LARGE_COUNT - 1).is_ok());
2616
2617 assert_eq!(mmap.read(0).unwrap(), 1);
2618 assert_eq!(mmap.read(0x1388).unwrap(), 0x1388);
2619 assert_eq!(mmap.read(LARGE_COUNT - 1).unwrap(), 0x270F);
2620 }
2621
2622 #[test]
2623 fn ok_new_with_sync_interval_none() {
2624 let (_dir, mut cfg) = tmp_cfg();
2625 cfg.sync_interval = None;
2626
2627 let mmap = MemMap::<u64>::new(cfg).unwrap();
2628 assert!(mmap.bg_thread.is_none());
2629
2630 let ticket = mmap.write(&0x55AA, 0).unwrap();
2631 assert!(!ticket.is_durable());
2632
2633 mmap.sync().unwrap();
2634 assert!(ticket.is_durable());
2635 assert_eq!(mmap.read(0).unwrap(), 0x55AA);
2636 }
2637
2638 #[test]
2639 fn ok_ticket_force_with_sync_interval_none() {
2640 let (_dir, mut cfg) = tmp_cfg();
2641 cfg.sync_interval = None;
2642
2643 let mmap = MemMap::<u64>::new(cfg).unwrap();
2644 assert!(mmap.bg_thread.is_none());
2645
2646 let ticket = mmap.write(&0x1122, 1).unwrap();
2647 assert!(!ticket.is_durable());
2648
2649 let epoch = ticket.force().unwrap();
2650 assert_eq!(epoch, ticket.epoch());
2651 assert!(ticket.is_durable());
2652 assert!(mmap.durable_epoch() >= ticket.epoch());
2653 }
2654
2655 #[test]
2656 fn ok_drop_with_sync_interval_none() {
2657 let (_dir, mut cfg) = tmp_cfg();
2658 cfg.sync_interval = None;
2659
2660 {
2661 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2662 assert!(mmap.bg_thread.is_none());
2663 let _ticket = mmap.write(&0xCAFE_F00D, 0).unwrap();
2664 drop(mmap);
2665 }
2666
2667 let reopened = MemMap::<u64>::open(cfg).unwrap();
2668 assert_eq!(reopened.read(0).unwrap(), 0xCAFE_F00D);
2669 }
2670
2671 #[test]
2672 fn ok_bg_worker_parks_when_idle_and_unparks_on_write() {
2673 let (_dir, mut cfg) = tmp_cfg();
2674 cfg.sync_interval = Some(Duration::from_millis(0x0F));
2675
2676 let mmap = MemMap::<u64>::new(cfg).unwrap();
2677
2678 let start = std::time::Instant::now();
2680 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2681 std::thread::sleep(Duration::from_millis(5));
2682 }
2683 assert!(mmap.is_parked());
2684
2685 let ticket = mmap.write(&0x9999, 0).unwrap();
2687 assert!(!mmap.is_parked());
2688
2689 let epoch = ticket.wait().unwrap();
2691 assert_eq!(epoch, ticket.epoch());
2692 assert!(ticket.is_durable());
2693 assert_eq!(mmap.read(0).unwrap(), 0x9999);
2694
2695 let start = std::time::Instant::now();
2697 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2698 std::thread::sleep(Duration::from_millis(5));
2699 }
2700 assert!(mmap.is_parked());
2701 }
2702
2703 #[test]
2704 fn ok_batched_writes_while_active_remain_unparked() {
2705 let (_dir, mut cfg) = tmp_cfg();
2706 cfg.sync_interval = Some(Duration::from_millis(0x1E));
2707
2708 let mmap = MemMap::<u64>::new(cfg).unwrap();
2709
2710 let start = std::time::Instant::now();
2712 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2713 std::thread::sleep(Duration::from_millis(5));
2714 }
2715 assert!(mmap.is_parked());
2716
2717 let t1 = mmap.write(&0x1111, 0).unwrap();
2719 assert!(!mmap.is_parked());
2720
2721 let t2 = mmap.write(&0x2222, 1).unwrap();
2723 let t3 = mmap.write(&0x3333, 2).unwrap();
2724 assert!(!mmap.is_parked());
2725
2726 t1.wait().unwrap();
2727 t2.wait().unwrap();
2728 t3.wait().unwrap();
2729
2730 assert!(t1.is_durable());
2731 assert!(t2.is_durable());
2732 assert!(t3.is_durable());
2733 assert_eq!(mmap.read(0).unwrap(), 0x1111);
2734 assert_eq!(mmap.read(1).unwrap(), 0x2222);
2735 assert_eq!(mmap.read(2).unwrap(), 0x3333);
2736 }
2737
2738 #[test]
2739 fn ok_concurrent_stress_park_unpark_no_lost_wakeups() {
2740 let (_dir, mut cfg) = tmp_cfg();
2741 cfg.initial_t = 0x10;
2742 cfg.sync_interval = Some(Duration::from_millis(5));
2743
2744 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2745 const THREADS: usize = 4;
2746 const WRITES_PER_THREAD: usize = 0x19;
2747 let mut handles = Vec::with_capacity(THREADS);
2748
2749 for t in 0..THREADS {
2750 let mmap_clone = Arc::clone(&mmap);
2751 handles.push(std::thread::spawn(move || {
2752 let mut tickets = Vec::with_capacity(WRITES_PER_THREAD);
2753 for i in 0..WRITES_PER_THREAD {
2754 let val = (t as u64) << 0x20 | (i as u64);
2755 let slot = t;
2756 let ticket = mmap_clone.write(&val, slot).unwrap();
2757 tickets.push(ticket);
2758
2759 if i % 5 == 0 {
2761 std::thread::sleep(Duration::from_millis(0x0F));
2762 }
2763 }
2764
2765 for ticket in tickets {
2766 ticket.wait().unwrap();
2767 assert!(ticket.is_durable());
2768 }
2769 }));
2770 }
2771
2772 for h in handles {
2773 h.join().unwrap();
2774 }
2775
2776 assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
2777 }
2778
2779 #[test]
2780 fn ok_spurious_wakeup_resumes_park_immediately() {
2781 let (_dir, mut cfg) = tmp_cfg();
2782 cfg.sync_interval = Some(Duration::from_secs(0x0A)); let mmap = MemMap::<u64>::new(cfg).unwrap();
2785
2786 let start = std::time::Instant::now();
2788 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2789 std::thread::sleep(Duration::from_millis(5));
2790 }
2791 assert!(mmap.is_parked());
2792
2793 mmap.inner.sync_condvar.notify_one();
2795 std::thread::sleep(Duration::from_millis(0x0A));
2796
2797 let start = std::time::Instant::now();
2800 while !mmap.is_parked() && start.elapsed() < Duration::from_millis(0x1F4) {
2801 std::thread::sleep(Duration::from_millis(5));
2802 }
2803 assert!(mmap.is_parked());
2804 }
2805
2806 #[test]
2807 fn ok_delete_cleans_up_file_when_mmap_already_unmapped() {
2808 let (_dir, cfg) = tmp_cfg();
2809 let path = cfg.versioned_path();
2810 let mmap = MemMap::<u64>::new(cfg).unwrap();
2811 assert!(path.exists());
2812
2813 {
2815 let mut guard = mmap.inner.mmap.write().unwrap();
2816 let _ = guard.take();
2817 }
2818
2819 assert!(mmap.delete().is_ok());
2821 assert!(!path.exists());
2822 }
2823
2824 mod memmap_grow {
2825 use super::*;
2826
2827 #[test]
2828 fn ok_grow_updates_length() {
2829 let (_dir, cfg) = tmp_cfg();
2830 let mmap = MemMap::<u64>::new(cfg).unwrap();
2831 let initial_len = INIT_T * core::mem::size_of::<u64>();
2832 assert_eq!(mmap.length(), initial_len);
2833
2834 mmap.grow(4).unwrap();
2835 assert_eq!(mmap.length(), initial_len + 4 * core::mem::size_of::<u64>());
2836 }
2837
2838 #[test]
2839 fn ok_grow_zero_count() {
2840 let (_dir, cfg) = tmp_cfg();
2841 let mmap = MemMap::<u64>::new(cfg).unwrap();
2842 let initial_len = mmap.length();
2843
2844 mmap.grow(0).unwrap();
2845 assert_eq!(mmap.length(), initial_len);
2846 }
2847
2848 #[test]
2849 fn ok_grow_extends_with_zero() {
2850 let (_dir, cfg) = tmp_cfg();
2851 let mmap = MemMap::<u64>::new(cfg).unwrap();
2852 mmap.grow(3).unwrap();
2853
2854 for idx in INIT_T..(INIT_T + 3) {
2855 assert_eq!(mmap.read(idx).unwrap(), 0u64);
2856 }
2857 }
2858
2859 #[test]
2860 fn ok_grow_preserves_existing_data() {
2861 let (_dir, cfg) = tmp_cfg();
2862 let mmap = MemMap::<u64>::new(cfg).unwrap();
2863
2864 for i in 0..INIT_T {
2865 mmap.write(&(0xCAFE_0000 + i as u64), i).unwrap();
2866 }
2867
2868 mmap.grow(4).unwrap();
2869
2870 for i in 0..INIT_T {
2871 assert_eq!(mmap.read(i).unwrap(), 0xCAFE_0000 + i as u64);
2872 }
2873
2874 for i in INIT_T..(INIT_T + 4) {
2875 mmap.write(&(0xBEEF_0000 + i as u64), i).unwrap();
2876 assert_eq!(mmap.read(i).unwrap(), 0xBEEF_0000 + i as u64);
2877 }
2878 }
2879
2880 #[test]
2881 fn ok_grow_syncs_dirty_pages_when_curr_epoch_gt_durable() {
2882 let (_dir, mut cfg) = tmp_cfg();
2883 cfg.sync_interval = None; let mmap = MemMap::<u64>::new(cfg).unwrap();
2886 let ticket = mmap.write(&0x1234_5678u64, 1).unwrap();
2887
2888 assert!(!ticket.is_durable());
2889 assert!(mmap.current_epoch() > mmap.durable_epoch());
2890
2891 mmap.grow(2).unwrap();
2893
2894 assert!(ticket.is_durable());
2895 assert!(mmap.durable_epoch() >= ticket.epoch());
2896 }
2897
2898 #[test]
2899 fn ok_grow_when_already_durable() {
2900 let (_dir, mut cfg) = tmp_cfg();
2901 cfg.sync_interval = None;
2902
2903 let mmap = MemMap::<u64>::new(cfg).unwrap();
2904 let ticket = mmap.write(&0x9999u64, 0).unwrap();
2905 mmap.sync().unwrap();
2906 assert!(ticket.is_durable());
2907 assert_eq!(mmap.current_epoch(), mmap.durable_epoch());
2908
2909 mmap.grow(2).unwrap();
2910 assert_eq!(mmap.read(0).unwrap(), 0x9999u64);
2911 }
2912
2913 #[test]
2914 fn err_grow_overflow_usize() {
2915 let (_dir, cfg) = tmp_cfg();
2916 let mmap = MemMap::<u64>::new(cfg).unwrap();
2917
2918 let err = mmap.grow(usize::MAX).unwrap_err();
2919 assert_eq!(err.reason, err::GRW.reason);
2920 }
2921
2922 #[test]
2923 fn ok_grow_multiple_times() {
2924 let (_dir, cfg) = tmp_cfg();
2925 let mmap = MemMap::<u64>::new(cfg).unwrap();
2926
2927 mmap.write(&0x0Au64, 0).unwrap();
2928 mmap.grow(2).unwrap();
2929 mmap.write(&0x14u64, INIT_T).unwrap();
2930 mmap.grow(3).unwrap();
2931 mmap.write(&0x1Eu64, INIT_T + 2).unwrap();
2932
2933 assert_eq!(mmap.read(0).unwrap(), 0x0Au64);
2934 assert_eq!(mmap.read(INIT_T).unwrap(), 0x14u64);
2935 assert_eq!(mmap.read(INIT_T + 2).unwrap(), 0x1Eu64);
2936 assert_eq!(mmap.length(), (INIT_T + 5) * core::mem::size_of::<u64>());
2937 }
2938
2939 #[test]
2940 fn ok_concurrent_grow_and_reads_writes() {
2941 let (_dir, mut cfg) = tmp_cfg();
2942 cfg.sync_interval = Some(Duration::from_millis(0x0A));
2943 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2944
2945 let num_writers = 4;
2946 let mut handles = Vec::new();
2947
2948 for w in 0..num_writers {
2949 let m = Arc::clone(&mmap);
2950 handles.push(std::thread::spawn(move || {
2951 for i in 0..0x32 {
2952 let _ = m.write(&((w * 0x3E8 + i) as u64), w % INIT_T);
2953 let _ = m.read(w % INIT_T);
2954 }
2955 }));
2956 }
2957
2958 let m_grower = Arc::clone(&mmap);
2959 let grower_handle = std::thread::spawn(move || {
2960 for _ in 0..3 {
2961 std::thread::sleep(Duration::from_millis(5));
2962 m_grower.grow(2).unwrap();
2963 }
2964 });
2965
2966 for h in handles {
2967 h.join().unwrap();
2968 }
2969 grower_handle.join().unwrap();
2970
2971 assert_eq!(mmap.length(), (INIT_T + 6) * core::mem::size_of::<u64>());
2972 }
2973 }
2974
2975 mod thread_failure_lifecycle {
2976 use super::*;
2977
2978 #[test]
2979 fn err_new_failure_paths_spawn_no_sync_thread() {
2980 let (_dir, cfg) = tmp_cfg();
2981 std::fs::write(cfg.versioned_path(), b"preexisting").unwrap();
2983
2984 let res = MemMap::<u64>::new(cfg);
2985 assert!(res.is_err());
2986 }
2988
2989 #[test]
2990 fn err_open_failure_paths_spawn_no_sync_thread() {
2991 let (_dir, mut cfg) = tmp_cfg();
2992 cfg.path = std::path::PathBuf::from("/non/existent/path/memmap.bin");
2993
2994 let res = MemMap::<u64>::open(cfg);
2995 assert!(res.is_err());
2996 }
2998
2999 #[test]
3000 fn ok_thread_shutdown_on_panic_unwinding() {
3001 let (_dir, cfg) = tmp_cfg();
3002 let thread_inner = Arc::new(atomic::AtomicBool::new(false));
3003 let thread_inner_clone = Arc::clone(&thread_inner);
3004
3005 let handle = std::thread::spawn(move || {
3007 let mmap = MemMap::<u64>::new(cfg).unwrap();
3008 let _inner = Arc::clone(&mmap.inner);
3009 thread_inner_clone.store(true, atomic::Ordering::Release);
3010 panic!("intentional test panic to trigger Drop unwinding");
3012 });
3013
3014 assert!(handle.join().is_err());
3015 assert!(thread_inner.load(atomic::Ordering::Acquire));
3016 }
3018
3019 #[test]
3020 fn ok_thread_shutdown_in_delete_even_if_already_unmapped() {
3021 let (_dir, cfg) = tmp_cfg();
3022 let mmap = MemMap::<u64>::new(cfg).unwrap();
3023
3024 {
3026 let mut guard = mmap.inner.mmap.write().unwrap();
3027 let _ = guard.take();
3028 }
3029
3030 assert!(mmap.delete().is_ok());
3032 }
3033
3034 #[test]
3035 fn ok_grow_failure_keeps_thread_healthy() {
3036 let (_dir, cfg) = tmp_cfg();
3037 let mmap = MemMap::<u64>::new(cfg).unwrap();
3038
3039 assert!(mmap.grow(usize::MAX).is_err());
3040
3041 let ticket = mmap.write(&0x2Au64, 0).unwrap();
3043 mmap.sync().unwrap();
3044 assert!(ticket.is_durable());
3045 assert_eq!(mmap.read(0).unwrap(), 0x2Au64);
3046 }
3047 }
3048
3049 mod memmap_transactions {
3050 use super::*;
3051
3052 #[test]
3053 fn ok_single_slot_cas_success_and_conflict() {
3054 let (_dir, cfg) = tmp_cfg();
3055 let mmap = MemMap::<u64>::new(cfg).unwrap();
3056
3057 assert_eq!(mmap.read(0).unwrap(), 0);
3058
3059 let t1 = mmap.compare_and_write(&0, &0x64, 0).unwrap();
3060 assert_eq!(mmap.read(0).unwrap(), 0x64);
3061 assert_eq!(mmap.current_epoch(), 1);
3062
3063 let err = mmap.compare_and_write(&0, &0xC8, 0).unwrap_err();
3064 assert_eq!(err.reason, err::CAS.reason);
3065 assert!(err.context.contains("slot data has changed"));
3066 assert_eq!(mmap.read(0).unwrap(), 0x64);
3067 assert_eq!(mmap.current_epoch(), 1); let t2 = mmap.compare_and_write(&0x64, &0xC8, 0).unwrap();
3070 assert_eq!(mmap.read(0).unwrap(), 0xC8);
3071 assert_eq!(mmap.current_epoch(), 2);
3072 assert!(t2.epoch() > t1.epoch());
3073 }
3074
3075 #[test]
3076 fn ok_tx_multiple_writes_single_epoch() {
3077 let (_dir, cfg) = tmp_cfg();
3078 let mmap = MemMap::<u64>::new(cfg).unwrap();
3079
3080 let epoch_before = mmap.current_epoch();
3081
3082 let mut tx = mmap.new_tx();
3083 assert!(tx.is_empty());
3084 tx.write(&0, &0x6F, 0).unwrap();
3085 tx.write(&0, &0xDE, 1).unwrap();
3086 tx.write(&0, &0x14D, 2).unwrap();
3087 assert_eq!(tx.len(), 3);
3088
3089 let ticket = tx.commit().unwrap();
3090 let epoch_after = mmap.current_epoch();
3091
3092 assert_eq!(epoch_after, epoch_before + 1);
3094 assert_eq!(ticket.epoch(), epoch_after);
3095
3096 assert_eq!(mmap.read(0).unwrap(), 0x6F);
3097 assert_eq!(mmap.read(1).unwrap(), 0xDE);
3098 assert_eq!(mmap.read(2).unwrap(), 0x14D);
3099
3100 ticket.wait().unwrap();
3101 assert!(ticket.is_durable());
3102 }
3103
3104 #[test]
3105 fn err_tx_cas_conflict_aborts_all_with_zero_side_effects() {
3106 let (_dir, cfg) = tmp_cfg();
3107 let mmap = MemMap::<u64>::new(cfg).unwrap();
3108
3109 mmap.write(&0x3E7, 1).unwrap();
3111 let epoch_before = mmap.current_epoch();
3112
3113 let mut tx = mmap.new_tx();
3114 tx.write(&0, &0x0A, 0).unwrap(); tx.write(&0, &0x14, 1).unwrap(); tx.write(&0, &0x1E, 2).unwrap(); let err = tx.commit().unwrap_err();
3119 assert_eq!(err.reason, err::CAS.reason);
3120 assert!(err.context.contains("state conflict at index 1"));
3121
3122 assert_eq!(mmap.read(0).unwrap(), 0);
3124 assert_eq!(mmap.read(1).unwrap(), 0x3E7);
3125 assert_eq!(mmap.read(2).unwrap(), 0);
3126
3127 assert_eq!(mmap.current_epoch(), epoch_before);
3129 }
3130
3131 #[test]
3132 fn ok_tx_retry_after_conflict() {
3133 let (_dir, cfg) = tmp_cfg();
3134 let mmap = MemMap::<u64>::new(cfg).unwrap();
3135
3136 mmap.write(&0x2A, 1).unwrap();
3138
3139 let mut tx1 = mmap.new_tx();
3141 tx1.write(&0, &0x64, 0).unwrap();
3142 tx1.write(&0, &0xC8, 1).unwrap();
3143 assert_eq!(tx1.commit().unwrap_err().reason, err::CAS.reason);
3144
3145 let curr_val = mmap.read(1).unwrap();
3147 assert_eq!(curr_val, 0x2A);
3148
3149 let mut tx2 = mmap.new_tx();
3150 tx2.write(&0, &0x64, 0).unwrap();
3151 tx2.write(&0x2A, &0xC8, 1).unwrap();
3152 let ticket = tx2.commit().unwrap();
3153
3154 assert_eq!(mmap.read(0).unwrap(), 0x64);
3155 assert_eq!(mmap.read(1).unwrap(), 0xC8);
3156 ticket.wait().unwrap();
3157 assert!(ticket.is_durable());
3158 }
3159
3160 #[test]
3161 fn err_tx_out_of_order() {
3162 let (_dir, cfg) = tmp_cfg();
3163 let mmap = MemMap::<u64>::new(cfg).unwrap();
3164
3165 let mut tx = mmap.new_tx();
3166 tx.write(&0, &0x0A, 2).unwrap();
3167 let err = tx.write(&0, &0x14, 1).unwrap_err();
3168 assert_eq!(err.reason, err::HCF.reason);
3169 assert!(err.context.contains("strictly ordered"));
3170 }
3171
3172 #[test]
3173 fn err_tx_duplicate_index() {
3174 let (_dir, cfg) = tmp_cfg();
3175 let mmap = MemMap::<u64>::new(cfg).unwrap();
3176
3177 let mut tx = mmap.new_tx();
3178 tx.write(&0, &0x0A, 1).unwrap();
3179 let err = tx.write(&0, &0x14, 1).unwrap_err();
3180 assert_eq!(err.reason, err::HCF.reason);
3181 assert!(err.context.contains("strictly ordered"));
3182 }
3183
3184 #[test]
3185 fn err_tx_out_of_bounds() {
3186 let (_dir, cfg) = tmp_cfg();
3187 let mmap = MemMap::<u64>::new(cfg).unwrap();
3188
3189 let mut tx = mmap.new_tx();
3190 let err = tx.write(&0, &0x0A, INIT_T).unwrap_err();
3191 assert_eq!(err.reason, err::HCF.reason);
3192 }
3193
3194 #[test]
3195 fn err_tx_empty_commit() {
3196 let (_dir, cfg) = tmp_cfg();
3197 let mmap = MemMap::<u64>::new(cfg).unwrap();
3198
3199 let tx = mmap.new_tx();
3200 let err = tx.commit().unwrap_err();
3201 assert_eq!(err.reason, err::INV.reason);
3202 assert!(err.context.contains("empty transaction"));
3203 }
3204
3205 #[test]
3206 fn ok_tx_discard_on_drop() {
3207 let (_dir, cfg) = tmp_cfg();
3208 let mmap = MemMap::<u64>::new(cfg).unwrap();
3209
3210 let epoch_before = mmap.current_epoch();
3211 {
3212 let mut tx = mmap.new_tx();
3213 tx.write(&0, &0x309, 0).unwrap();
3214 tx.write(&0, &0x378, 1).unwrap();
3215 }
3217
3218 assert_eq!(mmap.read(0).unwrap(), 0);
3219 assert_eq!(mmap.read(1).unwrap(), 0);
3220 assert_eq!(mmap.current_epoch(), epoch_before);
3221 }
3222
3223 #[test]
3224 fn ok_tx_concurrent_non_overlapping() {
3225 let (_dir, mut cfg) = tmp_cfg();
3226 cfg.initial_t = 0x10;
3227 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3228
3229 let mut handles = Vec::new();
3230 for thread_id in 0..4u64 {
3231 let m = mmap.clone();
3232 handles.push(std::thread::spawn(move || {
3233 let base = (thread_id as usize) * 4;
3234 let mut tx = m.new_tx();
3235 tx.write(&0, &(thread_id * 0x0A + 1), base).unwrap();
3236 tx.write(&0, &(thread_id * 0x0A + 2), base + 1).unwrap();
3237 tx.write(&0, &(thread_id * 0x0A + 3), base + 2).unwrap();
3238 tx.write(&0, &(thread_id * 0x0A + 4), base + 3).unwrap();
3239 let ticket = tx.commit().unwrap();
3240 ticket.wait().unwrap();
3241 }));
3242 }
3243
3244 for h in handles {
3245 h.join().unwrap();
3246 }
3247
3248 for thread_id in 0..4u64 {
3249 let base = (thread_id as usize) * 4;
3250 assert_eq!(mmap.read(base).unwrap(), thread_id * 0x0A + 1);
3251 assert_eq!(mmap.read(base + 1).unwrap(), thread_id * 0x0A + 2);
3252 assert_eq!(mmap.read(base + 2).unwrap(), thread_id * 0x0A + 3);
3253 assert_eq!(mmap.read(base + 3).unwrap(), thread_id * 0x0A + 4);
3254 }
3255 }
3256
3257 #[test]
3258 fn ok_tx_durability_across_reopen() {
3259 let (_dir, cfg) = tmp_cfg();
3260
3261 {
3262 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3263 let mut tx = mmap.new_tx();
3264 tx.write(&0, &0xAAAA_BBBB, 0).unwrap();
3265 tx.write(&0, &0xCCCC_DDDD, 1).unwrap();
3266 let ticket = tx.commit().unwrap();
3267 ticket.wait().unwrap();
3268 assert!(ticket.is_durable());
3269 }
3270
3271 {
3272 let reopened = MemMap::<u64>::open(cfg).unwrap();
3273 assert_eq!(reopened.read(0).unwrap(), 0xAAAA_BBBB);
3274 assert_eq!(reopened.read(1).unwrap(), 0xCCCC_DDDD);
3275 }
3276 }
3277
3278 #[test]
3279 fn ok_tx_unparks_idle_bg_worker() {
3280 let (_dir, cfg) = tmp_cfg();
3281 let mmap = MemMap::<u64>::new(cfg).unwrap();
3282
3283 let mut parked = false;
3285 for _ in 0..0x64 {
3286 if mmap.is_parked() {
3287 parked = true;
3288 break;
3289 }
3290 std::thread::sleep(Duration::from_millis(0x0A));
3291 }
3292 assert!(parked);
3293
3294 let mut tx = mmap.new_tx();
3295 tx.write(&0, &0x22B, 0).unwrap();
3296 let ticket = tx.commit().unwrap();
3297
3298 ticket.wait().unwrap();
3300 assert!(ticket.is_durable());
3301 assert_eq!(mmap.read(0).unwrap(), 0x22B);
3302 }
3303 }
3304
3305 mod memmap_page_size {
3306 use super::*;
3307
3308 #[test]
3309 fn ok_page_size_returns_valid_power_of_two() {
3310 let page_size = MemMap::<u64>::page_size().unwrap();
3311 assert!(page_size >= 0x1000);
3312 assert!(page_size.is_power_of_two());
3313 }
3314
3315 #[test]
3316 fn ok_page_size_consistency() {
3317 let ps1 = MemMap::<u64>::page_size().unwrap();
3318 let ps2 = MemMap::<u64>::page_size().unwrap();
3319 let ps3 = MemMap::<ValidT>::page_size().unwrap();
3320 assert_eq!(ps1, ps2);
3321 assert_eq!(ps1, ps3);
3322 }
3323
3324 #[test]
3325 fn ok_page_size_across_threads() {
3326 let expected = MemMap::<u64>::page_size().unwrap();
3327 let mut handles = Vec::new();
3328
3329 for _ in 0..8 {
3330 handles.push(std::thread::spawn(move || {
3331 let ps = MemMap::<u64>::page_size().unwrap();
3332 assert_eq!(ps, expected);
3333 }));
3334 }
3335
3336 for h in handles {
3337 h.join().unwrap();
3338 }
3339 }
3340 }
3341}