1use std::fs::{File, OpenOptions};
64use std::mem::size_of;
65use std::path::Path;
66use std::sync::atomic::{AtomicU64, Ordering};
67
68use memmap2::{Mmap, MmapMut, MmapOptions};
69
70pub const ARENA_MAGIC: u64 = 0x4150_5341_524E_4131;
71
72enum Mapping {
75 Writable(MmapMut),
76 ReadOnly(Mmap),
77}
78
79impl Mapping {
80 #[inline]
81 fn as_ptr(&self) -> *const u8 {
82 match self {
83 Mapping::Writable(m) => m.as_ptr(),
84 Mapping::ReadOnly(m) => m.as_ptr(),
85 }
86 }
87
88 #[inline]
89 fn is_writable(&self) -> bool {
90 matches!(self, Mapping::Writable(_))
91 }
92
93 fn flush(&self) -> Result<(), std::io::Error> {
94 match self {
95 Mapping::Writable(m) => m.flush(),
96 Mapping::ReadOnly(_) => Ok(()),
97 }
98 }
99
100 fn flush_async(&self) -> Result<(), std::io::Error> {
101 match self {
102 Mapping::Writable(m) => m.flush_async(),
103 Mapping::ReadOnly(_) => Ok(()),
104 }
105 }
106}
107
108#[repr(C, align(64))]
109pub struct ArenaHeader {
110 pub magic: u64,
111 pub capacity_bytes: u64,
112 pub used_bytes: AtomicU64,
113 _pad: [u8; 40],
114}
115
116const _: () = {
117 assert!(size_of::<ArenaHeader>() == 64);
118};
119
120pub const fn arena_file_size(capacity_bytes: usize) -> usize {
121 size_of::<ArenaHeader>() + capacity_bytes
122}
123
124#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub enum ArenaError {
126 Full,
127 InvalidRef,
128 InvalidUtf8,
129 LayoutMismatch,
130 ReadOnly,
132 IoError(std::io::ErrorKind),
133}
134
135impl From<std::io::Error> for ArenaError {
136 fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
137}
138
139#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
144pub struct StringRef {
145 pub offset: u32,
146 pub len: u32,
147}
148
149impl StringRef {
150 #[inline]
151 pub fn to_u64(self) -> u64 {
152 ((self.offset as u64) << 32) | (self.len as u64)
153 }
154 #[inline]
155 pub fn from_u64(v: u64) -> Self {
156 Self {
157 offset: (v >> 32) as u32,
158 len: v as u32,
159 }
160 }
161}
162
163pub struct SharedStringArena {
164 _file: File,
165 mmap: Mapping,
166 capacity_bytes: usize,
167 header_sidecar: subetha_core::HandshakeHeader,
168 ring_sidecar: Box<subetha_core::ObservationRing>,
169}
170
171unsafe impl Send for SharedStringArena {}
172unsafe impl Sync for SharedStringArena {}
173
174impl subetha_sidecar::AdaptiveInstance for SharedStringArena {
175 fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
176 fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
177 fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
178 Box::new(subetha_sidecar::NoMigrationPolicy)
179 }
180}
181
182impl SharedStringArena {
183 pub fn create(
190 path: impl AsRef<Path>, capacity_bytes: usize,
191 ) -> Result<Self, ArenaError> {
192 Self::check_capacity(capacity_bytes)?;
193 let (file, mmap) = crate::mmf_attach::create_or_attach(
194 path.as_ref(),
195 arena_file_size(capacity_bytes),
196 |ptr| unsafe { Self::init_region(ptr, capacity_bytes) },
197 |ptr| unsafe { (*(ptr as *const ArenaHeader)).magic == ARENA_MAGIC },
198 )?;
199 let this = Self {
200 _file: file, mmap: Mapping::Writable(mmap), capacity_bytes,
201 header_sidecar: subetha_core::HandshakeHeader::new(),
202 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
203 };
204 this.validate(capacity_bytes)?;
205 Ok(this)
206 }
207
208 pub fn reset(
212 path: impl AsRef<Path>, capacity_bytes: usize,
213 ) -> Result<Self, ArenaError> {
214 Self::check_capacity(capacity_bytes)?;
215 let (file, mmap) = crate::mmf_attach::reset(
216 path.as_ref(),
217 arena_file_size(capacity_bytes),
218 |ptr| unsafe { Self::init_region(ptr, capacity_bytes) },
219 )?;
220 Ok(Self {
221 _file: file, mmap: Mapping::Writable(mmap), capacity_bytes,
222 header_sidecar: subetha_core::HandshakeHeader::new(),
223 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
224 })
225 }
226
227 fn check_capacity(capacity_bytes: usize) -> Result<(), ArenaError> {
232 if capacity_bytes < 1 || capacity_bytes > u32::MAX as usize {
233 return Err(ArenaError::LayoutMismatch);
234 }
235 Ok(())
236 }
237
238 unsafe fn init_region(ptr: *mut u8, capacity_bytes: usize) {
246 let hdr = ptr as *mut ArenaHeader;
247 unsafe {
248 (*hdr).capacity_bytes = capacity_bytes as u64;
249 std::ptr::write_volatile(&raw mut (*hdr).magic, ARENA_MAGIC);
250 }
251 }
252
253 pub fn open(
254 path: impl AsRef<Path>, expected_capacity_bytes: usize,
255 ) -> Result<Self, ArenaError> {
256 if expected_capacity_bytes > u32::MAX as usize {
259 return Err(ArenaError::LayoutMismatch);
260 }
261 let total = arena_file_size(expected_capacity_bytes);
262 let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
263 if file.metadata()?.len() < total as u64 {
264 return Err(ArenaError::LayoutMismatch);
265 }
266 let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
267 let this = Self {
268 _file: file, mmap: Mapping::Writable(mmap),
269 capacity_bytes: expected_capacity_bytes,
270 header_sidecar: subetha_core::HandshakeHeader::new(),
271 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
272 };
273 this.validate(expected_capacity_bytes)?;
274 Ok(this)
275 }
276
277 pub fn open_read_only(
284 path: impl AsRef<Path>, expected_capacity_bytes: usize,
285 ) -> Result<Self, ArenaError> {
286 let total = arena_file_size(expected_capacity_bytes);
287 let file = OpenOptions::new().read(true).open(path.as_ref())?;
288 if file.metadata()?.len() < total as u64 {
289 return Err(ArenaError::LayoutMismatch);
290 }
291 let mmap = unsafe { MmapOptions::new().len(total).map(&file)? };
292 let this = Self {
293 _file: file, mmap: Mapping::ReadOnly(mmap),
294 capacity_bytes: expected_capacity_bytes,
295 header_sidecar: subetha_core::HandshakeHeader::new(),
296 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
297 };
298 this.validate(expected_capacity_bytes)?;
299 Ok(this)
300 }
301
302 fn validate(&self, expected_capacity_bytes: usize) -> Result<(), ArenaError> {
304 let hdr = self.header();
305 if hdr.magic != ARENA_MAGIC || hdr.capacity_bytes != expected_capacity_bytes as u64 {
306 return Err(ArenaError::LayoutMismatch);
307 }
308 Ok(())
309 }
310
311 #[inline]
313 pub fn is_writable(&self) -> bool {
314 self.mmap.is_writable()
315 }
316
317 #[inline]
318 pub fn capacity_bytes(&self) -> usize { self.capacity_bytes }
319
320 #[inline]
321 pub fn used_bytes(&self) -> usize {
322 self.header().used_bytes.load(Ordering::Acquire) as usize
323 }
324
325 #[inline]
326 pub fn remaining_bytes(&self) -> usize {
327 self.capacity_bytes.saturating_sub(self.used_bytes())
328 }
329
330 fn header(&self) -> &ArenaHeader {
331 unsafe { &*(self.mmap.as_ptr() as *const ArenaHeader) }
332 }
333
334 pub fn intern(&self, s: &str) -> Result<StringRef, ArenaError> {
340 self.intern_bytes(s.as_bytes())
341 }
342
343 pub fn intern_bytes(&self, bytes: &[u8]) -> Result<StringRef, ArenaError> {
348 if !self.mmap.is_writable() {
349 return Err(ArenaError::ReadOnly);
350 }
351 let len = bytes.len() as u64;
352 if len > self.capacity_bytes as u64 {
353 self.ring_sidecar
354 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
355 return Err(ArenaError::Full);
356 }
357 let offset = self.header().used_bytes.fetch_add(len, Ordering::AcqRel);
358 if offset.saturating_add(len) > self.capacity_bytes as u64 {
359 self.header().used_bytes.fetch_sub(len, Ordering::AcqRel);
360 self.ring_sidecar
361 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
362 return Err(ArenaError::Full);
363 }
364 if offset.saturating_add(len) > u32::MAX as u64 {
371 self.header().used_bytes.fetch_sub(len, Ordering::AcqRel);
372 self.ring_sidecar
373 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
374 return Err(ArenaError::Full);
375 }
376 let dst = unsafe {
377 self.mmap.as_ptr()
378 .add(size_of::<ArenaHeader>())
379 .add(offset as usize)
380 as *mut u8
381 };
382 unsafe {
383 std::ptr::copy_nonoverlapping(bytes.as_ptr(), dst, bytes.len());
384 }
385 self.ring_sidecar
386 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 0);
387 Ok(StringRef { offset: offset as u32, len: len as u32 })
388 }
389
390 pub fn get_bytes(&self, r: StringRef) -> Result<&[u8], ArenaError> {
393 let end = (r.offset as u64).saturating_add(r.len as u64);
394 if end > self.header().used_bytes.load(Ordering::Acquire) {
395 self.ring_sidecar
396 .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 1);
397 return Err(ArenaError::InvalidRef);
398 }
399 if end > self.capacity_bytes as u64 {
400 self.ring_sidecar
401 .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 1);
402 return Err(ArenaError::InvalidRef);
403 }
404 self.ring_sidecar
405 .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 0);
406 let base = unsafe {
407 self.mmap.as_ptr()
408 .add(size_of::<ArenaHeader>())
409 .add(r.offset as usize)
410 };
411 Ok(unsafe { std::slice::from_raw_parts(base, r.len as usize) })
412 }
413
414 pub fn get(&self, r: StringRef) -> Result<&str, ArenaError> {
418 let bytes = self.get_bytes(r)?;
419 std::str::from_utf8(bytes).map_err(|_| ArenaError::InvalidUtf8)
420 }
421
422 pub fn intern_and_get(&self, s: &str) -> Result<(StringRef, &str), ArenaError> {
425 let r = self.intern(s)?;
426 let got = self.get(r)?;
427 Ok((r, got))
428 }
429
430 pub fn clear(&self) {
435 if !self.mmap.is_writable() {
436 return;
437 }
438 self.header().used_bytes.store(0, Ordering::Release);
439 self.ring_sidecar
440 .push_op(crate::sidecar_ops::string_arena::OP_CLEAR, 0);
441 }
442
443 pub fn flush(&self) -> Result<(), ArenaError> {
444 self.mmap.flush()?;
445 Ok(())
446 }
447
448 pub fn flush_async(&self) -> Result<(), ArenaError> {
452 self.mmap.flush_async()?;
453 Ok(())
454 }
455}
456
457#[cfg(test)]
458mod tests {
459 use super::*;
460 use std::sync::Arc;
461 use std::thread;
462
463 fn tmp(name: &str) -> std::path::PathBuf {
464 let mut p = std::env::temp_dir();
465 let pid = std::process::id();
466 p.push(format!("subetha-arena-{name}-{pid}.bin"));
467 p
468 }
469
470 #[test]
473 fn second_create_attaches_and_keeps_strings() {
474 let p = tmp("attach");
475 std::fs::remove_file(&p).ok();
476 let a = SharedStringArena::create(&p, 4096).unwrap();
477 let r = a.intern("held").unwrap();
478
479 let a2 = SharedStringArena::create(&p, 4096).unwrap();
480 assert_eq!(a2.get(r).unwrap(), "held", "attach lost an interned string");
481 assert!(matches!(
482 SharedStringArena::create(&p, 2048),
483 Err(ArenaError::LayoutMismatch),
484 ));
485
486 drop(a);
489 drop(a2);
490 let fresh = SharedStringArena::reset(&p, 4096).unwrap();
491 assert_eq!(fresh.used_bytes(), 0, "reset kept interned bytes");
492 drop(fresh);
493 std::fs::remove_file(&p).ok();
494 }
495
496 #[test]
497 fn a_read_only_arena_resolves_refs_and_refuses_interning() {
498 let p = tmp("readonly");
499 let r = {
500 let w = SharedStringArena::create(&p, 4096).unwrap();
501 let r = w.intern("notepad.exe").unwrap();
502 w.flush().unwrap();
503 r
504 };
505 let ro = SharedStringArena::open_read_only(&p, 4096).unwrap();
506 assert!(!ro.is_writable());
507 assert_eq!(ro.get(r), Ok("notepad.exe"));
508 assert_eq!(ro.get_bytes(r), Ok(&b"notepad.exe"[..]));
509 assert_eq!(ro.used_bytes(), 11);
510 assert_eq!(ro.intern("more"), Err(ArenaError::ReadOnly));
511 assert_eq!(ro.intern_bytes(b"more"), Err(ArenaError::ReadOnly));
512 ro.clear();
513 assert_eq!(ro.used_bytes(), 11, "clear on a read-only arena is inert");
514 ro.flush().unwrap();
515 std::fs::remove_file(&p).ok();
516 }
517
518 #[test]
519 fn a_read_only_open_still_validates_the_header() {
520 let p = tmp("readonly-mismatch");
521 {
522 let w = SharedStringArena::create(&p, 4096).unwrap();
523 w.intern("x").unwrap();
524 w.flush().unwrap();
525 }
526 assert_eq!(
527 SharedStringArena::open_read_only(&p, 2048).err(),
528 Some(ArenaError::LayoutMismatch)
529 );
530 std::fs::remove_file(&p).ok();
531 }
532
533 #[test]
534 fn create_initial_state_is_empty() {
535 let p = tmp("init");
536 let a = SharedStringArena::create(&p, 1024).unwrap();
537 assert_eq!(a.capacity_bytes(), 1024);
538 assert_eq!(a.used_bytes(), 0);
539 assert_eq!(a.remaining_bytes(), 1024);
540 std::fs::remove_file(&p).ok();
541 }
542
543 #[test]
544 fn intern_and_get_round_trip() {
545 let p = tmp("rt");
546 let a = SharedStringArena::create(&p, 1024).unwrap();
547 let r1 = a.intern("hello").unwrap();
548 let r2 = a.intern("world").unwrap();
549 assert_eq!(a.get(r1).unwrap(), "hello");
550 assert_eq!(a.get(r2).unwrap(), "world");
551 assert_eq!(a.used_bytes(), 10);
552 std::fs::remove_file(&p).ok();
553 }
554
555 #[test]
556 fn empty_string_interns_with_zero_len() {
557 let p = tmp("empty");
558 let a = SharedStringArena::create(&p, 16).unwrap();
559 let r = a.intern("").unwrap();
560 assert_eq!(r.len, 0);
561 assert_eq!(a.get(r).unwrap(), "");
562 assert_eq!(a.used_bytes(), 0);
563 std::fs::remove_file(&p).ok();
564 }
565
566 #[test]
567 fn full_arena_returns_error() {
568 let p = tmp("full");
569 let a = SharedStringArena::create(&p, 10).unwrap();
570 a.intern("hello").unwrap();
571 a.intern("world").unwrap();
572 assert_eq!(a.intern("more").err(), Some(ArenaError::Full));
573 assert_eq!(a.used_bytes(), 10);
575 std::fs::remove_file(&p).ok();
576 }
577
578 #[test]
579 fn string_too_large_returns_full() {
580 let p = tmp("too-large");
581 let a = SharedStringArena::create(&p, 8).unwrap();
582 let big = "x".repeat(100);
583 assert_eq!(a.intern(&big).err(), Some(ArenaError::Full));
584 assert_eq!(a.used_bytes(), 0);
585 std::fs::remove_file(&p).ok();
586 }
587
588 #[test]
589 fn string_ref_packs_and_unpacks() {
590 let r = StringRef { offset: 0x1234_5678, len: 42 };
591 let packed = r.to_u64();
592 let unpacked = StringRef::from_u64(packed);
593 assert_eq!(unpacked, r);
594 }
595
596 #[test]
597 fn cross_handle_visibility() {
598 let p = tmp("cross-handle");
599 let writer = SharedStringArena::create(&p, 1024).unwrap();
600 let reader = SharedStringArena::open(&p, 1024).unwrap();
601 let r = writer.intern("cross-process").unwrap();
602 assert_eq!(reader.get(r).unwrap(), "cross-process");
603 std::fs::remove_file(&p).ok();
604 }
605
606 #[test]
607 fn invalid_ref_beyond_used_rejected() {
608 let p = tmp("invalid");
609 let a = SharedStringArena::create(&p, 1024).unwrap();
610 a.intern("hi").unwrap(); let bad = StringRef { offset: 100, len: 5 };
612 assert_eq!(a.get(bad).err(), Some(ArenaError::InvalidRef));
613 std::fs::remove_file(&p).ok();
614 }
615
616 #[test]
617 fn concurrent_interners_get_distinct_refs() {
618 let p = tmp("concurrent");
619 let a: Arc<SharedStringArena> = Arc::new(SharedStringArena::create(&p, 4096).unwrap());
620 let n_threads = 4;
621 let per_thread = 20;
622 let mut handles = vec![];
623 for t in 0..n_threads {
624 let a = a.clone();
625 handles.push(thread::spawn(move || {
626 let mut refs = vec![];
627 for i in 0..per_thread {
628 let s = format!("thread-{t}-msg-{i:03}");
629 let r = a.intern(&s).unwrap();
630 refs.push((s, r));
631 }
632 refs
633 }));
634 }
635 let all: Vec<(String, StringRef)> = handles.into_iter()
636 .flat_map(|h| h.join().unwrap())
637 .collect();
638 for (expected, r) in &all {
640 let got = a.get(*r).unwrap();
641 assert_eq!(got, expected,
642 "ref offset={} len={} should resolve to {expected}",
643 r.offset, r.len);
644 }
645 let mut refs: Vec<StringRef> = all.iter().map(|(_, r)| *r).collect();
647 refs.sort_by_key(|r| r.offset);
648 for w in refs.windows(2) {
649 let r1_end = w[0].offset + w[0].len;
650 assert!(r1_end <= w[1].offset,
651 "ref {:?} overlaps with ref {:?}", w[0], w[1]);
652 }
653 std::fs::remove_file(&p).ok();
654 }
655
656 #[test]
657 fn intern_and_get_helper_returns_both() {
658 let p = tmp("intern-and-get");
659 let a = SharedStringArena::create(&p, 1024).unwrap();
660 let (r, s) = a.intern_and_get("composite").unwrap();
661 assert_eq!(s, "composite");
662 assert_eq!(a.get(r).unwrap(), "composite");
663 std::fs::remove_file(&p).ok();
664 }
665
666 #[test]
667 fn utf8_validation_on_get() {
668 let p = tmp("utf8");
669 let a = SharedStringArena::create(&p, 128).unwrap();
670 let r = a.intern("hello").unwrap();
672 assert!(a.get(r).is_ok());
673 let r2 = a.intern_bytes(&[0xFF, 0xFE, 0xFD]).unwrap();
677 assert_eq!(a.get(r2).err(), Some(ArenaError::InvalidUtf8));
678 assert_eq!(a.get_bytes(r2).unwrap(), &[0xFF, 0xFE, 0xFD]);
679 std::fs::remove_file(&p).ok();
680 }
681
682 #[test]
683 fn clear_resets_used_bytes() {
684 let p = tmp("clear");
685 let a = SharedStringArena::create(&p, 128).unwrap();
686 a.intern("first").unwrap();
687 a.intern("second").unwrap();
688 assert!(a.used_bytes() > 0);
689 a.clear();
690 assert_eq!(a.used_bytes(), 0);
691 let r = a.intern("after-clear").unwrap();
693 assert_eq!(a.get(r).unwrap(), "after-clear");
694 assert_eq!(r.offset, 0);
695 std::fs::remove_file(&p).ok();
696 }
697
698 #[test]
699 fn disk_persistence_survives_reopen() {
700 let p = tmp("disk");
701 let r_persist;
702 {
703 let a = SharedStringArena::create(&p, 1024).unwrap();
704 r_persist = a.intern("persisted-string").unwrap();
705 a.flush().unwrap();
706 }
707 let a2 = SharedStringArena::open(&p, 1024).unwrap();
708 assert_eq!(a2.get(r_persist).unwrap(), "persisted-string");
709 let r2 = a2.intern("more-after-reopen").unwrap();
711 assert_eq!(a2.get(r2).unwrap(), "more-after-reopen");
712 std::fs::remove_file(&p).ok();
713 }
714
715 #[test]
720 fn create_refuses_a_capacity_a_ref_cannot_address() {
721 assert!(matches!(
722 SharedStringArena::create(tmp("too-big"), u32::MAX as usize + 1),
723 Err(ArenaError::LayoutMismatch),
724 ));
725 assert!(matches!(
726 SharedStringArena::reset(tmp("too-big-reset"), u32::MAX as usize + 1),
727 Err(ArenaError::LayoutMismatch),
728 ));
729 assert!(matches!(
730 SharedStringArena::create(tmp("zero-cap"), 0),
731 Err(ArenaError::LayoutMismatch),
732 ));
733 }
734
735 #[test]
738 fn open_refuses_a_capacity_a_ref_cannot_address() {
739 assert!(matches!(
740 SharedStringArena::open(tmp("open-too-big"), u32::MAX as usize + 1),
741 Err(ArenaError::LayoutMismatch),
742 ));
743 }
744
745 #[test]
746 fn deduplication_via_hashmap_composition() {
747 use crate::SharedHashMap;
750 use crate::shared_hash_map::fnv1a_64;
751
752 let p_arena = tmp("dedup-arena");
753 let p_index = tmp("dedup-index");
754 let arena = SharedStringArena::create(&p_arena, 256).unwrap();
755 let index: SharedHashMap<u64, u64> = SharedHashMap::create(&p_index, 32).unwrap();
756
757 let s = "deduplicate-me";
758 let h = fnv1a_64(s.as_bytes());
759
760 let r = if let Some(packed) = index.get(&h) {
762 StringRef::from_u64(packed)
763 } else {
764 let r = arena.intern(s).unwrap();
765 index.insert(h, r.to_u64()).unwrap();
766 r
767 };
768 let used_after_first = arena.used_bytes();
769
770 let r2 = if let Some(packed) = index.get(&h) {
772 StringRef::from_u64(packed)
773 } else {
774 let r = arena.intern(s).unwrap();
775 index.insert(h, r.to_u64()).unwrap();
776 r
777 };
778 assert_eq!(r, r2, "dedup should return the same ref");
779 assert_eq!(arena.used_bytes(), used_after_first,
780 "second intern should not consume more bytes");
781
782 std::fs::remove_file(&p_arena).ok();
783 std::fs::remove_file(&p_index).ok();
784 }
785}