1use crate::{
47 ack::{AckTicket, Completion, SyncTrigger, TEpoch},
48 error::{BindModule, ErrCode, FrozenError, FrozenResult},
49 file::{File, FileCfg},
50 hints,
51};
52use core::marker::PhantomData;
53use std::sync::{Arc, Condvar, Mutex, RwLock, atomic};
54
55mod interface;
56mod page_size;
57
58pub use page_size::PageSize;
59
60#[cfg(any(target_os = "linux", target_os = "macos"))]
61mod posix;
62
63#[cfg(target_os = "windows")]
64mod windows;
65
66pub(in crate::memmap) use interface::MemMapInterface;
67
68#[cfg(any(target_os = "linux", target_os = "macos"))]
69pub(in crate::memmap) type PlatformMemMap = posix::POSIXMemMap;
70
71#[cfg(target_os = "windows")]
72pub(in crate::memmap) type PlatformMemMap = windows::WINMemMap;
73
74pub(in crate::memmap) mod err {
76 use super::{ErrCode, FrozenError, FrozenResult};
77
78 const ERRDOMAIN: u8 = 0x10;
80
81 pub const HCF: ErrCode = ErrCode::new(0x02, "hault and catch fire");
83
84 pub const UNK: ErrCode = ErrCode::new(0x04, "unknown error");
86
87 pub const NMM: ErrCode = ErrCode::new(0x06, "not enough memory available on the device");
89
90 pub const SYN: ErrCode = ErrCode::new(0x08, "failed to sync/flush data to storage device");
92
93 pub const PRM: ErrCode = ErrCode::new(0x0A, "missing permissions for IO");
95
96 pub const FXE: ErrCode = ErrCode::new(0x10, "unable to spawn flush_tx");
98
99 pub const ZRO: ErrCode = ErrCode::new(0x0C, "type T must not be zero sized");
101
102 pub const DRP: ErrCode = ErrCode::new(0x12, "type T must not implement `Drop`");
104
105 pub const ALN: ErrCode = ErrCode::new(0x14, "type T must be 8-bytes aligned");
107
108 pub const SZE: ErrCode = ErrCode::new(0x16, "`size_of::<T>()` must be multiple of 8 bytes");
110
111 pub const INV: ErrCode =
114 ErrCode::new(0x18, "`size_of::<T>()` must not exceed system page size");
115
116 pub const DIR: ErrCode = ErrCode::new(0x1A, "directory does not exist");
118
119 pub const PTH: ErrCode = ErrCode::new(0x1C, "invalid path to file");
121
122 pub const EXS: ErrCode = ErrCode::new(0x1E, "file already exists");
124
125 pub const NEX: ErrCode = ErrCode::new(0x20, "file does not exist");
127
128 pub const GRW: ErrCode = ErrCode::new(0x22, "unable to grow memory-mapped file");
130
131 pub const CAS: ErrCode = ErrCode::new(0x24, "compare and swap check failed");
133
134 pub const TAG: ErrCode = ErrCode::new(0x26, "invalid name tag");
136
137 #[inline]
138 pub(in crate::memmap) fn raw_error<R, E: std::fmt::Display>(
139 code: ErrCode,
140 error: E,
141 ) -> FrozenResult<R> {
142 let err = FrozenError::unbound_raw(ERRDOMAIN, code, error);
143 Err(err)
144 }
145
146 #[inline]
147 pub(in crate::memmap) fn default_error<R>(code: ErrCode) -> FrozenResult<R> {
148 let err = FrozenError::unbound(ERRDOMAIN, code, "");
149 Err(err)
150 }
151
152 #[inline]
153 pub(in crate::memmap) fn make_raw_error<E: std::fmt::Display>(
154 code: ErrCode,
155 error: E,
156 ) -> FrozenError {
157 FrozenError::unbound_raw(ERRDOMAIN, code, error)
158 }
159}
160
161#[derive(Debug, Clone)]
189pub struct MemMapCfg {
190 pub module_id: u8,
192
193 pub path: std::path::PathBuf,
195
196 pub name: [u8; 4],
198
199 pub version: u32,
208
209 pub initial_pages: usize,
213
214 pub max_len: usize,
220
221 pub sync_interval: Option<std::time::Duration>,
227
228 pub memory_priority_hint: bool,
234
235 pub sequential_access_hint: bool,
242}
243
244impl MemMapCfg {
245 #[inline]
247 pub fn versioned_path(&self) -> std::path::PathBuf {
248 let name = core::str::from_utf8(&self.name).unwrap_or_default();
249 self.path.join(format!("{}_{}", name, self.version))
250 }
251}
252
253const WRITER_BIT: u32 = 0x8000_0000;
254const L1_CONTENTION: usize = 0x10;
255const L2_CONTENTION: usize = 0x20;
256
257#[derive(Debug)]
258struct SlotLocks {
259 locks: Box<[atomic::AtomicU32]>,
260}
261
262impl SlotLocks {
263 fn new(total_slots: usize) -> Self {
264 let mut slots = Vec::with_capacity(total_slots);
265 for _ in 0..total_slots {
266 slots.push(atomic::AtomicU32::new(0));
267 }
268
269 Self { locks: slots.into_boxed_slice() }
270 }
271
272 fn grow(&mut self, additional_slots: usize) {
273 if additional_slots == 0 {
274 return;
275 }
276
277 let total = self.locks.len() + additional_slots;
278 let mut slots = Vec::with_capacity(total);
279
280 for val in self.locks.iter() {
281 slots.push(atomic::AtomicU32::new(val.load(atomic::Ordering::Relaxed)));
282 }
283
284 for _ in 0..additional_slots {
285 slots.push(atomic::AtomicU32::new(0));
286 }
287
288 self.locks = slots.into_boxed_slice();
289 }
290
291 #[inline(always)]
292 fn read_lock(&self, index: usize) -> SlotReadGuard<'_> {
293 let val = &self.locks[index];
294 let mut spins = 0;
295
296 loop {
297 let state = val.load(atomic::Ordering::Relaxed);
298
299 if hints::likely(state & WRITER_BIT == 0)
300 && val
301 .compare_exchange_weak(
302 state,
303 state + 1,
304 atomic::Ordering::Acquire,
305 atomic::Ordering::Relaxed,
306 )
307 .is_ok()
308 {
309 return SlotReadGuard(val);
310 }
311
312 if hints::likely(spins < L1_CONTENTION) {
313 std::hint::spin_loop();
314 } else if spins < L2_CONTENTION {
315 std::thread::yield_now();
316 } else {
317 let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
318 std::thread::sleep(std::time::Duration::from_nanos(ns));
319 }
320
321 spins += 1;
322 }
323 }
324
325 #[inline(always)]
326 fn write_lock(&self, index: usize) -> SlotWriteGuard<'_> {
327 let val = &self.locks[index];
328 let mut spins = 0;
329
330 loop {
331 if val
332 .compare_exchange_weak(
333 0,
334 WRITER_BIT,
335 atomic::Ordering::Acquire,
336 atomic::Ordering::Relaxed,
337 )
338 .is_ok()
339 {
340 return SlotWriteGuard(val);
341 }
342
343 if hints::likely(spins < L1_CONTENTION) {
344 std::hint::spin_loop();
345 } else if spins < L2_CONTENTION {
346 std::thread::yield_now();
347 } else {
348 let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
349 std::thread::sleep(std::time::Duration::from_nanos(ns));
350 }
351
352 spins += 1;
353 }
354 }
355}
356
357struct SlotReadGuard<'a>(&'a atomic::AtomicU32);
359
360impl Drop for SlotReadGuard<'_> {
361 #[inline(always)]
362 fn drop(&mut self) {
363 self.0.fetch_sub(1, atomic::Ordering::Release);
364 }
365}
366
367struct SlotWriteGuard<'a>(&'a atomic::AtomicU32);
369
370impl Drop for SlotWriteGuard<'_> {
371 #[inline(always)]
372 fn drop(&mut self) {
373 self.0.store(0, atomic::Ordering::Release);
374 }
375}
376
377#[derive(Debug)]
379struct MemMapInner {
380 cfg: MemMapCfg,
381 page_size: PageSize,
382 mmap: RwLock<Option<PlatformMemMap>>,
383 slot_locks: RwLock<SlotLocks>,
384 length: atomic::AtomicUsize,
385 completion: Arc<Completion>,
386 sync_condvar: Condvar,
387 sync_mutex: Mutex<bool>,
388 shutdown: atomic::AtomicBool,
389 is_parked: atomic::AtomicBool,
390}
391
392impl SyncTrigger for MemMapInner {
393 fn trigger_sync(&self) -> FrozenResult<()> {
394 if self.cfg.sync_interval.is_some() {
395 {
396 let mut guard = self.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
397 *guard = true;
398 }
399
400 self.sync_condvar.notify_one();
401 Ok(())
402 } else {
403 self.sync_internal()
404 }
405 }
406}
407
408impl MemMapInner {
409 fn sync_mmap_if_needed(&self, mmap: &PlatformMemMap, sync_len: usize) -> FrozenResult<()> {
411 let target_epoch = self.completion.read_current_epoch();
412 let durable_epoch = self.completion.read_durable_epoch();
413
414 if target_epoch <= durable_epoch {
415 return Ok(());
416 }
417
418 let sync_res = mmap.sync(sync_len);
419 match sync_res {
420 Ok(()) => {
421 self.completion.mark_epoch_as_durable(target_epoch);
422 self.completion.del_err();
423 self.completion.notify_all_listeners();
424
425 Ok(())
426 }
427 Err(e) => {
428 let e = e.with_module(self.cfg.module_id);
429 self.completion.set_err(e.clone());
430 self.completion.notify_all_listeners();
431
432 Err(e)
433 }
434 }
435 }
436
437 fn sync_internal(&self) -> FrozenResult<()> {
438 let guard = self.mmap.read().unwrap_or_else(|e| e.into_inner());
439 let mmap = match guard.as_ref() {
440 Some(m) => m,
441 None => return Ok(()),
442 };
443
444 let length = self.length.load(atomic::Ordering::Acquire);
445 self.sync_mmap_if_needed(mmap, length)
446 }
447}
448
449pub struct MemMap<T: Sized + Clone> {
499 cfg: MemMapCfg,
500 file: Option<File>,
501 inner: Arc<MemMapInner>,
502 bg_thread: Option<std::thread::JoinHandle<()>>,
503 _marker: PhantomData<T>,
504}
505
506impl<T> core::fmt::Debug for MemMap<T>
507where
508 T: Sized + Clone,
509{
510 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
511 f.debug_struct("MemMap").field("cfg", &self.cfg).field("inner", &self.inner).finish()
512 }
513}
514
515unsafe impl<T> Send for MemMap<T> where T: Sized + Clone {}
516unsafe impl<T> Sync for MemMap<T> where T: Sized + Clone {}
517
518impl<T> MemMap<T>
519where
520 T: Sized + Clone,
521{
522 fn validate_t(cfg: &MemMapCfg) -> FrozenResult<PageSize> {
523 let mid = cfg.module_id;
524 if core::mem::needs_drop::<T>() {
525 return err::default_error(err::DRP).with_module(mid);
526 }
527
528 let size = core::mem::size_of::<T>();
529 if size == 0 {
530 return err::default_error(err::ZRO).with_module(mid);
531 }
532
533 if size % 8 != 0 {
534 return err::default_error(err::SZE).with_module(mid);
535 }
536
537 let align = core::mem::align_of::<T>();
538 if align != 8 {
539 return err::default_error(err::ALN).with_module(mid);
540 }
541
542 let page_size = PlatformMemMap::page_size().with_module(mid)?;
543
544 if size > page_size.as_usize() {
545 return err::default_error(err::INV).with_module(mid);
546 }
547
548 if page_size.as_usize() % size != 0 {
549 return err::raw_error(
550 err::INV,
551 format!("page size {page_size} must be divisible by size_of::<T>() ({size})"),
552 )
553 .with_module(mid);
554 }
555
556 if cfg.initial_pages == 0 {
557 return err::default_error(err::INV).with_module(mid);
558 }
559
560 let initial_bytes = match cfg.initial_pages.checked_mul(page_size.as_usize()) {
561 Some(b) => b,
562 None => {
563 return err::raw_error(err::INV, "initial_pages calculation overflowed usize")
564 .with_module(mid);
565 }
566 };
567
568 if cfg.max_len == 0 {
569 return err::default_error(err::INV).with_module(mid);
570 }
571
572 if cfg.max_len % page_size.as_usize() != 0 {
573 return err::raw_error(
574 err::INV,
575 format!("max_len {} must be a multiple of page size {page_size}", cfg.max_len),
576 )
577 .with_module(mid);
578 }
579
580 if cfg.max_len < initial_bytes {
581 return err::raw_error(
582 err::INV,
583 format!(
584 "max_len {} cannot be less than initial file size {initial_bytes}",
585 cfg.max_len
586 ),
587 )
588 .with_module(mid);
589 }
590
591 Ok(page_size)
592 }
593
594 fn validate_path(cfg: &MemMapCfg) -> FrozenResult<()> {
595 let mid = cfg.module_id;
596
597 if !cfg.path.exists() || !cfg.path.is_dir() {
598 return err::default_error(err::DIR).with_module(mid);
599 }
600
601 if !cfg.name.iter().all(|b| b.is_ascii_alphanumeric()) {
602 return err::default_error(err::TAG).with_module(mid);
603 }
604
605 Ok(())
606 }
607
608 fn file_cfg(cfg: &MemMapCfg, page_size: PageSize) -> FrozenResult<FileCfg> {
609 let initial_bytes = match cfg.initial_pages.checked_mul(page_size.as_usize()) {
610 Some(b) => b,
611 None => {
612 return err::raw_error(err::INV, "initial_pages calculation overflowed usize")
613 .with_module(cfg.module_id);
614 }
615 };
616
617 let size = core::mem::size_of::<T>();
618 let initial_available_buffers = initial_bytes / size;
619
620 Ok(FileCfg {
621 module_id: cfg.module_id,
622 path: cfg.versioned_path(),
623 buffer_size: size,
624 initial_available_buffers,
625 sync_interval: None,
626 sequential_access_hint: cfg.sequential_access_hint,
627 })
628 }
629
630 fn from_components(
631 cfg: MemMapCfg,
632 mmap: PlatformMemMap,
633 file: File,
634 page_size: PageSize,
635 ) -> FrozenResult<Self> {
636 if cfg.memory_priority_hint {
637 let _ = PlatformMemMap::apply_memory_priority_hint();
638 }
639
640 if cfg.sequential_access_hint {
641 let _ = mmap.advise_sequential(0, mmap.length());
642 }
643
644 let length = mmap.length();
645 let completion = Arc::new(Completion::default());
646 let sync_interval = cfg.sync_interval;
647 let total_slots = length / core::mem::size_of::<T>();
648 let slot_locks = SlotLocks::new(total_slots);
649
650 let inner = Arc::new(MemMapInner {
651 cfg: cfg.clone(),
652 page_size,
653 mmap: RwLock::new(Some(mmap)),
654 slot_locks: RwLock::new(slot_locks),
655 length: atomic::AtomicUsize::new(length),
656 completion: completion.clone(),
657 sync_condvar: Condvar::new(),
658 sync_mutex: Mutex::new(false),
659 shutdown: atomic::AtomicBool::new(false),
660 is_parked: atomic::AtomicBool::new(false),
661 });
662
663 let weak_inner: std::sync::Weak<dyn SyncTrigger> =
664 Arc::downgrade(&inner) as std::sync::Weak<dyn SyncTrigger>;
665 completion.set_sync_trigger(weak_inner);
666
667 let bg_thread = if let Some(interval) = sync_interval {
668 let worker_inner = Arc::clone(&inner);
669
670 let handle = std::thread::Builder::new().name("frozen-mmap-sync-worker".into()).spawn(
671 move || {
672 let mut guard =
673 worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
674
675 while !worker_inner.shutdown.load(atomic::Ordering::Acquire) {
676 if *guard {
677 *guard = false;
678 drop(guard);
679
680 let _ = worker_inner.sync_internal();
681 guard =
682 worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
683
684 continue;
685 }
686
687 let target_epoch = worker_inner.completion.read_current_epoch();
688 let durable_epoch = worker_inner.completion.read_durable_epoch();
689
690 if target_epoch == durable_epoch {
691 worker_inner.is_parked.store(true, atomic::Ordering::Release);
692
693 if worker_inner.completion.read_current_epoch() != durable_epoch {
694 worker_inner.is_parked.store(false, atomic::Ordering::Release);
695 continue;
696 }
697
698 let new_guard = worker_inner
699 .sync_condvar
700 .wait(guard)
701 .unwrap_or_else(|e| e.into_inner());
702
703 guard = new_guard;
704 worker_inner.is_parked.store(false, atomic::Ordering::Release);
705
706 if worker_inner.shutdown.load(atomic::Ordering::Acquire) || *guard {
707 continue;
708 }
709
710 if worker_inner.completion.read_current_epoch()
711 == worker_inner.completion.read_durable_epoch()
712 {
713 continue;
714 }
715
716 let (new_guard, _) = worker_inner
717 .sync_condvar
718 .wait_timeout(guard, interval)
719 .unwrap_or_else(|e| e.into_inner());
720 guard = new_guard;
721 } else {
722 worker_inner.is_parked.store(false, atomic::Ordering::Release);
723
724 let (new_guard, _) = worker_inner
725 .sync_condvar
726 .wait_timeout(guard, interval)
727 .unwrap_or_else(|e| e.into_inner());
728
729 guard = new_guard;
730 }
731
732 if worker_inner.shutdown.load(atomic::Ordering::Acquire) {
733 break;
734 }
735
736 *guard = false;
737 drop(guard);
738
739 let _ = worker_inner.sync_internal();
740 guard = worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
741 }
742 },
743 );
744
745 let handle = match handle {
746 Ok(h) => h,
747 Err(spawn_err) => {
748 let mut mmap_guard = inner.mmap.write().unwrap_or_else(|e| e.into_inner());
749 let mut err = err::make_raw_error(err::FXE, spawn_err);
750
751 if let Some(m) = mmap_guard.take() {
752 if let Err(unmap_err) = m.unmap() {
753 err.add_suppressed(unmap_err);
754 }
755 }
756
757 return Err(err.with_module(cfg.module_id));
758 }
759 };
760
761 Some(handle)
762 } else {
763 None
764 };
765
766 Ok(Self { cfg, file: Some(file), inner, bg_thread, _marker: PhantomData })
767 }
768
769 pub fn new(cfg: MemMapCfg) -> FrozenResult<Self> {
810 let page_size = Self::validate_t(&cfg)?;
811 Self::validate_path(&cfg)?;
812
813 let vpath = cfg.versioned_path();
814 if vpath.exists() {
815 return err::default_error(err::EXS).with_module(cfg.module_id);
816 }
817
818 let file_cfg = Self::file_cfg(&cfg, page_size)?;
819 let file = match File::new(file_cfg) {
820 Ok(f) => f,
821 Err(e) if e.reason == err::EXS.reason => {
822 return err::default_error(err::EXS).with_module(cfg.module_id);
823 }
824 Err(e) => return Err(e),
825 };
826
827 let mmap = match PlatformMemMap::new(file.fd(), file.length(), cfg.max_len) {
828 Ok(m) => m,
829 Err(mut e) => {
830 if let Err(del_err) = file.delete() {
831 e.add_suppressed(del_err);
832 }
833
834 return Err(e.with_module(cfg.module_id));
835 }
836 };
837
838 Self::from_components(cfg, mmap, file, page_size)
839 }
840
841 pub fn open(cfg: MemMapCfg) -> FrozenResult<Self> {
884 let page_size = Self::validate_t(&cfg)?;
885 Self::validate_path(&cfg)?;
886
887 let vpath = cfg.versioned_path();
888 if !vpath.exists() {
889 return err::default_error(err::NEX).with_module(cfg.module_id);
890 }
891
892 if vpath.is_dir() {
893 return err::default_error(err::PTH).with_module(cfg.module_id);
894 }
895
896 let file_cfg = Self::file_cfg(&cfg, page_size)?;
897 let file = File::open(file_cfg)?;
898
899 if cfg.max_len < file.length() {
900 return err::raw_error(
901 err::INV,
902 format!(
903 "max_len {} cannot be less than existing file length {}",
904 cfg.max_len,
905 file.length()
906 ),
907 )
908 .with_module(cfg.module_id);
909 }
910
911 let mmap = PlatformMemMap::new(file.fd(), file.length(), cfg.max_len)
912 .with_module(cfg.module_id)?;
913 Self::from_components(cfg, mmap, file, page_size)
914 }
915
916 pub fn open_or_create(cfg: MemMapCfg) -> FrozenResult<Self> {
959 let page_size = Self::validate_t(&cfg)?;
960 Self::validate_path(&cfg)?;
961
962 let vpath = cfg.versioned_path();
963 if vpath.exists() && vpath.is_dir() {
964 return err::default_error(err::PTH).with_module(cfg.module_id);
965 }
966
967 let file_cfg = Self::file_cfg(&cfg, page_size)?;
968 let file = File::open_or_create(file_cfg)?;
969
970 if cfg.max_len < file.length() {
971 return err::raw_error(
972 err::INV,
973 format!(
974 "max_len {} cannot be less than existing file length {}",
975 cfg.max_len,
976 file.length()
977 ),
978 )
979 .with_module(cfg.module_id);
980 }
981
982 let mmap = PlatformMemMap::new(file.fd(), file.length(), cfg.max_len)
983 .with_module(cfg.module_id)?;
984 Self::from_components(cfg, mmap, file, page_size)
985 }
986
987 #[inline]
1015 pub fn exists(cfg: &MemMapCfg) -> bool {
1016 if !cfg.path.is_dir() || !cfg.name.iter().all(|b| b.is_ascii_alphanumeric()) {
1017 return false;
1018 }
1019
1020 cfg.versioned_path().is_file()
1021 }
1022
1023 #[inline]
1035 pub fn page_size() -> FrozenResult<PageSize> {
1036 PlatformMemMap::page_size()
1037 }
1038
1039 #[inline]
1067 pub fn apply_memory_priority_hint(&self) -> FrozenResult<()> {
1068 PlatformMemMap::apply_memory_priority_hint()
1069 }
1070
1071 #[inline]
1095 pub fn advise_sequential(&self) -> FrozenResult<()> {
1096 let length = self.inner.length.load(atomic::Ordering::Acquire);
1097 self.advise_sequential_range(0, length)
1098 }
1099
1100 #[inline]
1103 pub fn advise_sequential_range(&self, offset: usize, length: usize) -> FrozenResult<()> {
1104 let mmap_guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1105 let mmap = match mmap_guard.as_ref() {
1106 Some(m) => m,
1107 None => return err::default_error(err::INV).with_module(self.cfg.module_id),
1108 };
1109
1110 mmap.advise_sequential(offset, length).with_module(self.cfg.module_id)
1111 }
1112
1113 #[inline]
1115 pub fn prefetch(&self, offset: usize, length: usize) -> FrozenResult<()> {
1116 let mmap_guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1117 let mmap = match mmap_guard.as_ref() {
1118 Some(m) => m,
1119 None => return err::default_error(err::INV).with_module(self.cfg.module_id),
1120 };
1121
1122 mmap.prefetch(offset, length).with_module(self.cfg.module_id)
1123 }
1124
1125 #[inline]
1152 pub fn length(&self) -> usize {
1153 self.inner.length.load(atomic::Ordering::Acquire)
1154 }
1155
1156 #[inline]
1192 pub fn resident_bytes(&self) -> FrozenResult<usize> {
1193 let mmap_guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1194 let mmap = match mmap_guard.as_ref() {
1195 Some(m) => m,
1196 None => return err::default_error(err::INV).with_module(self.cfg.module_id),
1197 };
1198
1199 mmap.resident_bytes().with_module(self.cfg.module_id)
1200 }
1201
1202 pub fn grow(&self, pages: usize) -> FrozenResult<()> {
1259 let mid = self.cfg.module_id;
1260 if pages == 0 {
1261 return Ok(());
1262 }
1263
1264 let page_size = self.inner.page_size.as_usize();
1265 let bytes_to_add = match pages.checked_mul(page_size) {
1266 Some(bytes) => bytes,
1267 None => {
1268 return err::raw_error(err::GRW, "file growth calculation overflowed usize")
1269 .with_module(mid);
1270 }
1271 };
1272
1273 let size = core::mem::size_of::<T>();
1274 let additional_slots = bytes_to_add / size;
1275
1276 let file = match self.file.as_ref() {
1277 Some(f) => f,
1278 None => return err::default_error(err::INV).with_module(mid),
1279 };
1280
1281 let mut slot_locks_guard = self.inner.slot_locks.write().unwrap_or_else(|e| e.into_inner());
1287 let mut mmap_guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
1288 let mmap = match mmap_guard.as_mut() {
1289 Some(m) => m,
1290 None => return err::default_error(err::INV).with_module(mid),
1291 };
1292
1293 let curr_len = self.inner.length.load(atomic::Ordering::Acquire);
1294 let new_length = match curr_len.checked_add(bytes_to_add) {
1295 Some(len) => len,
1296 None => {
1297 return err::raw_error(err::GRW, "file growth calculation overflowed usize")
1298 .with_module(mid);
1299 }
1300 };
1301
1302 if new_length > self.cfg.max_len {
1303 return err::raw_error(err::GRW, "file growth exceeds configured max_len")
1304 .with_module(mid);
1305 }
1306
1307 self.inner.sync_mmap_if_needed(mmap, curr_len)?;
1310
1311 file.grow(additional_slots).with_module(mid)?;
1312
1313 file.sync().with_module(mid)?;
1315
1316 mmap.grow(file.fd(), curr_len, bytes_to_add).with_module(mid)?;
1317
1318 if self.cfg.sequential_access_hint {
1319 let _ = mmap.advise_sequential(curr_len, bytes_to_add);
1320 }
1321
1322 slot_locks_guard.grow(additional_slots);
1323 self.inner.length.store(new_length, atomic::Ordering::Release);
1324
1325 drop(mmap_guard);
1326 drop(slot_locks_guard);
1327
1328 self.inner.completion.increment_current_epoch();
1329 Ok(())
1330 }
1331
1332 #[inline]
1359 pub fn cfg(&self) -> &MemMapCfg {
1360 &self.cfg
1361 }
1362
1363 #[inline]
1393 pub fn current_epoch(&self) -> TEpoch {
1394 self.inner.completion.read_current_epoch()
1395 }
1396
1397 #[inline]
1425 pub fn durable_epoch(&self) -> TEpoch {
1426 self.inner.completion.read_durable_epoch()
1427 }
1428
1429 #[inline]
1464 pub fn sync(&self) -> FrozenResult<()> {
1465 self.inner.sync_internal()
1466 }
1467
1468 #[inline(always)]
1507 pub fn read<R, F>(&self, index: usize, f: F) -> FrozenResult<R>
1508 where
1509 F: FnOnce(&T) -> R,
1510 {
1511 let mid = self.cfg.module_id;
1512 let size = core::mem::size_of::<T>();
1513
1514 let offset = match index.checked_mul(size) {
1515 Some(off) => off,
1516 None => return err::default_error(err::INV).with_module(mid),
1517 };
1518
1519 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1520 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1521 return err::default_error(err::HCF).with_module(mid);
1522 }
1523
1524 let _slot_guard = slot_locks_guard.read_lock(index);
1525 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1526
1527 let mmap = match guard.as_ref() {
1528 Some(m) => m,
1529 None => return err::default_error(err::INV).with_module(mid),
1530 };
1531
1532 let ptr = unsafe { mmap.as_ptr::<T>(offset) };
1533 let target_ref = unsafe { &*ptr };
1534 let res = f(target_ref);
1535
1536 drop(guard);
1537 drop(_slot_guard);
1538 drop(slot_locks_guard);
1539
1540 Ok(res)
1541 }
1542
1543 #[inline(always)]
1595 pub fn read_exact<Target, R, F>(&self, start: usize, f: F) -> FrozenResult<R>
1596 where
1597 F: FnOnce(&Target) -> R,
1598 {
1599 let mid = self.cfg.module_id;
1600 let slot_size = core::mem::size_of::<T>();
1601 let target_size = core::mem::size_of::<Target>();
1602
1603 if hints::unlikely(target_size == 0) {
1604 return err::default_error(err::ZRO).with_module(mid);
1605 }
1606
1607 if hints::unlikely(target_size % slot_size != 0) {
1608 return err::default_error(err::SZE).with_module(mid);
1609 }
1610
1611 if hints::unlikely(core::mem::align_of::<Target>() > core::mem::align_of::<T>()) {
1612 return err::default_error(err::ALN).with_module(mid);
1613 }
1614
1615 let n = target_size / slot_size;
1616 let end_slot = match start.checked_add(n) {
1617 Some(end) => end,
1618 None => return err::default_error(err::INV).with_module(mid),
1619 };
1620
1621 let offset = match start.checked_mul(slot_size) {
1622 Some(off) => off,
1623 None => return err::default_error(err::INV).with_module(mid),
1624 };
1625
1626 let end_offset = match end_slot.checked_mul(slot_size) {
1627 Some(end) => end,
1628 None => return err::default_error(err::INV).with_module(mid),
1629 };
1630
1631 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1632 if hints::unlikely(end_offset > self.length()) {
1633 return err::default_error(err::HCF).with_module(mid);
1634 }
1635
1636 let mut slot_guards = Vec::with_capacity(n);
1637 for slot in start..end_slot {
1638 slot_guards.push(slot_locks_guard.read_lock(slot));
1639 }
1640
1641 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1642 let mmap = match guard.as_ref() {
1643 Some(m) => m,
1644 None => return err::default_error(err::INV).with_module(mid),
1645 };
1646
1647 let ptr = unsafe { mmap.as_ptr::<Target>(offset) };
1648 if hints::unlikely((ptr as usize) % core::mem::align_of::<Target>() != 0) {
1649 return err::default_error(err::ALN).with_module(mid);
1650 }
1651
1652 let target_ref = unsafe { &*ptr };
1653 let res = f(target_ref);
1654
1655 drop(guard);
1656 drop(slot_guards);
1657 drop(slot_locks_guard);
1658
1659 Ok(res)
1660 }
1661
1662 #[inline(always)]
1702 pub fn write<R, F>(&self, index: usize, f: F) -> FrozenResult<(AckTicket, R)>
1703 where
1704 F: FnOnce(&mut T) -> R,
1705 {
1706 let mid = self.cfg.module_id;
1707 let size = core::mem::size_of::<T>();
1708
1709 let offset = match index.checked_mul(size) {
1710 Some(off) => off,
1711 None => return err::default_error(err::INV).with_module(mid),
1712 };
1713
1714 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1715 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1716 return err::default_error(err::HCF).with_module(mid);
1717 }
1718
1719 let _slot_guard = slot_locks_guard.write_lock(index);
1720 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1721
1722 let mmap = match guard.as_ref() {
1723 Some(m) => m,
1724 None => return err::default_error(err::INV).with_module(mid),
1725 };
1726
1727 let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1728 let res = f(unsafe { &mut *ptr });
1729
1730 drop(guard);
1731 drop(_slot_guard);
1732 drop(slot_locks_guard);
1733
1734 let epoch = self.inner.completion.increment_current_epoch();
1735 if self.inner.is_parked.load(atomic::Ordering::Acquire) {
1736 let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1737 if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1738 self.inner.sync_condvar.notify_one();
1739 }
1740 }
1741
1742 Ok((AckTicket::new(epoch, self.inner.completion.clone()), res))
1743 }
1744
1745 #[inline(always)]
1803 pub fn write_exact<Target, R, F>(&self, start: usize, f: F) -> FrozenResult<(AckTicket, R)>
1804 where
1805 F: FnOnce(&mut Target) -> R,
1806 {
1807 let mid = self.cfg.module_id;
1808 let slot_size = core::mem::size_of::<T>();
1809 let target_size = core::mem::size_of::<Target>();
1810
1811 if hints::unlikely(target_size == 0) {
1812 return err::default_error(err::ZRO).with_module(mid);
1813 }
1814
1815 if hints::unlikely(target_size % slot_size != 0) {
1816 return err::default_error(err::SZE).with_module(mid);
1817 }
1818
1819 if hints::unlikely(core::mem::align_of::<Target>() > core::mem::align_of::<T>()) {
1820 return err::default_error(err::ALN).with_module(mid);
1821 }
1822
1823 let n = target_size / slot_size;
1824 let end_slot = match start.checked_add(n) {
1825 Some(end) => end,
1826 None => return err::default_error(err::INV).with_module(mid),
1827 };
1828
1829 let offset = match start.checked_mul(slot_size) {
1830 Some(off) => off,
1831 None => return err::default_error(err::INV).with_module(mid),
1832 };
1833
1834 let end_offset = match end_slot.checked_mul(slot_size) {
1835 Some(end) => end,
1836 None => return err::default_error(err::INV).with_module(mid),
1837 };
1838
1839 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1840 if hints::unlikely(end_offset > self.length()) {
1841 return err::default_error(err::HCF).with_module(mid);
1842 }
1843
1844 let mut slot_guards = Vec::with_capacity(n);
1845 for slot in start..end_slot {
1846 slot_guards.push(slot_locks_guard.write_lock(slot));
1847 }
1848
1849 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1850 let mmap = match guard.as_ref() {
1851 Some(m) => m,
1852 None => return err::default_error(err::INV).with_module(mid),
1853 };
1854
1855 let ptr = unsafe { mmap.as_mut_ptr::<Target>(offset) };
1856 if hints::unlikely((ptr as usize) % core::mem::align_of::<Target>() != 0) {
1857 return err::default_error(err::ALN).with_module(mid);
1858 }
1859
1860 let target_ref = unsafe { &mut *ptr };
1861 let res = f(target_ref);
1862
1863 drop(guard);
1864 drop(slot_guards);
1865 drop(slot_locks_guard);
1866
1867 let epoch = self.inner.completion.increment_current_epoch();
1868 if self.inner.is_parked.load(atomic::Ordering::Acquire) {
1869 let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1870
1871 if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1872 self.inner.sync_condvar.notify_one();
1873 }
1874 }
1875
1876 Ok((AckTicket::new(epoch, self.inner.completion.clone()), res))
1877 }
1878
1879 #[cfg(test)]
1880 pub(crate) fn is_parked(&self) -> bool {
1881 self.inner.is_parked.load(atomic::Ordering::Acquire)
1882 }
1883
1884 pub fn delete(mut self) -> FrozenResult<()> {
1923 let mid = self.cfg.module_id;
1924 self.inner.shutdown.store(true, atomic::Ordering::Release);
1925
1926 {
1927 let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1928 *guard = true;
1929 }
1930
1931 self.inner.sync_condvar.notify_all();
1932 if let Some(handle) = self.bg_thread.take() {
1933 let _ = handle.join();
1934 }
1935
1936 let mut guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
1937 let unmap_res =
1938 if let Some(mmap) = guard.take() { mmap.unmap().with_module(mid) } else { Ok(()) };
1939
1940 let file = match self.file.take() {
1941 Some(f) => f,
1942 None => {
1943 unmap_res?;
1944 return err::default_error(err::INV).with_module(mid);
1945 }
1946 };
1947
1948 let delete_res = file.delete().with_module(mid);
1949 match (unmap_res, delete_res) {
1950 (Err(mut unmap_err), Err(del_err)) => {
1951 unmap_err.add_suppressed(del_err);
1952 Err(unmap_err)
1953 }
1954 (Err(unmap_err), Ok(())) => Err(unmap_err),
1955 (Ok(()), Err(del_err)) => Err(del_err),
1956 (Ok(()), Ok(())) => Ok(()),
1957 }
1958 }
1959}
1960
1961impl<T> MemMap<T>
1962where
1963 T: Sized + Clone + PartialEq,
1964{
1965 pub fn compare_and_write(
2013 &self,
2014 expected: &T,
2015 val: &T,
2016 index: usize,
2017 ) -> FrozenResult<AckTicket> {
2018 let mid = self.cfg.module_id;
2019 let size = core::mem::size_of::<T>();
2020
2021 let offset = match index.checked_mul(size) {
2022 Some(off) => off,
2023 None => return err::default_error(err::INV).with_module(mid),
2024 };
2025
2026 let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
2027 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
2028 return err::default_error(err::HCF).with_module(mid);
2029 }
2030
2031 let _slot_guard = slot_locks_guard.write_lock(index);
2032 let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
2033
2034 let mmap = match guard.as_ref() {
2035 Some(m) => m,
2036 None => return err::default_error(err::INV).with_module(mid),
2037 };
2038
2039 let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
2040 let current = unsafe { ptr.read() };
2041
2042 if current != *expected {
2043 drop(guard);
2044 drop(_slot_guard);
2045 drop(slot_locks_guard);
2046
2047 return err::raw_error(
2048 err::CAS,
2049 format!("state conflict at index {}: slot data has changed", index),
2050 )
2051 .with_module(mid);
2052 }
2053
2054 unsafe { core::ptr::write(ptr, val.clone()) };
2055
2056 drop(guard);
2057 drop(_slot_guard);
2058 drop(slot_locks_guard);
2059
2060 let epoch = self.inner.completion.increment_current_epoch();
2061 if self.inner.is_parked.load(atomic::Ordering::Acquire) {
2062 let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
2063
2064 if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
2065 self.inner.sync_condvar.notify_one();
2066 }
2067 }
2068
2069 Ok(AckTicket::new(epoch, self.inner.completion.clone()))
2070 }
2071
2072 #[inline]
2116 pub fn new_tx(&self) -> MemMapTransaction<'_, T> {
2117 MemMapTransaction { memmap: self, ops: Vec::new() }
2118 }
2119}
2120
2121#[derive(Debug)]
2160pub struct MemMapTransaction<'a, T: Sized + Clone + PartialEq> {
2161 memmap: &'a MemMap<T>,
2162 ops: Vec<(usize, T, T)>,
2163}
2164
2165impl<T> MemMapTransaction<'_, T>
2166where
2167 T: Sized + Clone + PartialEq,
2168{
2169 #[inline(always)]
2207 pub fn write(&mut self, expected: &T, new: &T, index: usize) -> FrozenResult<&mut Self> {
2208 let mid = self.memmap.cfg.module_id;
2209 if let Some(err) = self.memmap.inner.completion.get_err() {
2210 return Err(err);
2211 }
2212
2213 if let Some((last_idx, _, _)) = self.ops.last() {
2214 if index <= *last_idx {
2215 return err::raw_error(
2216 err::HCF,
2217 "tx writes must be strictly ordered, with no more then single ops on given index",
2218 )
2219 .with_module(mid);
2220 }
2221 }
2222
2223 let size = core::mem::size_of::<T>();
2224 let offset = match index.checked_mul(size) {
2225 Some(off) => off,
2226 None => return err::default_error(err::INV).with_module(mid),
2227 };
2228
2229 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.memmap.length())) {
2230 return err::default_error(err::HCF).with_module(mid);
2231 }
2232
2233 self.ops.push((index, expected.clone(), new.clone()));
2234 Ok(self)
2235 }
2236
2237 pub fn commit(self) -> FrozenResult<AckTicket> {
2275 let mid = self.memmap.cfg.module_id;
2276 if self.ops.is_empty() {
2277 return err::raw_error(err::INV, "empty transaction").with_module(mid);
2278 }
2279
2280 if let Some(err) = self.memmap.inner.completion.get_err() {
2281 return Err(err);
2282 }
2283
2284 let size = core::mem::size_of::<T>();
2285 let slot_locks_guard =
2286 self.memmap.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
2287
2288 let current_length = self.memmap.length();
2289 for (index, _, _) in &self.ops {
2290 let offset = match index.checked_mul(size) {
2291 Some(off) => off,
2292 None => return err::default_error(err::INV).with_module(mid),
2293 };
2294
2295 if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > current_length)) {
2296 return err::default_error(err::HCF).with_module(mid);
2297 }
2298 }
2299
2300 let mut slot_guards = Vec::with_capacity(self.ops.len());
2301 for (index, _, _) in &self.ops {
2302 slot_guards.push(slot_locks_guard.write_lock(*index));
2303 }
2304
2305 let guard = self.memmap.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
2306 let mmap = match guard.as_ref() {
2307 Some(m) => m,
2308 None => return err::default_error(err::INV).with_module(mid),
2309 };
2310
2311 for (index, expected, _) in &self.ops {
2313 let offset = *index * size;
2314 let ptr = unsafe { mmap.as_ptr::<T>(offset) };
2315 let current = unsafe { ptr.read() };
2316
2317 if current != *expected {
2318 drop(guard);
2319 drop(slot_guards);
2320 drop(slot_locks_guard);
2321
2322 return err::raw_error(
2323 err::CAS,
2324 format!("state conflict at index {}: slot data has changed", index),
2325 )
2326 .with_module(mid);
2327 }
2328 }
2329
2330 for (index, _, new) in self.ops {
2332 let offset = index * size;
2333 let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
2334
2335 unsafe { core::ptr::write(ptr, new) };
2336 }
2337
2338 drop(guard);
2339 drop(slot_guards);
2340 drop(slot_locks_guard);
2341
2342 let epoch = self.memmap.inner.completion.increment_current_epoch();
2343 if self.memmap.inner.is_parked.load(atomic::Ordering::Acquire) {
2344 let _guard = self.memmap.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
2345 if self.memmap.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
2346 self.memmap.inner.sync_condvar.notify_one();
2347 }
2348 }
2349
2350 Ok(AckTicket::new(epoch, self.memmap.inner.completion.clone()))
2351 }
2352
2353 #[inline]
2382 pub fn len(&self) -> usize {
2383 self.ops.len()
2384 }
2385
2386 #[inline]
2412 pub fn is_empty(&self) -> bool {
2413 self.ops.is_empty()
2414 }
2415}
2416
2417impl<T> Drop for MemMap<T>
2418where
2419 T: Sized + Clone,
2420{
2421 fn drop(&mut self) {
2422 self.inner.shutdown.store(true, atomic::Ordering::Release);
2423
2424 {
2425 let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
2426 *guard = true;
2427 }
2428
2429 self.inner.sync_condvar.notify_all();
2430
2431 if let Some(handle) = self.bg_thread.take() {
2432 let _ = handle.join();
2433 }
2434
2435 let _ = self.inner.sync_internal();
2436 }
2437}
2438
2439#[cfg(test)]
2440mod tests {
2441 use super::*;
2442 use std::time::Duration;
2443
2444 const MID: u8 = 0;
2445 const INIT_PAGES: usize = 1;
2446 const INIT_T: usize = 512;
2447
2448 #[derive(Debug, Clone)]
2449 #[repr(C, align(8))]
2450 struct ValidT {
2451 a: u64,
2452 b: u64,
2453 }
2454
2455 #[derive(Debug, Clone)]
2456 #[repr(C)]
2457 struct BadAlignT {
2458 a: u32,
2459 b: u32,
2460 }
2461
2462 #[derive(Debug, Clone)]
2463 #[repr(C)]
2464 struct BadSizeT {
2465 a: [u8; 0x0C],
2466 }
2467
2468 #[derive(Debug, Clone)]
2469 #[repr(C, align(8))]
2470 struct DropT(u64);
2471
2472 impl Drop for DropT {
2473 fn drop(&mut self) {}
2474 }
2475
2476 #[derive(Debug, Clone)]
2477 #[repr(C, align(8))]
2478 struct ZstT;
2479
2480 #[derive(Debug, Clone)]
2481 #[repr(C, align(8))]
2482 struct OversizedT {
2483 data: [u8; 0x20000],
2484 }
2485
2486 fn tmp_cfg() -> (tempfile::TempDir, MemMapCfg) {
2487 let dir = tempfile::tempdir().unwrap();
2488 let cfg = MemMapCfg {
2489 module_id: MID,
2490 path: dir.path().to_path_buf(),
2491 name: *b"tbfx",
2492 version: 1,
2493 initial_pages: INIT_PAGES,
2494 max_len: 0x10_0000,
2495 sync_interval: Some(Duration::from_millis(0x64)),
2496 memory_priority_hint: false,
2497 sequential_access_hint: false,
2498 };
2499
2500 (dir, cfg)
2501 }
2502
2503 #[test]
2504 fn ok_validate_t() {
2505 let (_dir, cfg) = tmp_cfg();
2506 assert!(MemMap::<u64>::validate_t(&cfg).is_ok());
2507 let mut cfg_valid = cfg.clone();
2508 cfg_valid.initial_pages = 2;
2509 assert!(MemMap::<ValidT>::validate_t(&cfg_valid).is_ok());
2510 }
2511
2512 #[test]
2513 fn err_validate_t_invariants() {
2514 let (_dir, cfg) = tmp_cfg();
2515 assert_eq!(MemMap::<DropT>::validate_t(&cfg).unwrap_err().reason, err::DRP.reason);
2516 assert_eq!(MemMap::<BadAlignT>::validate_t(&cfg).unwrap_err().reason, err::ALN.reason);
2517 assert_eq!(MemMap::<ZstT>::validate_t(&cfg).unwrap_err().reason, err::ZRO.reason);
2518 assert_eq!(MemMap::<BadSizeT>::validate_t(&cfg).unwrap_err().reason, err::SZE.reason);
2519 assert_eq!(MemMap::<OversizedT>::validate_t(&cfg).unwrap_err().reason, err::INV.reason);
2520
2521 let mut zero_cfg = cfg.clone();
2522 zero_cfg.initial_pages = 0;
2523 assert_eq!(MemMap::<u64>::validate_t(&zero_cfg).unwrap_err().reason, err::INV.reason);
2524
2525 let mut overflow_cfg = cfg.clone();
2526 overflow_cfg.initial_pages = usize::MAX;
2527 assert_eq!(MemMap::<u64>::validate_t(&overflow_cfg).unwrap_err().reason, err::INV.reason);
2528
2529 let mut zero_max_len = cfg.clone();
2530 zero_max_len.max_len = 0;
2531 assert_eq!(MemMap::<u64>::validate_t(&zero_max_len).unwrap_err().reason, err::INV.reason);
2532
2533 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2534 let mut unaligned_max_len = cfg.clone();
2535 unaligned_max_len.max_len = page_size + 1;
2536 assert_eq!(
2537 MemMap::<u64>::validate_t(&unaligned_max_len).unwrap_err().reason,
2538 err::INV.reason
2539 );
2540
2541 let mut small_max_len = cfg.clone();
2542 small_max_len.initial_pages = 2;
2543 small_max_len.max_len = page_size;
2544 assert_eq!(MemMap::<u64>::validate_t(&small_max_len).unwrap_err().reason, err::INV.reason);
2545 }
2546
2547 #[test]
2548 fn err_open_when_max_len_less_than_existing_file() {
2549 let (_dir, mut cfg) = tmp_cfg();
2550 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2551 cfg.initial_pages = 2;
2552 cfg.max_len = 4 * page_size;
2553
2554 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2555 assert_eq!(mmap.length(), 2 * page_size);
2556 drop(mmap);
2557
2558 let mut open_cfg = cfg.clone();
2559 open_cfg.initial_pages = 1;
2560 open_cfg.max_len = page_size;
2561 let err = MemMap::<u64>::open(open_cfg).unwrap_err();
2562 assert_eq!(err.reason, err::INV.reason);
2563 }
2564
2565 #[test]
2566 fn ok_new_sets_file_sync_interval_none() {
2567 let (_dir, cfg) = tmp_cfg();
2568
2569 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2570 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2571 assert_eq!(mmap.length(), page_size * INIT_PAGES);
2572 assert_eq!(mmap.cfg().sync_interval, Some(Duration::from_millis(0x64)));
2573 }
2574
2575 #[test]
2576 fn err_new_when_file_exists() {
2577 let (_dir, cfg) = tmp_cfg();
2578 let _mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2579 let err = MemMap::<u64>::new(cfg).unwrap_err();
2580 assert_eq!(err.reason, err::EXS.reason);
2581 }
2582
2583 #[test]
2584 fn err_new_when_t_invalid() {
2585 let (_dir, cfg) = tmp_cfg();
2586 assert_eq!(MemMap::<DropT>::new(cfg.clone()).unwrap_err().reason, err::DRP.reason);
2587 assert!(!cfg.versioned_path().exists());
2588 }
2589
2590 #[test]
2591 fn ok_open() {
2592 let (_dir, cfg) = tmp_cfg();
2593 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2594 {
2595 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2596 assert_eq!(mmap.length(), page_size * INIT_PAGES);
2597 }
2598
2599 let opened = MemMap::<u64>::open(cfg).unwrap();
2600 assert_eq!(opened.length(), page_size * INIT_PAGES);
2601 }
2602
2603 #[test]
2604 fn err_open_when_file_missing() {
2605 let (_dir, cfg) = tmp_cfg();
2606 let err = MemMap::<u64>::open(cfg).unwrap_err();
2607 assert_eq!(err.reason, err::NEX.reason);
2608 }
2609
2610 #[test]
2611 fn ok_open_or_create() {
2612 let (_dir, cfg) = tmp_cfg();
2613 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2614 {
2615 let mmap = MemMap::<u64>::open_or_create(cfg.clone()).unwrap();
2616 assert_eq!(mmap.length(), page_size * INIT_PAGES);
2617 }
2618
2619 let reopened = MemMap::<u64>::open_or_create(cfg).unwrap();
2620 assert_eq!(reopened.length(), page_size * INIT_PAGES);
2621 }
2622
2623 #[test]
2624 fn err_path_when_dir_missing() {
2625 let dir = tempfile::tempdir().unwrap();
2626 let non_existent_dir = dir.path().join("missing_dir_xyz");
2627 let cfg = MemMapCfg {
2628 module_id: MID,
2629 path: non_existent_dir,
2630 name: *b"tbfx",
2631 version: 1,
2632 initial_pages: INIT_PAGES,
2633 max_len: 0x10_0000,
2634 sync_interval: Some(Duration::from_millis(0x64)),
2635 memory_priority_hint: false,
2636 sequential_access_hint: false,
2637 };
2638
2639 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2640 assert_eq!(err_new.reason, err::DIR.reason);
2641
2642 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2643 assert_eq!(err_open.reason, err::DIR.reason);
2644
2645 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2646 assert_eq!(err_open_or_create.reason, err::DIR.reason);
2647 }
2648
2649 #[test]
2650 fn err_path_when_not_a_dir() {
2651 let dir = tempfile::tempdir().unwrap();
2652 let file_path = dir.path().join("regular_file");
2653 std::fs::write(&file_path, b"test").unwrap();
2654
2655 let cfg = MemMapCfg {
2656 module_id: MID,
2657 path: file_path,
2658 name: *b"tbfx",
2659 version: 1,
2660 initial_pages: INIT_PAGES,
2661 max_len: 0x10_0000,
2662 sync_interval: Some(Duration::from_millis(0x64)),
2663 memory_priority_hint: false,
2664 sequential_access_hint: false,
2665 };
2666
2667 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2668 assert_eq!(err_new.reason, err::DIR.reason);
2669
2670 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2671 assert_eq!(err_open.reason, err::DIR.reason);
2672
2673 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2674 assert_eq!(err_open_or_create.reason, err::DIR.reason);
2675 }
2676
2677 #[test]
2678 fn err_path_when_empty() {
2679 let cfg = MemMapCfg {
2680 module_id: MID,
2681 path: std::path::PathBuf::from(""),
2682 name: *b"tbfx",
2683 version: 1,
2684 initial_pages: INIT_PAGES,
2685 max_len: 0x10_0000,
2686 sync_interval: Some(Duration::from_millis(0x64)),
2687 memory_priority_hint: false,
2688 sequential_access_hint: false,
2689 };
2690
2691 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2692 assert_eq!(err_new.reason, err::DIR.reason);
2693
2694 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2695 assert_eq!(err_open.reason, err::DIR.reason);
2696
2697 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2698 assert_eq!(err_open_or_create.reason, err::DIR.reason);
2699 }
2700
2701 #[test]
2702 fn err_name_invalid() {
2703 let dir = tempfile::tempdir().unwrap();
2704 let invalid_tags = [*b"tb-x", *b"tb x", *b"tb\0x", [0xFF, 0x00, 0x01, 0x02]];
2705 for tag in invalid_tags {
2706 let cfg = MemMapCfg {
2707 module_id: MID,
2708 path: dir.path().to_path_buf(),
2709 name: tag,
2710 version: 1,
2711 initial_pages: INIT_PAGES,
2712 max_len: 0x10_0000,
2713 sync_interval: Some(Duration::from_millis(0x64)),
2714 memory_priority_hint: false,
2715 sequential_access_hint: false,
2716 };
2717
2718 let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2719 assert_eq!(err_new.reason, err::TAG.reason);
2720
2721 let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2722 assert_eq!(err_open.reason, err::TAG.reason);
2723
2724 let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2725 assert_eq!(err_open_or_create.reason, err::TAG.reason);
2726 }
2727 }
2728
2729 #[test]
2730 fn ok_delete() {
2731 let (_dir, cfg) = tmp_cfg();
2732 let path = cfg.versioned_path();
2733 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2734 assert!(path.exists());
2735 assert!(MemMap::<u64>::exists(&cfg));
2736 assert!(mmap.delete().is_ok());
2737 assert!(!path.exists());
2738 assert!(!MemMap::<u64>::exists(&cfg));
2739 }
2740
2741 #[test]
2742 fn ok_versioned_path_resolution() {
2743 let dir = std::path::PathBuf::from("/var/data");
2744 let cfg = MemMapCfg {
2745 module_id: MID,
2746 path: dir.clone(),
2747 name: *b"tbfx",
2748 version: 0,
2749 initial_pages: 1,
2750 max_len: 0x10_0000,
2751 sync_interval: Some(Duration::from_millis(0x64)),
2752 memory_priority_hint: false,
2753 sequential_access_hint: false,
2754 };
2755 assert_eq!(cfg.versioned_path(), dir.join("tbfx_0"));
2756
2757 let cfg = MemMapCfg {
2758 module_id: MID,
2759 path: dir.clone(),
2760 name: *b"DATA",
2761 version: 5,
2762 initial_pages: 1,
2763 max_len: 0x10_0000,
2764 sync_interval: Some(Duration::from_millis(0x64)),
2765 memory_priority_hint: false,
2766 sequential_access_hint: false,
2767 };
2768 assert_eq!(cfg.versioned_path(), dir.join("DATA_5"));
2769 }
2770
2771 #[test]
2772 fn ok_versioned_path_and_multiversion() {
2773 let dir = tempfile::tempdir().unwrap();
2774
2775 let cfg_v1 = MemMapCfg {
2776 module_id: MID,
2777 path: dir.path().to_path_buf(),
2778 name: *b"tbfx",
2779 version: 1,
2780 initial_pages: 1,
2781 max_len: 0x10_0000,
2782 sync_interval: Some(Duration::from_millis(0x64)),
2783 memory_priority_hint: false,
2784 sequential_access_hint: false,
2785 };
2786 assert_eq!(cfg_v1.versioned_path(), dir.path().join("tbfx_1"));
2787
2788 let cfg_v2 = MemMapCfg {
2789 module_id: MID,
2790 path: dir.path().to_path_buf(),
2791 name: *b"tbfx",
2792 version: 2,
2793 initial_pages: 1,
2794 max_len: 0x10_0000,
2795 sync_interval: Some(Duration::from_millis(0x64)),
2796 memory_priority_hint: false,
2797 sequential_access_hint: false,
2798 };
2799 assert_eq!(cfg_v2.versioned_path(), dir.path().join("tbfx_2"));
2800
2801 assert!(!MemMap::<u64>::exists(&cfg_v1));
2802 assert!(!MemMap::<ValidT>::exists(&cfg_v2));
2803
2804 let mmap_v1 = MemMap::<u64>::new(cfg_v1.clone()).unwrap();
2806 let mmap_v2 = MemMap::<ValidT>::new(cfg_v2.clone()).unwrap();
2807 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2808
2809 assert!(MemMap::<u64>::exists(&cfg_v1));
2810 assert!(MemMap::<ValidT>::exists(&cfg_v2));
2811 assert!(cfg_v1.versioned_path().exists());
2812 assert!(cfg_v2.versioned_path().exists());
2813 assert_eq!(mmap_v1.length(), page_size);
2814 assert_eq!(mmap_v2.length(), page_size);
2815
2816 assert!(mmap_v1.delete().is_ok());
2817 assert!(!MemMap::<u64>::exists(&cfg_v1));
2818 assert!(MemMap::<ValidT>::exists(&cfg_v2));
2819 assert!(!cfg_v1.versioned_path().exists());
2820 assert!(cfg_v2.versioned_path().exists());
2821
2822 assert!(mmap_v2.delete().is_ok());
2823 assert!(!MemMap::<ValidT>::exists(&cfg_v2));
2824 assert!(!cfg_v2.versioned_path().exists());
2825 }
2826
2827 #[test]
2828 fn ok_apply_memory_priority_hint() {
2829 let (_dir, cfg) = tmp_cfg();
2830 let mmap = MemMap::<u64>::new(cfg).unwrap();
2831 assert!(mmap.apply_memory_priority_hint().is_ok());
2832 }
2833
2834 #[test]
2835 fn ok_memory_priority_hint_initialization() {
2836 let (_dir, mut cfg) = tmp_cfg();
2837 cfg.memory_priority_hint = true;
2838
2839 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2840 assert_eq!(mmap.length(), MemMap::<u64>::page_size().unwrap().as_usize() * INIT_PAGES);
2841 drop(mmap);
2842
2843 let opened = MemMap::<u64>::open(cfg.clone()).unwrap();
2844 assert_eq!(opened.length(), MemMap::<u64>::page_size().unwrap().as_usize() * INIT_PAGES);
2845 drop(opened);
2846
2847 let open_or_created = MemMap::<u64>::open_or_create(cfg).unwrap();
2848 assert_eq!(
2849 open_or_created.length(),
2850 MemMap::<u64>::page_size().unwrap().as_usize() * INIT_PAGES
2851 );
2852 assert!(open_or_created.delete().is_ok());
2853 }
2854
2855 #[test]
2856 fn ok_sequential_access_hint_initialization() {
2857 let (_dir, mut cfg) = tmp_cfg();
2858 cfg.sequential_access_hint = true;
2859
2860 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2861 assert_eq!(mmap.length(), MemMap::<u64>::page_size().unwrap().as_usize() * INIT_PAGES);
2862 assert!(mmap.cfg().sequential_access_hint);
2863 drop(mmap);
2864
2865 let opened = MemMap::<u64>::open(cfg.clone()).unwrap();
2866 assert_eq!(opened.length(), MemMap::<u64>::page_size().unwrap().as_usize() * INIT_PAGES);
2867 drop(opened);
2868
2869 let open_or_created = MemMap::<u64>::open_or_create(cfg).unwrap();
2870 assert_eq!(
2871 open_or_created.length(),
2872 MemMap::<u64>::page_size().unwrap().as_usize() * INIT_PAGES
2873 );
2874 assert!(open_or_created.delete().is_ok());
2875 }
2876
2877 #[test]
2878 fn ok_advise_sequential_and_prefetch_api() {
2879 let (_dir, cfg) = tmp_cfg();
2880 let mmap = MemMap::<u64>::new(cfg).unwrap();
2881
2882 assert!(mmap.advise_sequential().is_ok());
2883 assert!(mmap.advise_sequential_range(0, 0).is_ok());
2884 assert!(mmap.prefetch(0, 0).is_ok());
2885
2886 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2887 assert!(mmap.advise_sequential_range(0, page_size).is_ok());
2888 assert!(mmap.prefetch(0, page_size).is_ok());
2889
2890 let total_len = mmap.length();
2891 let err_seq = mmap.advise_sequential_range(0, total_len + 1).unwrap_err();
2892 assert_eq!(err_seq.reason, err::HCF.reason);
2893
2894 let err_pref = mmap.prefetch(total_len, 1).unwrap_err();
2895 assert_eq!(err_pref.reason, err::HCF.reason);
2896 assert!(mmap.delete().is_ok());
2897 }
2898
2899 #[test]
2900 fn ok_resident_bytes_api() {
2901 let (_dir, cfg) = tmp_cfg();
2902 let mmap = MemMap::<u64>::new(cfg).unwrap();
2903 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2904
2905 mmap.write(0, |v| *v = 0x1234).unwrap();
2906 let resident = mmap.resident_bytes().unwrap();
2907 assert!(resident >= page_size);
2908 assert_eq!(resident % page_size, 0);
2909 assert!(resident <= mmap.length());
2910 assert!(mmap.delete().is_ok());
2911 }
2912
2913 #[test]
2914 fn ok_sequential_hint_preserved_across_growth() {
2915 let (_dir, mut cfg) = tmp_cfg();
2916 cfg.sequential_access_hint = true;
2917
2918 let mmap = MemMap::<u64>::new(cfg).unwrap();
2919 let initial_len = mmap.length();
2920 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
2921
2922 assert!(mmap.grow(1).is_ok());
2923 assert_eq!(mmap.length(), initial_len + page_size);
2924 assert!(mmap.delete().is_ok());
2925 }
2926
2927 #[test]
2928 fn ok_write_read_single() {
2929 let (_dir, cfg) = tmp_cfg();
2930 let mmap = MemMap::<u64>::new(cfg).unwrap();
2931
2932 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
2933 let (ticket, _) = mmap.write(0, |v| *v = 0xDEAD_C0DE_CAFE_BABE).unwrap();
2934 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xDEAD_C0DE_CAFE_BABE);
2935 assert_eq!(ticket.epoch(), 1);
2936 }
2937
2938 #[test]
2939 fn ok_write_read_multiple_indices() {
2940 let (_dir, cfg) = tmp_cfg();
2941 let mmap = MemMap::<u64>::new(cfg).unwrap();
2942
2943 for i in 0..INIT_T {
2944 let val = (i as u64 + 1).wrapping_mul(0x1111_2222_3333_4444);
2945 mmap.write(i, |v| *v = val).unwrap();
2946 }
2947
2948 for i in 0..INIT_T {
2949 let expected = (i as u64 + 1).wrapping_mul(0x1111_2222_3333_4444);
2950 assert_eq!(mmap.read(i, |v| *v).unwrap(), expected);
2951 }
2952 }
2953
2954 #[test]
2955 fn ok_write_read_struct() {
2956 let (_dir, cfg) = tmp_cfg();
2957 let mmap = MemMap::<ValidT>::new(cfg).unwrap();
2958
2959 let val0 = ValidT { a: 0x0A, b: 0x14 };
2960 let val1 = ValidT { a: 0x64, b: 0xC8 };
2961
2962 mmap.write(0, |v| *v = val0).unwrap();
2963 mmap.write(1, |v| *v = val1).unwrap();
2964
2965 let r0 = mmap.read(0, |v| v.clone()).unwrap();
2966 let r1 = mmap.read(1, |v| v.clone()).unwrap();
2967
2968 assert_eq!(r0.a, 0x0A);
2969 assert_eq!(r0.b, 0x14);
2970 assert_eq!(r1.a, 0x64);
2971 assert_eq!(r1.b, 0xC8);
2972 }
2973
2974 #[test]
2975 fn ok_read_callback_zero_copy_projection() {
2976 let (_dir, cfg) = tmp_cfg();
2977 let mmap = MemMap::<ValidT>::new(cfg).unwrap();
2978
2979 let val = ValidT { a: 0x1234, b: 0x5678 };
2980 mmap.write(0, |v| *v = val).unwrap();
2981
2982 let a_val = mmap.read(0, |v| v.a).unwrap();
2983 assert_eq!(a_val, 0x1234);
2984
2985 let sum = mmap.read(0, |v| v.a + v.b).unwrap();
2986 assert_eq!(sum, 0x1234 + 0x5678);
2987
2988 let is_match = mmap.read(0, |v| v.b == 0x5678).unwrap();
2989 assert!(is_match);
2990 }
2991
2992 #[test]
2993 fn ok_read_callback_unwind_releases_lock() {
2994 let (_dir, cfg) = tmp_cfg();
2995 let mmap = MemMap::<u64>::new(cfg).unwrap();
2996 mmap.write(0, |v| *v = 0x42).unwrap();
2997
2998 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2999 let _ = mmap.read(0, |_| panic!("simulated panic in read callback"));
3000 }));
3001 assert!(result.is_err());
3002
3003 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x42);
3004 assert!(mmap.write(0, |v| *v = 0x99).is_ok());
3005 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x99);
3006 }
3007
3008 #[test]
3009 fn ok_write_callback_in_place_mutation() {
3010 let (_dir, cfg) = tmp_cfg();
3011 let mmap = MemMap::<ValidT>::new(cfg).unwrap();
3012
3013 mmap.write(0, |v| {
3014 v.a = 0x1111;
3015 v.b = 0x2222;
3016 })
3017 .unwrap();
3018
3019 mmap.write(0, |v| {
3020 v.b = 0x9999;
3021 })
3022 .unwrap();
3023
3024 let val = mmap.read(0, |v| v.clone()).unwrap();
3025 assert_eq!(val.a, 0x1111);
3026 assert_eq!(val.b, 0x9999);
3027 }
3028
3029 #[test]
3030 fn ok_write_callback_returns_derived_value() {
3031 let (_dir, cfg) = tmp_cfg();
3032 let mmap = MemMap::<u64>::new(cfg).unwrap();
3033
3034 let (ticket, old_prev) = mmap
3035 .write(0, |v| {
3036 let prev = *v;
3037 *v = 42;
3038 prev
3039 })
3040 .unwrap();
3041
3042 assert_eq!(old_prev, 0);
3043 assert_eq!(ticket.epoch(), 1);
3044 assert_eq!(mmap.read(0, |v| *v).unwrap(), 42);
3045 }
3046
3047 #[test]
3048 fn ok_write_callback_unwind_releases_lock() {
3049 let (_dir, cfg) = tmp_cfg();
3050 let mmap = MemMap::<u64>::new(cfg).unwrap();
3051 let epoch_before = mmap.current_epoch();
3052
3053 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3054 let _ = mmap.write(0, |v| {
3055 *v = 0xBAD;
3056 panic!("intentional panic inside write callback");
3057 });
3058 }));
3059 assert!(result.is_err());
3060
3061 assert_eq!(mmap.current_epoch(), epoch_before);
3062 assert!(mmap.write(0, |v| *v = 0x600D).is_ok());
3063 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x600D);
3064 }
3065
3066 #[test]
3067 fn ok_concurrent_writes_and_reads_single_slot() {
3068 let (_dir, mut cfg) = tmp_cfg();
3069 cfg.initial_pages = 1;
3070 let mmap = Arc::new(MemMap::<ValidT>::new(cfg).unwrap());
3071
3072 let barrier = Arc::new(std::sync::Barrier::new(4));
3073 let mut handles = Vec::new();
3074
3075 for t in 0..2 {
3076 let m = Arc::clone(&mmap);
3077 let b = Arc::clone(&barrier);
3078 handles.push(std::thread::spawn(move || {
3079 b.wait();
3080 for i in 0..0xC8 {
3081 let val = ((t + 1) as u64) << 0x20 | (i as u64);
3082 let _ = m.write(0, |node| {
3083 node.a = val;
3084 node.b = val;
3085 });
3086 }
3087 }));
3088 }
3089
3090 for _ in 0..2 {
3091 let m = Arc::clone(&mmap);
3092 let b = Arc::clone(&barrier);
3093 handles.push(std::thread::spawn(move || {
3094 b.wait();
3095 for _ in 0..0x190 {
3096 let _ = m.read(0, |node| {
3097 assert_eq!(node.a, node.b);
3098 });
3099 }
3100 }));
3101 }
3102
3103 for h in handles {
3104 h.join().unwrap();
3105 }
3106
3107 mmap.read(0, |node| {
3108 assert_eq!(node.a, node.b);
3109 })
3110 .unwrap();
3111 }
3112
3113 #[test]
3114 fn err_write_read_out_of_bounds() {
3115 let (_dir, cfg) = tmp_cfg();
3116 let mmap = MemMap::<u64>::new(cfg).unwrap();
3117 let slots = mmap.length() / core::mem::size_of::<u64>();
3118
3119 let err_r = mmap.read(slots, |_| ()).unwrap_err();
3120 assert_eq!(err_r.reason, err::HCF.reason);
3121
3122 let err_w = mmap.write(slots, |_| ()).unwrap_err();
3123 assert_eq!(err_w.reason, err::HCF.reason);
3124
3125 let err_r_large = mmap.read(slots + 0x64, |_| ()).unwrap_err();
3126 assert_eq!(err_r_large.reason, err::HCF.reason);
3127
3128 let err_w_large = mmap.write(slots + 0x64, |_| ()).unwrap_err();
3129 assert_eq!(err_w_large.reason, err::HCF.reason);
3130 }
3131
3132 #[test]
3133 fn err_write_read_overflow_index() {
3134 let (_dir, cfg) = tmp_cfg();
3135 let mmap = MemMap::<u64>::new(cfg).unwrap();
3136
3137 let err_r = mmap.read(usize::MAX, |_| ()).unwrap_err();
3138 assert_eq!(err_r.reason, err::INV.reason);
3139
3140 let err_w = mmap.write(usize::MAX, |_| ()).unwrap_err();
3141 assert_eq!(err_w.reason, err::INV.reason);
3142 }
3143
3144 mod read_exact {
3145 use super::*;
3146
3147 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
3148 #[repr(C)]
3149 struct TestNode {
3150 id: u64,
3151 flags: u64,
3152 keys: [u64; 2],
3153 }
3154
3155 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
3156 #[repr(C)]
3157 struct TestBadSizeTarget([u8; 12]);
3158
3159 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
3160 #[repr(C, align(16))]
3161 struct TestOveralignedTarget {
3162 _a: u64,
3163 _b: u64,
3164 }
3165
3166 #[test]
3167 fn ok_read_exact_single_and_multi_slot() {
3168 let (_dir, cfg) = tmp_cfg();
3169 let mmap = MemMap::<u64>::new(cfg).unwrap();
3170
3171 mmap.write(0, |v| *v = 0x1111).unwrap();
3172 mmap.write(1, |v| *v = 0x2222).unwrap();
3173 mmap.write(2, |v| *v = 0x3333).unwrap();
3174 mmap.write(3, |v| *v = 0x4444).unwrap();
3175
3176 let single = mmap.read_exact(0, |val: &u64| *val).unwrap();
3177 assert_eq!(single, 0x1111);
3178
3179 let node = mmap.read_exact(0, |n: &TestNode| *n).unwrap();
3180 assert_eq!(node, TestNode { id: 0x1111, flags: 0x2222, keys: [0x3333, 0x4444] });
3181
3182 let keys = mmap.read_exact(2, |k: &[u64; 2]| *k).unwrap();
3183 assert_eq!(keys, [0x3333, 0x4444]);
3184
3185 let flag_is_set = mmap.read_exact(0, |n: &TestNode| n.flags == 0x2222).unwrap();
3186 assert!(flag_is_set);
3187 }
3188
3189 #[test]
3190 fn err_read_exact_invariants() {
3191 let (_dir, cfg) = tmp_cfg();
3192 let mmap = MemMap::<u64>::new(cfg).unwrap();
3193
3194 let err_zro = mmap.read_exact::<(), _, _>(0, |_| ()).unwrap_err();
3195 assert_eq!(err_zro.reason, err::ZRO.reason);
3196
3197 let err_sze = mmap.read_exact::<TestBadSizeTarget, _, _>(0, |_| ()).unwrap_err();
3198 assert_eq!(err_sze.reason, err::SZE.reason);
3199
3200 let err_aln = mmap.read_exact::<TestOveralignedTarget, _, _>(0, |_| ()).unwrap_err();
3201 assert_eq!(err_aln.reason, err::ALN.reason);
3202
3203 let err_inv = mmap.read_exact::<u64, _, _>(usize::MAX, |_| ()).unwrap_err();
3204 assert_eq!(err_inv.reason, err::INV.reason);
3205
3206 let err_inv_add =
3207 mmap.read_exact::<TestNode, _, _>(usize::MAX - 1, |_| ()).unwrap_err();
3208 assert_eq!(err_inv_add.reason, err::INV.reason);
3209
3210 let slots = mmap.length() / core::mem::size_of::<u64>();
3211 let err_hcf_start = mmap.read_exact::<u64, _, _>(slots, |_| ()).unwrap_err();
3212 assert_eq!(err_hcf_start.reason, err::HCF.reason);
3213
3214 let err_hcf_end = mmap.read_exact::<TestNode, _, _>(slots - 2, |_| ()).unwrap_err();
3215 assert_eq!(err_hcf_end.reason, err::HCF.reason);
3216 }
3217
3218 #[test]
3219 fn ok_read_exact_concurrent_readers() {
3220 let (_dir, cfg) = tmp_cfg();
3221 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3222
3223 mmap.write(0, |v| *v = 0xCAFE).unwrap();
3224 mmap.write(1, |v| *v = 0xBABE).unwrap();
3225 mmap.write(2, |v| *v = 0x1234).unwrap();
3226 mmap.write(3, |v| *v = 0x5678).unwrap();
3227
3228 let barrier = Arc::new(std::sync::Barrier::new(4));
3229 let mut handles = Vec::new();
3230
3231 for _ in 0..4 {
3232 let mmap_clone = Arc::clone(&mmap);
3233 let barrier_clone = Arc::clone(&barrier);
3234 handles.push(std::thread::spawn(move || {
3235 barrier_clone.wait();
3236 for _ in 0..0x1F4 {
3237 let id = mmap_clone.read_exact(0, |n: &TestNode| n.id).unwrap();
3238 assert_eq!(id, 0xCAFE);
3239 let k1 = mmap_clone.read_exact(0, |n: &TestNode| n.keys[1]).unwrap();
3240 assert_eq!(k1, 0x5678);
3241 }
3242 }));
3243 }
3244
3245 for h in handles {
3246 h.join().unwrap();
3247 }
3248 }
3249
3250 #[test]
3251 fn ok_read_exact_concurrent_writers_isolation() {
3252 let (_dir, cfg) = tmp_cfg();
3253 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3254
3255 let barrier = Arc::new(std::sync::Barrier::new(5));
3256 let mut handles = Vec::new();
3257
3258 {
3259 let mmap_clone = Arc::clone(&mmap);
3260 let barrier_clone = Arc::clone(&barrier);
3261 handles.push(std::thread::spawn(move || {
3262 barrier_clone.wait();
3263 for i in 0..0x1F4 {
3264 let v = i as u64;
3265 let _ = mmap_clone.write(0, |val| *val = v).unwrap();
3266 let _ = mmap_clone.write(1, |val| *val = v).unwrap();
3267 }
3268 }));
3269 }
3270
3271 for _ in 0..4 {
3272 let mmap_clone = Arc::clone(&mmap);
3273 let barrier_clone = Arc::clone(&barrier);
3274 handles.push(std::thread::spawn(move || {
3275 barrier_clone.wait();
3276 for _ in 0..0x1F4 {
3277 mmap_clone
3278 .read_exact(0, |pair: &[u64; 2]| {
3279 assert!(pair[0] <= 0x1F4 && pair[1] <= 0x1F4);
3280 })
3281 .unwrap();
3282 }
3283 }));
3284 }
3285
3286 for h in handles {
3287 h.join().unwrap();
3288 }
3289 }
3290
3291 #[test]
3292 fn ok_read_exact_unwind_releases_locks() {
3293 let (_dir, cfg) = tmp_cfg();
3294 let mmap = MemMap::<u64>::new(cfg).unwrap();
3295
3296 mmap.write(0, |v| *v = 0x1111).unwrap();
3297 mmap.write(1, |v| *v = 0x2222).unwrap();
3298
3299 let panic_res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3300 let _ = mmap.read_exact(0, |_pair: &[u64; 2]| {
3301 panic!("intentional panic inside read_exact callback");
3302 });
3303 }));
3304 assert!(panic_res.is_err());
3305
3306 let (ticket, _) = mmap.write(0, |v| *v = 0x9999).unwrap();
3307 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999);
3308 assert_eq!(ticket.epoch(), 3);
3309
3310 let (ticket2, _) = mmap.write(1, |v| *v = 0x8888).unwrap();
3311 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x8888);
3312 assert_eq!(ticket2.epoch(), 4);
3313 }
3314 }
3315
3316 mod write_exact {
3317 use super::*;
3318
3319 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
3320 #[repr(C)]
3321 struct TestNode {
3322 id: u64,
3323 flags: u64,
3324 keys: [u64; 2],
3325 }
3326
3327 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
3328 #[repr(C)]
3329 struct TestBadSizeTarget([u8; 12]);
3330
3331 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
3332 #[repr(C, align(16))]
3333 struct TestOveralignedTarget {
3334 _a: u64,
3335 _b: u64,
3336 }
3337
3338 #[test]
3339 fn ok_write_exact_single_and_multi_slot() {
3340 let (_dir, cfg) = tmp_cfg();
3341 let mmap = MemMap::<u64>::new(cfg).unwrap();
3342
3343 let (ticket_single, res) = mmap
3344 .write_exact(0, |val: &mut u64| {
3345 *val = 0xAAAA;
3346 42u32
3347 })
3348 .unwrap();
3349 assert_eq!(res, 42);
3350 assert_eq!(ticket_single.epoch(), 1);
3351 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xAAAA);
3352
3353 let (ticket_multi, _) = mmap
3354 .write_exact(0, |node: &mut TestNode| {
3355 node.id = 0x1111;
3356 node.flags = 0x2222;
3357 node.keys = [0x3333, 0x4444];
3358 })
3359 .unwrap();
3360 assert_eq!(ticket_multi.epoch(), 2);
3361 assert_eq!(
3362 mmap.read_exact(0, |n: &TestNode| *n).unwrap(),
3363 TestNode { id: 0x1111, flags: 0x2222, keys: [0x3333, 0x4444] }
3364 );
3365
3366 let (ticket_partial, _) = mmap
3367 .write_exact(2, |keys: &mut [u64; 2]| {
3368 keys[0] = 0x5555;
3369 keys[1] = 0x6666;
3370 })
3371 .unwrap();
3372 assert_eq!(ticket_partial.epoch(), 3);
3373 assert_eq!(mmap.read(2, |v| *v).unwrap(), 0x5555);
3374 assert_eq!(mmap.read(3, |v| *v).unwrap(), 0x6666);
3375 }
3376
3377 #[test]
3378 fn err_write_exact_invariants() {
3379 let (_dir, cfg) = tmp_cfg();
3380 let mmap = MemMap::<u64>::new(cfg).unwrap();
3381
3382 let err_zro = mmap.write_exact::<(), _, _>(0, |_| ()).unwrap_err();
3383 assert_eq!(err_zro.reason, err::ZRO.reason);
3384
3385 let err_sze = mmap.write_exact::<TestBadSizeTarget, _, _>(0, |_| ()).unwrap_err();
3386 assert_eq!(err_sze.reason, err::SZE.reason);
3387
3388 let err_aln = mmap.write_exact::<TestOveralignedTarget, _, _>(0, |_| ()).unwrap_err();
3389 assert_eq!(err_aln.reason, err::ALN.reason);
3390
3391 let err_inv = mmap.write_exact::<u64, _, _>(usize::MAX, |_| ()).unwrap_err();
3392 assert_eq!(err_inv.reason, err::INV.reason);
3393
3394 let err_inv_add =
3395 mmap.write_exact::<TestNode, _, _>(usize::MAX - 1, |_| ()).unwrap_err();
3396 assert_eq!(err_inv_add.reason, err::INV.reason);
3397
3398 let slots = mmap.length() / core::mem::size_of::<u64>();
3399 let err_hcf_start = mmap.write_exact::<u64, _, _>(slots, |_| ()).unwrap_err();
3400 assert_eq!(err_hcf_start.reason, err::HCF.reason);
3401
3402 let err_hcf_end = mmap.write_exact::<TestNode, _, _>(slots - 2, |_| ()).unwrap_err();
3403 assert_eq!(err_hcf_end.reason, err::HCF.reason);
3404 }
3405
3406 #[test]
3407 fn ok_write_exact_concurrent_writers() {
3408 let (_dir, cfg) = tmp_cfg();
3409 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3410
3411 let barrier = Arc::new(std::sync::Barrier::new(2));
3412 let mut handles = Vec::new();
3413
3414 {
3415 let mmap_clone = Arc::clone(&mmap);
3416 let barrier_clone = Arc::clone(&barrier);
3417 handles.push(std::thread::spawn(move || {
3418 barrier_clone.wait();
3419 for i in 0..0x1F4 {
3420 let _ = mmap_clone
3421 .write_exact(0, |node: &mut TestNode| {
3422 node.id = i as u64;
3423 node.flags = (i as u64) + 1;
3424 node.keys = [(i as u64) + 2, (i as u64) + 3];
3425 })
3426 .unwrap();
3427 }
3428 }));
3429 }
3430
3431 {
3432 let mmap_clone = Arc::clone(&mmap);
3433 let barrier_clone = Arc::clone(&barrier);
3434 handles.push(std::thread::spawn(move || {
3435 barrier_clone.wait();
3436 for i in 0..0x1F4 {
3437 let _ = mmap_clone
3438 .write_exact(4, |node: &mut TestNode| {
3439 node.id = (i as u64) * 10;
3440 node.flags = (i as u64) * 10 + 1;
3441 node.keys = [(i as u64) * 10 + 2, (i as u64) * 10 + 3];
3442 })
3443 .unwrap();
3444 }
3445 }));
3446 }
3447
3448 for h in handles {
3449 h.join().unwrap();
3450 }
3451
3452 let node0 = mmap.read_exact(0, |n: &TestNode| *n).unwrap();
3453 assert_eq!(node0.id, 0x1F3);
3454 assert_eq!(node0.flags, 0x1F4);
3455 assert_eq!(node0.keys, [0x1F5, 0x1F6]);
3456
3457 let node1 = mmap.read_exact(4, |n: &TestNode| *n).unwrap();
3458 assert_eq!(node1.id, 0x1F3 * 10);
3459 assert_eq!(node1.flags, 0x1F3 * 10 + 1);
3460 assert_eq!(node1.keys, [0x1F3 * 10 + 2, 0x1F3 * 10 + 3]);
3461 }
3462
3463 #[test]
3464 fn ok_write_exact_concurrent_readers_writers_isolation() {
3465 let (_dir, cfg) = tmp_cfg();
3466 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3467
3468 let barrier = Arc::new(std::sync::Barrier::new(5));
3469 let mut handles = Vec::new();
3470
3471 {
3472 let mmap_clone = Arc::clone(&mmap);
3473 let barrier_clone = Arc::clone(&barrier);
3474 handles.push(std::thread::spawn(move || {
3475 barrier_clone.wait();
3476 for i in 0..0x1F4 {
3477 let v = i as u64;
3478 let _ = mmap_clone
3479 .write_exact(0, |pair: &mut [u64; 2]| {
3480 pair[0] = v;
3481 pair[1] = v;
3482 })
3483 .unwrap();
3484 }
3485 }));
3486 }
3487
3488 for _ in 0..4 {
3489 let mmap_clone = Arc::clone(&mmap);
3490 let barrier_clone = Arc::clone(&barrier);
3491 handles.push(std::thread::spawn(move || {
3492 barrier_clone.wait();
3493 for _ in 0..0x1F4 {
3494 mmap_clone
3495 .read_exact(0, |pair: &[u64; 2]| {
3496 assert_eq!(pair[0], pair[1]);
3497 })
3498 .unwrap();
3499 }
3500 }));
3501 }
3502
3503 for h in handles {
3504 h.join().unwrap();
3505 }
3506 }
3507
3508 #[test]
3509 fn ok_write_exact_unwind_releases_locks() {
3510 let (_dir, cfg) = tmp_cfg();
3511 let mmap = MemMap::<u64>::new(cfg).unwrap();
3512
3513 let panic_res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3514 let _ = mmap.write_exact(0, |_pair: &mut [u64; 2]| {
3515 panic!("intentional panic inside write_exact callback");
3516 });
3517 }));
3518 assert!(panic_res.is_err());
3519
3520 let (ticket, _) = mmap.write(0, |v| *v = 0x9999).unwrap();
3521 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999);
3522 assert_eq!(ticket.epoch(), 1);
3523
3524 let (ticket2, _) = mmap.write(1, |v| *v = 0x8888).unwrap();
3525 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x8888);
3526 assert_eq!(ticket2.epoch(), 2);
3527 }
3528
3529 #[test]
3530 fn ok_write_exact_durability() {
3531 let (_dir, cfg) = tmp_cfg();
3532 {
3533 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3534 let (ticket, _) = mmap
3535 .write_exact(0, |node: &mut TestNode| {
3536 node.id = 0xDEAD;
3537 node.flags = 0xC0DE;
3538 node.keys = [0xCAFE, 0xBABE];
3539 })
3540 .unwrap();
3541 mmap.inner.sync_internal().unwrap();
3542 assert_eq!(ticket.epoch(), 1);
3543 }
3544
3545 let reopened = MemMap::<u64>::open(cfg).unwrap();
3546 assert_eq!(
3547 reopened.read_exact(0, |n: &TestNode| *n).unwrap(),
3548 TestNode { id: 0xDEAD, flags: 0xC0DE, keys: [0xCAFE, 0xBABE] }
3549 );
3550 }
3551 }
3552
3553 #[test]
3554 fn ok_durability_across_reopen() {
3555 let (_dir, cfg) = tmp_cfg();
3556 {
3557 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3558 mmap.write(1, |v| *v = 0xCAFE_BABE_1234_5678).unwrap();
3559 mmap.inner.sync_internal().unwrap();
3560 }
3561
3562 let reopened = MemMap::<u64>::open(cfg).unwrap();
3563 assert_eq!(reopened.read(1, |v| *v).unwrap(), 0xCAFE_BABE_1234_5678);
3564 }
3565
3566 #[test]
3567 fn ok_background_sync_advances_epoch() {
3568 let (_dir, mut cfg) = tmp_cfg();
3569 cfg.sync_interval = Some(Duration::from_millis(0x19));
3570
3571 let mmap = MemMap::<u64>::new(cfg).unwrap();
3572 let (ticket, _) = mmap.write(0, |v| *v = 0x1234).unwrap();
3573 let epoch = ticket.wait().unwrap();
3574
3575 assert_eq!(epoch, ticket.epoch());
3576 assert!(ticket.is_durable());
3577 assert!(mmap.durable_epoch() >= ticket.epoch());
3578 }
3579
3580 #[test]
3581 fn ok_ticket_force_wakes_bg_worker() {
3582 let (_dir, mut cfg) = tmp_cfg();
3583 cfg.sync_interval = Some(Duration::from_secs(0x0A));
3585
3586 let mmap = MemMap::<u64>::new(cfg).unwrap();
3587 let (ticket, _) = mmap.write(0, |v| *v = 0xABCD).unwrap();
3588 assert!(!ticket.is_durable());
3589
3590 let start = std::time::Instant::now();
3591 let epoch = ticket.force().unwrap();
3592 let elapsed = start.elapsed();
3593
3594 assert_eq!(epoch, ticket.epoch());
3595 assert!(ticket.is_durable());
3596 assert!(elapsed < Duration::from_secs(2));
3597 }
3598
3599 #[test]
3600 fn ok_drop_persists_unflushed_writes() {
3601 let (_dir, mut cfg) = tmp_cfg();
3602 cfg.sync_interval = Some(Duration::from_secs(0x1E));
3603
3604 {
3605 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3606 let (_ticket, _) = mmap.write(0, |v| *v = 0xC0DE_CAFE).unwrap();
3607 drop(mmap);
3608 }
3609
3610 let reopened = MemMap::<u64>::open(cfg).unwrap();
3611 assert_eq!(reopened.read(0, |v| *v).unwrap(), 0xC0DE_CAFE);
3612 }
3613
3614 #[test]
3615 fn ok_concurrent_slot_locking() {
3616 let (_dir, mut cfg) = tmp_cfg();
3617 cfg.sync_interval = Some(Duration::from_secs(0x3C));
3618
3619 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3620 let barrier = Arc::new(std::sync::Barrier::new(8));
3621 let mut handles = Vec::new();
3622
3623 for t in 0..4 {
3625 let mmap_clone = Arc::clone(&mmap);
3626 let barrier_clone = Arc::clone(&barrier);
3627 handles.push(std::thread::spawn(move || {
3628 barrier_clone.wait();
3629 for i in 0..0x1F4 {
3630 let val = ((t + 1) as u64) << 0x20 | (i as u64);
3631 let _ = mmap_clone.write(0, |slot| *slot = val).unwrap();
3632 }
3633 }));
3634 }
3635
3636 for _ in 0..4 {
3638 let mmap_clone = Arc::clone(&mmap);
3639 let barrier_clone = Arc::clone(&barrier);
3640 handles.push(std::thread::spawn(move || {
3641 barrier_clone.wait();
3642 for _ in 0..0x3E8 {
3643 let val = mmap_clone.read(0, |v| *v).unwrap();
3644 let thread_id = (val >> 0x20) as usize;
3645 assert!(thread_id <= 4, "read value must be valid without torn reads");
3646 }
3647 }));
3648 }
3649
3650 for h in handles {
3651 h.join().unwrap();
3652 }
3653
3654 let barrier2 = Arc::new(std::sync::Barrier::new(4));
3656 let mut read_handles = Vec::new();
3657 for _ in 0..4 {
3658 let mmap_clone = Arc::clone(&mmap);
3659 let barrier_clone = Arc::clone(&barrier2);
3660 read_handles.push(std::thread::spawn(move || {
3661 barrier_clone.wait();
3662 for _ in 0..0x1F4 {
3663 mmap_clone.read(0, |_| ()).unwrap();
3664 }
3665 }));
3666 }
3667
3668 for h in read_handles {
3669 h.join().unwrap();
3670 }
3671 }
3672
3673 #[test]
3674 fn err_concurrent_new_same_path() {
3675 let (_dir, cfg) = tmp_cfg();
3676 let barrier = Arc::new(std::sync::Barrier::new(2));
3677
3678 let t1 = {
3679 let cfg = cfg.clone();
3680 let b = Arc::clone(&barrier);
3681 std::thread::spawn(move || {
3682 b.wait();
3683 MemMap::<u64>::new(cfg)
3684 })
3685 };
3686
3687 let t2 = {
3688 let b = Arc::clone(&barrier);
3689 std::thread::spawn(move || {
3690 b.wait();
3691 MemMap::<u64>::new(cfg)
3692 })
3693 };
3694
3695 let r1 = t1.join().unwrap();
3696 let r2 = t2.join().unwrap();
3697
3698 match (r1, r2) {
3699 (Ok(m), Err(e)) | (Err(e), Ok(m)) => {
3700 assert_eq!(e.reason, err::EXS.reason);
3701 assert!(m.delete().is_ok());
3702 }
3703 (Ok(_), Ok(_)) => {
3704 panic!("Concurrent MemMap::new on identical path must not both succeed");
3705 }
3706 (Err(e1), Err(e2)) => {
3707 panic!("Unexpected error on both threads: {:?}, {:?}", e1, e2);
3708 }
3709 }
3710 }
3711
3712 #[test]
3713 fn err_concurrent_open_same_path() {
3714 let (_dir, cfg) = tmp_cfg();
3715 {
3716 let m = MemMap::<u64>::new(cfg.clone()).unwrap();
3717 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
3718 assert_eq!(m.length(), page_size * INIT_PAGES);
3719 }
3720
3721 let barrier = Arc::new(std::sync::Barrier::new(2));
3722
3723 let t1 = {
3724 let cfg = cfg.clone();
3725 let b = Arc::clone(&barrier);
3726 std::thread::spawn(move || {
3727 b.wait();
3728 MemMap::<u64>::open(cfg)
3729 })
3730 };
3731
3732 let t2 = {
3733 let b = Arc::clone(&barrier);
3734 std::thread::spawn(move || {
3735 b.wait();
3736 MemMap::<u64>::open(cfg)
3737 })
3738 };
3739
3740 let r1 = t1.join().unwrap();
3741 let r2 = t2.join().unwrap();
3742
3743 match (r1, r2) {
3744 (Ok(m), Err(e)) | (Err(e), Ok(m)) => {
3745 assert_eq!(e.domain, 0x08, "Error domain must be File (0x08)");
3746 assert_eq!(
3747 e.reason, 0x1C,
3748 "Concurrent open should yield LCK (0x1C) from kernel flock: {:?}",
3749 e
3750 );
3751 assert!(m.delete().is_ok());
3752 }
3753 (Ok(_), Ok(_)) => {
3754 panic!("Exclusive advisory lock violation: both concurrent open calls succeeded");
3755 }
3756 (Err(e1), Err(e2)) => {
3757 panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
3758 }
3759 }
3760 }
3761
3762 #[test]
3763 fn ok_reopen_after_drop_releases_lock() {
3764 let (_dir, cfg) = tmp_cfg();
3765 let m1 = MemMap::<u64>::new(cfg.clone()).unwrap();
3766 m1.write(0, |v| *v = 0x1122_3344).unwrap();
3767 drop(m1);
3768
3769 let m2 = MemMap::<u64>::open(cfg).unwrap();
3770 assert_eq!(m2.read(0, |v| *v).unwrap(), 0x1122_3344);
3771 assert!(m2.delete().is_ok());
3772 }
3773
3774 #[test]
3775 fn ok_delete_and_recreate_cycle() {
3776 let (_dir, cfg) = tmp_cfg();
3777 for i in 0..4 {
3778 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3779 let val = (i as u64 + 1) * 0xAA55;
3780 mmap.write(0, |v| *v = val).unwrap();
3781 assert_eq!(mmap.read(0, |v| *v).unwrap(), val);
3782 assert!(mmap.delete().is_ok());
3783 assert!(!cfg.versioned_path().exists());
3784 }
3785 }
3786
3787 #[test]
3788 fn ok_concurrent_distinct_slots_no_contention() {
3789 let (_dir, mut cfg) = tmp_cfg();
3790 const THREADS: usize = 8;
3791 cfg.sync_interval = Some(Duration::from_secs(0x3C));
3792
3793 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3794 let barrier = Arc::new(std::sync::Barrier::new(THREADS));
3795 let mut handles = Vec::with_capacity(THREADS);
3796
3797 for t in 0..THREADS {
3798 let mmap_clone = Arc::clone(&mmap);
3799 let barrier_clone = Arc::clone(&barrier);
3800 handles.push(std::thread::spawn(move || {
3801 barrier_clone.wait();
3802 for i in 0..0x3E8 {
3803 let val = (t as u64) << 0x20 | (i as u64);
3804 let _ = mmap_clone.write(t, |slot| *slot = val).unwrap();
3805 let read_val = mmap_clone.read(t, |v| *v).unwrap();
3806 assert_eq!(read_val >> 0x20, t as u64);
3807 }
3808 }));
3809 }
3810
3811 for h in handles {
3812 h.join().unwrap();
3813 }
3814
3815 for t in 0..THREADS {
3816 let val = mmap.read(t, |v| *v).unwrap();
3817 assert_eq!(val >> 0x20, t as u64);
3818 }
3819 }
3820
3821 #[test]
3822 fn ok_stress_interleaved_reads_writes() {
3823 let (_dir, mut cfg) = tmp_cfg();
3824 const NUM_SLOTS: usize = 0x10;
3825 const NUM_THREADS: usize = 8;
3826 cfg.sync_interval = Some(Duration::from_millis(0x32));
3827
3828 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3829 let barrier = Arc::new(std::sync::Barrier::new(NUM_THREADS));
3830 let mut handles = Vec::with_capacity(NUM_THREADS);
3831
3832 for t in 0..NUM_THREADS {
3833 let mmap_clone = Arc::clone(&mmap);
3834 let barrier_clone = Arc::clone(&barrier);
3835 handles.push(std::thread::spawn(move || {
3836 barrier_clone.wait();
3837 for i in 0..0x5DC {
3838 let slot = (t + i) % NUM_SLOTS;
3839 if i % 2 == 0 {
3840 let val = ((t + 1) as u64) << 0x20 | (i as u64);
3841 let _ = mmap_clone.write(slot, |s| *s = val).unwrap();
3842 } else {
3843 mmap_clone.read(slot, |_| ()).unwrap();
3844 }
3845 }
3846 }));
3847 }
3848
3849 for h in handles {
3850 h.join().unwrap();
3851 }
3852
3853 assert!(mmap.inner.sync_internal().is_ok());
3854 }
3855
3856 #[test]
3857 fn ok_concurrent_writers_with_background_sync() {
3858 let (_dir, mut cfg) = tmp_cfg();
3859 cfg.sync_interval = Some(Duration::from_millis(5));
3860
3861 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3862 let barrier = Arc::new(std::sync::Barrier::new(4));
3863 let mut handles = Vec::new();
3864
3865 for t in 0..4 {
3866 let mmap_clone = Arc::clone(&mmap);
3867 let barrier_clone = Arc::clone(&barrier);
3868 handles.push(std::thread::spawn(move || {
3869 barrier_clone.wait();
3870 let mut tickets = Vec::with_capacity(0x32);
3871 for i in 0..0x32 {
3872 let val = (t as u64) * 0x3E8 + i as u64;
3873 tickets.push(mmap_clone.write(t, |slot| *slot = val).unwrap().0);
3874 std::thread::sleep(Duration::from_millis(1));
3875 }
3876 for ticket in tickets {
3877 let _ = ticket.wait().unwrap();
3878 assert!(ticket.is_durable());
3879 }
3880 }));
3881 }
3882
3883 for h in handles {
3884 h.join().unwrap();
3885 }
3886
3887 assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
3888 }
3889
3890 #[test]
3891 fn ok_concurrent_ticket_force_stress() {
3892 let (_dir, mut cfg) = tmp_cfg();
3893 cfg.sync_interval = Some(Duration::from_secs(0x1E));
3894
3895 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3896 let barrier = Arc::new(std::sync::Barrier::new(4));
3897 let mut handles = Vec::new();
3898
3899 for t in 0..4 {
3900 let mmap_clone = Arc::clone(&mmap);
3901 let barrier_clone = Arc::clone(&barrier);
3902 handles.push(std::thread::spawn(move || {
3903 barrier_clone.wait();
3904 for i in 0..0x14 {
3905 let val = (t as u64) * 0x64 + i as u64;
3906 let (ticket, _) = mmap_clone.write(t, |slot| *slot = val).unwrap();
3907 let durable_epoch = ticket.force().unwrap();
3908 assert!(ticket.is_durable());
3909 assert!(durable_epoch >= ticket.epoch());
3910 }
3911 }));
3912 }
3913
3914 for h in handles {
3915 h.join().unwrap();
3916 }
3917 }
3918
3919 #[test]
3920 fn ok_exact_boundary_indices() {
3921 let (_dir, cfg) = tmp_cfg();
3922 let mmap = MemMap::<u64>::new(cfg).unwrap();
3923 let slots = mmap.length() / core::mem::size_of::<u64>();
3924
3925 assert!(mmap.write(slots - 1, |v| *v = 0xF00D).is_ok());
3926 assert_eq!(mmap.read(slots - 1, |v| *v).unwrap(), 0xF00D);
3927
3928 assert_eq!(mmap.write(slots, |_| ()).unwrap_err().reason, err::HCF.reason);
3929 assert_eq!(mmap.read(slots, |_| ()).unwrap_err().reason, err::HCF.reason);
3930 }
3931
3932 #[test]
3933 fn ok_large_slot_count_initialization() {
3934 let (_dir, mut cfg) = tmp_cfg();
3935 const LARGE_PAGES: usize = 0x14;
3936 cfg.initial_pages = LARGE_PAGES;
3937
3938 let mmap = MemMap::<u64>::new(cfg).unwrap();
3939 let page_size = mmap.inner.page_size.as_usize();
3940 assert_eq!(mmap.length(), LARGE_PAGES * page_size);
3941
3942 let large_count = mmap.length() / core::mem::size_of::<u64>();
3943 let mid = large_count / 2;
3944
3945 assert!(mmap.write(0, |v| *v = 1).is_ok());
3946 assert!(mmap.write(mid, |v| *v = mid as u64).is_ok());
3947 assert!(mmap.write(large_count - 1, |v| *v = (large_count - 1) as u64).is_ok());
3948
3949 assert_eq!(mmap.read(0, |v| *v).unwrap(), 1);
3950 assert_eq!(mmap.read(mid, |v| *v).unwrap(), mid as u64);
3951 assert_eq!(mmap.read(large_count - 1, |v| *v).unwrap(), (large_count - 1) as u64);
3952 }
3953
3954 #[test]
3955 fn ok_new_with_sync_interval_none() {
3956 let (_dir, mut cfg) = tmp_cfg();
3957 cfg.sync_interval = None;
3958
3959 let mmap = MemMap::<u64>::new(cfg).unwrap();
3960 assert!(mmap.bg_thread.is_none());
3961
3962 let (ticket, _) = mmap.write(0, |v| *v = 0x55AA).unwrap();
3963 assert!(!ticket.is_durable());
3964
3965 mmap.sync().unwrap();
3966 assert!(ticket.is_durable());
3967 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x55AA);
3968 }
3969
3970 #[test]
3971 fn ok_ticket_force_with_sync_interval_none() {
3972 let (_dir, mut cfg) = tmp_cfg();
3973 cfg.sync_interval = None;
3974
3975 let mmap = MemMap::<u64>::new(cfg).unwrap();
3976 assert!(mmap.bg_thread.is_none());
3977
3978 let (ticket, _) = mmap.write(1, |v| *v = 0x1122).unwrap();
3979 assert!(!ticket.is_durable());
3980
3981 let epoch = ticket.force().unwrap();
3982 assert_eq!(epoch, ticket.epoch());
3983 assert!(ticket.is_durable());
3984 assert!(mmap.durable_epoch() >= ticket.epoch());
3985 }
3986
3987 #[test]
3988 fn ok_drop_with_sync_interval_none() {
3989 let (_dir, mut cfg) = tmp_cfg();
3990 cfg.sync_interval = None;
3991
3992 {
3993 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3994 assert!(mmap.bg_thread.is_none());
3995 let (_ticket, _) = mmap.write(0, |v| *v = 0xCAFE_F00D).unwrap();
3996 drop(mmap);
3997 }
3998
3999 let reopened = MemMap::<u64>::open(cfg).unwrap();
4000 assert_eq!(reopened.read(0, |v| *v).unwrap(), 0xCAFE_F00D);
4001 }
4002
4003 #[test]
4004 fn ok_bg_worker_parks_when_idle_and_unparks_on_write() {
4005 let (_dir, mut cfg) = tmp_cfg();
4006 cfg.sync_interval = Some(Duration::from_millis(0x0F));
4007
4008 let mmap = MemMap::<u64>::new(cfg).unwrap();
4009
4010 let start = std::time::Instant::now();
4012 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
4013 std::thread::sleep(Duration::from_millis(5));
4014 }
4015 assert!(mmap.is_parked());
4016
4017 let (ticket, _) = mmap.write(0, |v| *v = 0x9999).unwrap();
4019 assert!(!mmap.is_parked());
4020
4021 let epoch = ticket.wait().unwrap();
4023 assert_eq!(epoch, ticket.epoch());
4024 assert!(ticket.is_durable());
4025 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999);
4026
4027 let start = std::time::Instant::now();
4029 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
4030 std::thread::sleep(Duration::from_millis(5));
4031 }
4032 assert!(mmap.is_parked());
4033 }
4034
4035 #[test]
4036 fn ok_batched_writes_while_active_remain_unparked() {
4037 let (_dir, mut cfg) = tmp_cfg();
4038 cfg.sync_interval = Some(Duration::from_millis(0x1E));
4039
4040 let mmap = MemMap::<u64>::new(cfg).unwrap();
4041
4042 let start = std::time::Instant::now();
4044 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
4045 std::thread::sleep(Duration::from_millis(5));
4046 }
4047 assert!(mmap.is_parked());
4048
4049 let (t1, _) = mmap.write(0, |v| *v = 0x1111).unwrap();
4051 assert!(!mmap.is_parked());
4052
4053 let (t2, _) = mmap.write(1, |v| *v = 0x2222).unwrap();
4055 let (t3, _) = mmap.write(2, |v| *v = 0x3333).unwrap();
4056 assert!(!mmap.is_parked());
4057
4058 t1.wait().unwrap();
4059 t2.wait().unwrap();
4060 t3.wait().unwrap();
4061
4062 assert!(t1.is_durable());
4063 assert!(t2.is_durable());
4064 assert!(t3.is_durable());
4065 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x1111);
4066 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x2222);
4067 assert_eq!(mmap.read(2, |v| *v).unwrap(), 0x3333);
4068 }
4069
4070 #[test]
4071 fn ok_concurrent_stress_park_unpark_no_lost_wakeups() {
4072 let (_dir, mut cfg) = tmp_cfg();
4073 cfg.sync_interval = Some(Duration::from_millis(5));
4074
4075 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
4076 const THREADS: usize = 4;
4077 const WRITES_PER_THREAD: usize = 0x19;
4078 let mut handles = Vec::with_capacity(THREADS);
4079
4080 for t in 0..THREADS {
4081 let mmap_clone = Arc::clone(&mmap);
4082 handles.push(std::thread::spawn(move || {
4083 let mut tickets = Vec::with_capacity(WRITES_PER_THREAD);
4084 for i in 0..WRITES_PER_THREAD {
4085 let val = (t as u64) << 0x20 | (i as u64);
4086 let slot = t;
4087 let (ticket, _) = mmap_clone.write(slot, |s| *s = val).unwrap();
4088 tickets.push(ticket);
4089
4090 if i % 5 == 0 {
4092 std::thread::sleep(Duration::from_millis(0x0F));
4093 }
4094 }
4095
4096 for ticket in tickets {
4097 ticket.wait().unwrap();
4098 assert!(ticket.is_durable());
4099 }
4100 }));
4101 }
4102
4103 for h in handles {
4104 h.join().unwrap();
4105 }
4106
4107 assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
4108 }
4109
4110 #[test]
4111 fn ok_spurious_wakeup_resumes_park_immediately() {
4112 let (_dir, mut cfg) = tmp_cfg();
4113 cfg.sync_interval = Some(Duration::from_secs(0x0A)); let mmap = MemMap::<u64>::new(cfg).unwrap();
4116
4117 let start = std::time::Instant::now();
4119 while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
4120 std::thread::sleep(Duration::from_millis(5));
4121 }
4122 assert!(mmap.is_parked());
4123
4124 mmap.inner.sync_condvar.notify_one();
4126 std::thread::sleep(Duration::from_millis(0x0A));
4127
4128 let start = std::time::Instant::now();
4131 while !mmap.is_parked() && start.elapsed() < Duration::from_millis(0x1F4) {
4132 std::thread::sleep(Duration::from_millis(5));
4133 }
4134 assert!(mmap.is_parked());
4135 }
4136
4137 #[test]
4138 fn ok_delete_cleans_up_file_when_mmap_already_unmapped() {
4139 let (_dir, cfg) = tmp_cfg();
4140 let path = cfg.versioned_path();
4141 let mmap = MemMap::<u64>::new(cfg).unwrap();
4142 assert!(path.exists());
4143
4144 {
4146 let mut guard = mmap.inner.mmap.write().unwrap();
4147 let _ = guard.take();
4148 }
4149
4150 assert!(mmap.delete().is_ok());
4152 assert!(!path.exists());
4153 }
4154
4155 mod memmap_grow {
4156 use super::*;
4157
4158 #[test]
4159 fn ok_grow_updates_length() {
4160 let (_dir, cfg) = tmp_cfg();
4161 let mmap = MemMap::<u64>::new(cfg).unwrap();
4162 let initial_len = mmap.length();
4163 let page_size = mmap.inner.page_size.as_usize();
4164
4165 mmap.grow(4).unwrap();
4166 assert_eq!(mmap.length(), initial_len + 4 * page_size);
4167 }
4168
4169 #[test]
4170 fn ok_grow_zero_count() {
4171 let (_dir, cfg) = tmp_cfg();
4172 let mmap = MemMap::<u64>::new(cfg).unwrap();
4173 let initial_len = mmap.length();
4174
4175 mmap.grow(0).unwrap();
4176 assert_eq!(mmap.length(), initial_len);
4177 }
4178
4179 #[test]
4180 fn ok_grow_extends_with_zero() {
4181 let (_dir, cfg) = tmp_cfg();
4182 let mmap = MemMap::<u64>::new(cfg).unwrap();
4183 let initial_len = mmap.length();
4184 let page_size = mmap.inner.page_size.as_usize();
4185 let page_slots = page_size / core::mem::size_of::<u64>();
4186 let initial_slots = initial_len / core::mem::size_of::<u64>();
4187 mmap.grow(1).unwrap();
4188
4189 for idx in initial_slots..(initial_slots + page_slots) {
4190 assert_eq!(mmap.read(idx, |v| *v).unwrap(), 0u64);
4191 }
4192 }
4193
4194 #[test]
4195 fn ok_grow_preserves_existing_data() {
4196 let (_dir, cfg) = tmp_cfg();
4197 let mmap = MemMap::<u64>::new(cfg).unwrap();
4198 let initial_len = mmap.length();
4199 let page_size = mmap.inner.page_size.as_usize();
4200 let page_slots = page_size / core::mem::size_of::<u64>();
4201 let initial_slots = initial_len / core::mem::size_of::<u64>();
4202
4203 for i in 0..initial_slots {
4204 mmap.write(i, |v| *v = 0xCAFE_0000 + i as u64).unwrap();
4205 }
4206
4207 mmap.grow(1).unwrap();
4208
4209 for i in 0..initial_slots {
4210 assert_eq!(mmap.read(i, |v| *v).unwrap(), 0xCAFE_0000 + i as u64);
4211 }
4212
4213 for i in initial_slots..(initial_slots + page_slots) {
4214 mmap.write(i, |v| *v = 0xC0DE_0000 + i as u64).unwrap();
4215 assert_eq!(mmap.read(i, |v| *v).unwrap(), 0xC0DE_0000 + i as u64);
4216 }
4217 }
4218
4219 #[test]
4220 fn ok_grow_syncs_dirty_pages_when_curr_epoch_gt_durable() {
4221 let (_dir, mut cfg) = tmp_cfg();
4222 cfg.sync_interval = None; let mmap = MemMap::<u64>::new(cfg).unwrap();
4225 let (ticket, _) = mmap.write(1, |v| *v = 0x1234_5678u64).unwrap();
4226
4227 assert!(!ticket.is_durable());
4228 assert!(mmap.current_epoch() > mmap.durable_epoch());
4229
4230 mmap.grow(2).unwrap();
4232
4233 assert!(ticket.is_durable());
4234 assert!(mmap.durable_epoch() >= ticket.epoch());
4235 }
4236
4237 #[test]
4238 fn ok_grow_when_already_durable() {
4239 let (_dir, mut cfg) = tmp_cfg();
4240 cfg.sync_interval = None;
4241
4242 let mmap = MemMap::<u64>::new(cfg).unwrap();
4243 let (ticket, _) = mmap.write(0, |v| *v = 0x9999u64).unwrap();
4244 mmap.sync().unwrap();
4245 assert!(ticket.is_durable());
4246 assert_eq!(mmap.current_epoch(), mmap.durable_epoch());
4247
4248 mmap.grow(2).unwrap();
4249 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x9999u64);
4250 }
4251
4252 #[test]
4253 fn err_grow_overflow_usize() {
4254 let (_dir, cfg) = tmp_cfg();
4255 let mmap = MemMap::<u64>::new(cfg).unwrap();
4256
4257 let err = mmap.grow(usize::MAX).unwrap_err();
4258 assert_eq!(err.reason, err::GRW.reason);
4259 }
4260
4261 #[test]
4262 fn ok_grow_multiple_times() {
4263 let (_dir, cfg) = tmp_cfg();
4264 let mmap = MemMap::<u64>::new(cfg).unwrap();
4265 let initial_len = mmap.length();
4266 let page_size = mmap.inner.page_size.as_usize();
4267 let initial_slots = initial_len / core::mem::size_of::<u64>();
4268
4269 mmap.write(0, |v| *v = 0x0Au64).unwrap();
4270 mmap.grow(2).unwrap();
4271 mmap.write(initial_slots, |v| *v = 0x14u64).unwrap();
4272 mmap.grow(3).unwrap();
4273 mmap.write(initial_slots + 2, |v| *v = 0x1Eu64).unwrap();
4274
4275 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x0Au64);
4276 assert_eq!(mmap.read(initial_slots, |v| *v).unwrap(), 0x14u64);
4277 assert_eq!(mmap.read(initial_slots + 2, |v| *v).unwrap(), 0x1Eu64);
4278 assert_eq!(mmap.length(), initial_len + 5 * page_size);
4279 }
4280
4281 #[test]
4282 fn ok_concurrent_grow_and_reads_writes() {
4283 let (_dir, mut cfg) = tmp_cfg();
4284 cfg.sync_interval = Some(Duration::from_millis(0x0A));
4285 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
4286 let initial_len = mmap.length();
4287 let page_size = mmap.inner.page_size.as_usize();
4288 let initial_slots = initial_len / core::mem::size_of::<u64>();
4289
4290 let num_writers = 4;
4291 let mut handles = Vec::new();
4292
4293 for w in 0..num_writers {
4294 let m = Arc::clone(&mmap);
4295 handles.push(std::thread::spawn(move || {
4296 for i in 0..0x32 {
4297 let _ = m.write(w % initial_slots, |v| *v = (w * 0x3E8 + i) as u64);
4298 m.read(w % initial_slots, |_| ()).unwrap();
4299 }
4300 }));
4301 }
4302
4303 let m_grower = Arc::clone(&mmap);
4304 let grower_handle = std::thread::spawn(move || {
4305 for _ in 0..3 {
4306 std::thread::sleep(Duration::from_millis(5));
4307 m_grower.grow(2).unwrap();
4308 }
4309 });
4310
4311 for h in handles {
4312 h.join().unwrap();
4313 }
4314 grower_handle.join().unwrap();
4315
4316 assert_eq!(mmap.length(), initial_len + 6 * page_size);
4317 }
4318
4319 #[test]
4320 fn err_grow_exceeds_max_len() {
4321 let (_dir, mut cfg) = tmp_cfg();
4322 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
4323 cfg.initial_pages = 1;
4324 cfg.max_len = 2 * page_size;
4325
4326 let mmap = MemMap::<u64>::new(cfg).unwrap();
4327 let err = mmap.grow(2).unwrap_err();
4328 assert_eq!(err.reason, err::GRW.reason);
4329 }
4330
4331 #[test]
4332 fn ok_grow_up_to_max_len() {
4333 let (_dir, mut cfg) = tmp_cfg();
4334 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
4335 cfg.initial_pages = 1;
4336 cfg.max_len = 3 * page_size;
4337
4338 let mmap = MemMap::<u64>::new(cfg).unwrap();
4339 mmap.grow(2).unwrap();
4340 assert_eq!(mmap.length(), 3 * page_size);
4341 }
4342
4343 #[test]
4344 fn ok_grow_in_place_preserves_base_ptr() {
4345 let (_dir, mut cfg) = tmp_cfg();
4346 let page_size = MemMap::<u64>::page_size().unwrap().as_usize();
4347 cfg.initial_pages = 1;
4348 cfg.max_len = 4 * page_size;
4349
4350 let mmap = MemMap::<u64>::new(cfg).unwrap();
4351 mmap.write(0, |v| *v = 0xAA_BBu64).unwrap();
4352
4353 let base_ptr_before = mmap.read(0, |v| v as *const u64).unwrap();
4354
4355 mmap.grow(2).unwrap();
4356
4357 let base_ptr_after = mmap.read(0, |v| v as *const u64).unwrap();
4358 assert_eq!(base_ptr_before, base_ptr_after);
4359 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xAA_BBu64);
4360
4361 let new_slot = (page_size / core::mem::size_of::<u64>()) + 1;
4362 mmap.write(new_slot, |v| *v = 0xCC_DDu64).unwrap();
4363 assert_eq!(mmap.read(new_slot, |v| *v).unwrap(), 0xCC_DDu64);
4364 }
4365 }
4366
4367 mod thread_failure_lifecycle {
4368 use super::*;
4369
4370 #[test]
4371 fn err_new_failure_paths_spawn_no_sync_thread() {
4372 let (_dir, cfg) = tmp_cfg();
4373 std::fs::write(cfg.versioned_path(), b"preexisting").unwrap();
4375
4376 let res = MemMap::<u64>::new(cfg);
4377 assert!(res.is_err());
4378 }
4380
4381 #[test]
4382 fn err_open_failure_paths_spawn_no_sync_thread() {
4383 let (_dir, mut cfg) = tmp_cfg();
4384 cfg.path = std::path::PathBuf::from("/non/existent/path/memmap.bin");
4385
4386 let res = MemMap::<u64>::open(cfg);
4387 assert!(res.is_err());
4388 }
4390
4391 #[test]
4392 fn ok_thread_shutdown_on_panic_unwinding() {
4393 let (_dir, cfg) = tmp_cfg();
4394 let thread_inner = Arc::new(atomic::AtomicBool::new(false));
4395 let thread_inner_clone = Arc::clone(&thread_inner);
4396
4397 let handle = std::thread::spawn(move || {
4399 let mmap = MemMap::<u64>::new(cfg).unwrap();
4400 let _inner = Arc::clone(&mmap.inner);
4401 thread_inner_clone.store(true, atomic::Ordering::Release);
4402 panic!("intentional test panic to trigger Drop unwinding");
4404 });
4405
4406 assert!(handle.join().is_err());
4407 assert!(thread_inner.load(atomic::Ordering::Acquire));
4408 }
4410
4411 #[test]
4412 fn ok_thread_shutdown_in_delete_even_if_already_unmapped() {
4413 let (_dir, cfg) = tmp_cfg();
4414 let mmap = MemMap::<u64>::new(cfg).unwrap();
4415
4416 {
4418 let mut guard = mmap.inner.mmap.write().unwrap();
4419 let _ = guard.take();
4420 }
4421
4422 assert!(mmap.delete().is_ok());
4424 }
4425
4426 #[test]
4427 fn ok_grow_page_multiples() {
4428 let (_dir, cfg) = tmp_cfg();
4429 let mmap = MemMap::<u64>::new(cfg).unwrap();
4430 let page_size = mmap.inner.page_size.as_usize();
4431 assert_eq!(mmap.length(), page_size);
4432
4433 mmap.grow(1).unwrap();
4434 assert_eq!(mmap.length(), 2 * page_size);
4435
4436 mmap.grow(3).unwrap();
4437 assert_eq!(mmap.length(), 5 * page_size);
4438 }
4439
4440 #[test]
4441 fn ok_grow_failure_keeps_thread_healthy() {
4442 let (_dir, cfg) = tmp_cfg();
4443 let mmap = MemMap::<u64>::new(cfg).unwrap();
4444
4445 assert!(mmap.grow(usize::MAX).is_err());
4446
4447 let (ticket, _) = mmap.write(0, |v| *v = 0x2Au64).unwrap();
4449 mmap.sync().unwrap();
4450 assert!(ticket.is_durable());
4451 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x2Au64);
4452 }
4453 }
4454
4455 mod memmap_transactions {
4456 use super::*;
4457
4458 #[test]
4459 fn ok_single_slot_cas_success_and_conflict() {
4460 let (_dir, cfg) = tmp_cfg();
4461 let mmap = MemMap::<u64>::new(cfg).unwrap();
4462
4463 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
4464
4465 let t1 = mmap.compare_and_write(&0, &0x64, 0).unwrap();
4466 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x64);
4467 assert_eq!(mmap.current_epoch(), 1);
4468
4469 let err = mmap.compare_and_write(&0, &0xC8, 0).unwrap_err();
4470 assert_eq!(err.reason, err::CAS.reason);
4471 assert!(err.context.contains("slot data has changed"));
4472 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x64);
4473 assert_eq!(mmap.current_epoch(), 1); let t2 = mmap.compare_and_write(&0x64, &0xC8, 0).unwrap();
4476 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0xC8);
4477 assert_eq!(mmap.current_epoch(), 2);
4478 assert!(t2.epoch() > t1.epoch());
4479 }
4480
4481 #[test]
4482 fn ok_tx_multiple_writes_single_epoch() {
4483 let (_dir, cfg) = tmp_cfg();
4484 let mmap = MemMap::<u64>::new(cfg).unwrap();
4485
4486 let epoch_before = mmap.current_epoch();
4487
4488 let mut tx = mmap.new_tx();
4489 assert!(tx.is_empty());
4490 tx.write(&0, &0x6F, 0).unwrap();
4491 tx.write(&0, &0xDE, 1).unwrap();
4492 tx.write(&0, &0x14D, 2).unwrap();
4493 assert_eq!(tx.len(), 3);
4494
4495 let ticket = tx.commit().unwrap();
4496 let epoch_after = mmap.current_epoch();
4497
4498 assert_eq!(epoch_after, epoch_before + 1);
4500 assert_eq!(ticket.epoch(), epoch_after);
4501
4502 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x6F);
4503 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0xDE);
4504 assert_eq!(mmap.read(2, |v| *v).unwrap(), 0x14D);
4505
4506 ticket.wait().unwrap();
4507 assert!(ticket.is_durable());
4508 }
4509
4510 #[test]
4511 fn err_tx_cas_conflict_aborts_all_with_zero_side_effects() {
4512 let (_dir, cfg) = tmp_cfg();
4513 let mmap = MemMap::<u64>::new(cfg).unwrap();
4514
4515 mmap.write(1, |v| *v = 0x3E7).unwrap();
4517 let epoch_before = mmap.current_epoch();
4518
4519 let mut tx = mmap.new_tx();
4520 tx.write(&0, &0x0A, 0).unwrap(); tx.write(&0, &0x14, 1).unwrap(); tx.write(&0, &0x1E, 2).unwrap(); let err = tx.commit().unwrap_err();
4525 assert_eq!(err.reason, err::CAS.reason);
4526 assert!(err.context.contains("state conflict at index 1"));
4527
4528 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
4530 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0x3E7);
4531 assert_eq!(mmap.read(2, |v| *v).unwrap(), 0);
4532
4533 assert_eq!(mmap.current_epoch(), epoch_before);
4535 }
4536
4537 #[test]
4538 fn ok_tx_retry_after_conflict() {
4539 let (_dir, cfg) = tmp_cfg();
4540 let mmap = MemMap::<u64>::new(cfg).unwrap();
4541
4542 mmap.write(1, |v| *v = 0x2A).unwrap();
4544
4545 let mut tx1 = mmap.new_tx();
4547 tx1.write(&0, &0x64, 0).unwrap();
4548 tx1.write(&0, &0xC8, 1).unwrap();
4549 assert_eq!(tx1.commit().unwrap_err().reason, err::CAS.reason);
4550
4551 let curr_val = mmap.read(1, |v| *v).unwrap();
4553 assert_eq!(curr_val, 0x2A);
4554
4555 let mut tx2 = mmap.new_tx();
4556 tx2.write(&0, &0x64, 0).unwrap();
4557 tx2.write(&0x2A, &0xC8, 1).unwrap();
4558 let ticket = tx2.commit().unwrap();
4559
4560 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x64);
4561 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0xC8);
4562 ticket.wait().unwrap();
4563 assert!(ticket.is_durable());
4564 }
4565
4566 #[test]
4567 fn err_tx_out_of_order() {
4568 let (_dir, cfg) = tmp_cfg();
4569 let mmap = MemMap::<u64>::new(cfg).unwrap();
4570
4571 let mut tx = mmap.new_tx();
4572 tx.write(&0, &0x0A, 2).unwrap();
4573 let err = tx.write(&0, &0x14, 1).unwrap_err();
4574 assert_eq!(err.reason, err::HCF.reason);
4575 assert!(err.context.contains("strictly ordered"));
4576 }
4577
4578 #[test]
4579 fn err_tx_duplicate_index() {
4580 let (_dir, cfg) = tmp_cfg();
4581 let mmap = MemMap::<u64>::new(cfg).unwrap();
4582
4583 let mut tx = mmap.new_tx();
4584 tx.write(&0, &0x0A, 1).unwrap();
4585 let err = tx.write(&0, &0x14, 1).unwrap_err();
4586 assert_eq!(err.reason, err::HCF.reason);
4587 assert!(err.context.contains("strictly ordered"));
4588 }
4589
4590 #[test]
4591 fn err_tx_out_of_bounds() {
4592 let (_dir, cfg) = tmp_cfg();
4593 let mmap = MemMap::<u64>::new(cfg).unwrap();
4594 let slots = mmap.length() / core::mem::size_of::<u64>();
4595
4596 let mut tx = mmap.new_tx();
4597 let err = tx.write(&0, &0x0A, slots).unwrap_err();
4598 assert_eq!(err.reason, err::HCF.reason);
4599 }
4600
4601 #[test]
4602 fn err_tx_empty_commit() {
4603 let (_dir, cfg) = tmp_cfg();
4604 let mmap = MemMap::<u64>::new(cfg).unwrap();
4605
4606 let tx = mmap.new_tx();
4607 let err = tx.commit().unwrap_err();
4608 assert_eq!(err.reason, err::INV.reason);
4609 assert!(err.context.contains("empty transaction"));
4610 }
4611
4612 #[test]
4613 fn ok_tx_discard_on_drop() {
4614 let (_dir, cfg) = tmp_cfg();
4615 let mmap = MemMap::<u64>::new(cfg).unwrap();
4616
4617 let epoch_before = mmap.current_epoch();
4618 {
4619 let mut tx = mmap.new_tx();
4620 tx.write(&0, &0x309, 0).unwrap();
4621 tx.write(&0, &0x378, 1).unwrap();
4622 }
4624
4625 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0);
4626 assert_eq!(mmap.read(1, |v| *v).unwrap(), 0);
4627 assert_eq!(mmap.current_epoch(), epoch_before);
4628 }
4629
4630 #[test]
4631 fn ok_tx_concurrent_non_overlapping() {
4632 let (_dir, cfg) = tmp_cfg();
4633 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
4634
4635 let mut handles = Vec::new();
4636 for thread_id in 0..4u64 {
4637 let m = mmap.clone();
4638 handles.push(std::thread::spawn(move || {
4639 let base = (thread_id as usize) * 4;
4640 let mut tx = m.new_tx();
4641 tx.write(&0, &(thread_id * 0x0A + 1), base).unwrap();
4642 tx.write(&0, &(thread_id * 0x0A + 2), base + 1).unwrap();
4643 tx.write(&0, &(thread_id * 0x0A + 3), base + 2).unwrap();
4644 tx.write(&0, &(thread_id * 0x0A + 4), base + 3).unwrap();
4645 let ticket = tx.commit().unwrap();
4646 ticket.wait().unwrap();
4647 }));
4648 }
4649
4650 for h in handles {
4651 h.join().unwrap();
4652 }
4653
4654 for thread_id in 0..4u64 {
4655 let base = (thread_id as usize) * 4;
4656 assert_eq!(mmap.read(base, |v| *v).unwrap(), thread_id * 0x0A + 1);
4657 assert_eq!(mmap.read(base + 1, |v| *v).unwrap(), thread_id * 0x0A + 2);
4658 assert_eq!(mmap.read(base + 2, |v| *v).unwrap(), thread_id * 0x0A + 3);
4659 assert_eq!(mmap.read(base + 3, |v| *v).unwrap(), thread_id * 0x0A + 4);
4660 }
4661 }
4662
4663 #[test]
4664 fn ok_tx_durability_across_reopen() {
4665 let (_dir, cfg) = tmp_cfg();
4666
4667 {
4668 let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
4669 let mut tx = mmap.new_tx();
4670 tx.write(&0, &0xAAAA_BBBB, 0).unwrap();
4671 tx.write(&0, &0xCCCC_DDDD, 1).unwrap();
4672 let ticket = tx.commit().unwrap();
4673 ticket.wait().unwrap();
4674 assert!(ticket.is_durable());
4675 }
4676
4677 {
4678 let reopened = MemMap::<u64>::open(cfg).unwrap();
4679 assert_eq!(reopened.read(0, |v| *v).unwrap(), 0xAAAA_BBBB);
4680 assert_eq!(reopened.read(1, |v| *v).unwrap(), 0xCCCC_DDDD);
4681 }
4682 }
4683
4684 #[test]
4685 fn ok_tx_unparks_idle_bg_worker() {
4686 let (_dir, cfg) = tmp_cfg();
4687 let mmap = MemMap::<u64>::new(cfg).unwrap();
4688
4689 let mut parked = false;
4691 for _ in 0..0x64 {
4692 if mmap.is_parked() {
4693 parked = true;
4694 break;
4695 }
4696 std::thread::sleep(Duration::from_millis(0x0A));
4697 }
4698 assert!(parked);
4699
4700 let mut tx = mmap.new_tx();
4701 tx.write(&0, &0x22B, 0).unwrap();
4702 let ticket = tx.commit().unwrap();
4703
4704 ticket.wait().unwrap();
4706 assert!(ticket.is_durable());
4707 assert_eq!(mmap.read(0, |v| *v).unwrap(), 0x22B);
4708 }
4709
4710 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
4711 #[repr(C)]
4712 struct TxNode {
4713 id: u64,
4714 flags: u64,
4715 keys: [u64; 2],
4716 }
4717
4718 #[test]
4719 fn ok_concurrent_tx_read_exact_and_write_exact_same_slots() {
4720 let (_dir, cfg) = tmp_cfg();
4721 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
4722
4723 let barrier = Arc::new(std::sync::Barrier::new(6));
4724 let mut handles = Vec::new();
4725
4726 for t in 0..2 {
4727 let m = Arc::clone(&mmap);
4728 let b = Arc::clone(&barrier);
4729 handles.push(std::thread::spawn(move || {
4730 b.wait();
4731 for i in 0..0xC8 {
4732 let val = ((t + 1) as u64) << 0x20 | (i as u64);
4733 let _ = m
4734 .write_exact(0, |node: &mut TxNode| {
4735 node.id = val;
4736 node.flags = val;
4737 node.keys = [val, val];
4738 })
4739 .unwrap();
4740 }
4741 }));
4742 }
4743
4744 for t in 2..4 {
4745 let m = Arc::clone(&mmap);
4746 let b = Arc::clone(&barrier);
4747 handles.push(std::thread::spawn(move || {
4748 b.wait();
4749 for i in 0..0xC8 {
4750 let new_val = ((t + 1) as u64) << 0x20 | (i as u64);
4751 let current_node = m.read_exact(0, |n: &TxNode| *n).unwrap();
4752 let mut tx = m.new_tx();
4753 if tx.write(¤t_node.id, &new_val, 0).is_ok()
4754 && tx.write(¤t_node.flags, &new_val, 1).is_ok()
4755 && tx.write(¤t_node.keys[0], &new_val, 2).is_ok()
4756 && tx.write(¤t_node.keys[1], &new_val, 3).is_ok()
4757 {
4758 let _ = tx.commit();
4759 }
4760 }
4761 }));
4762 }
4763
4764 for _ in 0..2 {
4765 let m = Arc::clone(&mmap);
4766 let b = Arc::clone(&barrier);
4767 handles.push(std::thread::spawn(move || {
4768 b.wait();
4769 for _ in 0..0x190 {
4770 m.read_exact(0, |node: &TxNode| {
4771 assert_eq!(node.id, node.flags);
4772 assert_eq!(node.id, node.keys[0]);
4773 assert_eq!(node.id, node.keys[1]);
4774 })
4775 .unwrap();
4776 }
4777 }));
4778 }
4779
4780 for h in handles {
4781 h.join().unwrap();
4782 }
4783
4784 let final_node = mmap.read_exact(0, |n: &TxNode| *n).unwrap();
4785 assert_eq!(final_node.id, final_node.flags);
4786 assert_eq!(final_node.id, final_node.keys[0]);
4787 assert_eq!(final_node.id, final_node.keys[1]);
4788 }
4789
4790 #[test]
4791 fn ok_tx_retry_loop_against_concurrent_write_exact_and_read_exact() {
4792 let (_dir, cfg) = tmp_cfg();
4793 let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
4794
4795 let barrier = Arc::new(std::sync::Barrier::new(4));
4796 let mut handles = Vec::new();
4797
4798 {
4799 let m = Arc::clone(&mmap);
4800 let b = Arc::clone(&barrier);
4801 handles.push(std::thread::spawn(move || {
4802 b.wait();
4803 for i in 0..0x64 {
4804 let _ = m
4805 .write_exact(0, |pair: &mut [u64; 2]| {
4806 pair[0] = i as u64;
4807 pair[1] = i as u64;
4808 })
4809 .unwrap();
4810 }
4811 }));
4812 }
4813
4814 {
4815 let m = Arc::clone(&mmap);
4816 let b = Arc::clone(&barrier);
4817 handles.push(std::thread::spawn(move || {
4818 b.wait();
4819 let mut successful_commits = 0;
4820 while successful_commits < 10 {
4821 let current = m.read_exact(0, |pair: &[u64; 2]| *pair).unwrap();
4822 let new_val = 0x8000_0000 + successful_commits;
4823 let mut tx = m.new_tx();
4824 if tx.write(¤t[0], &new_val, 0).is_ok()
4825 && tx.write(¤t[1], &new_val, 1).is_ok()
4826 && tx.commit().is_ok()
4827 {
4828 successful_commits += 1;
4829 }
4830 }
4831 }));
4832 }
4833
4834 for _ in 0..2 {
4835 let m = Arc::clone(&mmap);
4836 let b = Arc::clone(&barrier);
4837 handles.push(std::thread::spawn(move || {
4838 b.wait();
4839 for _ in 0..0xC8 {
4840 m.read_exact(0, |pair: &[u64; 2]| {
4841 assert_eq!(pair[0], pair[1]);
4842 })
4843 .unwrap();
4844 }
4845 }));
4846 }
4847
4848 for h in handles {
4849 h.join().unwrap();
4850 }
4851
4852 mmap.read_exact(0, |pair: &[u64; 2]| {
4853 assert_eq!(pair[0], pair[1]);
4854 })
4855 .unwrap();
4856 }
4857 }
4858
4859 mod memmap_page_size {
4860 use super::*;
4861
4862 #[test]
4863 fn ok_page_size_returns_valid_power_of_two() {
4864 let page_size = MemMap::<u64>::page_size().unwrap();
4865 assert!(page_size.as_usize() >= 0x1000);
4866 assert!(page_size.is_power_of_two());
4867 assert_eq!(page_size.as_usize(), page_size.as_u32() as usize);
4868
4869 let matched_bytes = match page_size {
4870 PageSize::K4 => 4096,
4871 PageSize::K16 => 16384,
4872 PageSize::K64 => 65536,
4873 };
4874 assert_eq!(matched_bytes, page_size.as_usize());
4875 }
4876
4877 #[test]
4878 fn ok_page_size_consistency() {
4879 let ps1 = MemMap::<u64>::page_size().unwrap();
4880 let ps2 = MemMap::<u64>::page_size().unwrap();
4881 let ps3 = MemMap::<ValidT>::page_size().unwrap();
4882 assert_eq!(ps1, ps2);
4883 assert_eq!(ps1, ps3);
4884 }
4885
4886 #[test]
4887 fn ok_page_size_across_threads() {
4888 let expected = MemMap::<u64>::page_size().unwrap();
4889 let mut handles = Vec::new();
4890
4891 for _ in 0..8 {
4892 handles.push(std::thread::spawn(move || {
4893 let ps = MemMap::<u64>::page_size().unwrap();
4894 assert_eq!(ps, expected);
4895 }));
4896 }
4897
4898 for h in handles {
4899 h.join().unwrap();
4900 }
4901 }
4902
4903 #[test]
4904 fn ok_page_size_matches_inner_stored() {
4905 let (_dir, cfg) = tmp_cfg();
4906 let mmap = MemMap::<u64>::new(cfg).unwrap();
4907 let sys_page_size = MemMap::<u64>::page_size().unwrap();
4908 assert_eq!(mmap.inner.page_size, sys_page_size);
4909 }
4910
4911 #[test]
4912 fn ok_initial_pages_multi() {
4913 let (_dir, mut cfg) = tmp_cfg();
4914 cfg.initial_pages = 3;
4915 let mmap = MemMap::<u64>::new(cfg).unwrap();
4916 let page_size = mmap.inner.page_size.as_usize();
4917 assert_eq!(mmap.length(), 3 * page_size);
4918 let slots = mmap.length() / core::mem::size_of::<u64>();
4919 assert!(mmap.write(slots - 1, |v| *v = 0xAA).is_ok());
4920 assert_eq!(mmap.read(slots - 1, |v| *v).unwrap(), 0xAA);
4921 }
4922 }
4923}