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 assert!(capacity_bytes >= 1);
193 assert!(capacity_bytes <= u32::MAX as usize,
194 "capacity_bytes must fit in u32 for StringRef offset");
195 let (file, mmap) = crate::mmf_attach::create_or_attach(
196 path.as_ref(),
197 arena_file_size(capacity_bytes),
198 |ptr| unsafe { Self::init_region(ptr, capacity_bytes) },
199 |ptr| unsafe { (*(ptr as *const ArenaHeader)).magic == ARENA_MAGIC },
200 )?;
201 let this = Self {
202 _file: file, mmap: Mapping::Writable(mmap), capacity_bytes,
203 header_sidecar: subetha_core::HandshakeHeader::new(),
204 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
205 };
206 this.validate(capacity_bytes)?;
207 Ok(this)
208 }
209
210 pub fn reset(
214 path: impl AsRef<Path>, capacity_bytes: usize,
215 ) -> Result<Self, ArenaError> {
216 assert!(capacity_bytes >= 1);
217 assert!(capacity_bytes <= u32::MAX as usize,
218 "capacity_bytes must fit in u32 for StringRef offset");
219 let (file, mmap) = crate::mmf_attach::reset(
220 path.as_ref(),
221 arena_file_size(capacity_bytes),
222 |ptr| unsafe { Self::init_region(ptr, capacity_bytes) },
223 )?;
224 Ok(Self {
225 _file: file, mmap: Mapping::Writable(mmap), capacity_bytes,
226 header_sidecar: subetha_core::HandshakeHeader::new(),
227 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
228 })
229 }
230
231 unsafe fn init_region(ptr: *mut u8, capacity_bytes: usize) {
239 let hdr = ptr as *mut ArenaHeader;
240 unsafe {
241 (*hdr).capacity_bytes = capacity_bytes as u64;
242 std::ptr::write_volatile(&raw mut (*hdr).magic, ARENA_MAGIC);
243 }
244 }
245
246 pub fn open(
247 path: impl AsRef<Path>, expected_capacity_bytes: usize,
248 ) -> Result<Self, ArenaError> {
249 let total = arena_file_size(expected_capacity_bytes);
250 let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
251 if file.metadata()?.len() < total as u64 {
252 return Err(ArenaError::LayoutMismatch);
253 }
254 let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
255 let this = Self {
256 _file: file, mmap: Mapping::Writable(mmap),
257 capacity_bytes: expected_capacity_bytes,
258 header_sidecar: subetha_core::HandshakeHeader::new(),
259 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
260 };
261 this.validate(expected_capacity_bytes)?;
262 Ok(this)
263 }
264
265 pub fn open_read_only(
272 path: impl AsRef<Path>, expected_capacity_bytes: usize,
273 ) -> Result<Self, ArenaError> {
274 let total = arena_file_size(expected_capacity_bytes);
275 let file = OpenOptions::new().read(true).open(path.as_ref())?;
276 if file.metadata()?.len() < total as u64 {
277 return Err(ArenaError::LayoutMismatch);
278 }
279 let mmap = unsafe { MmapOptions::new().len(total).map(&file)? };
280 let this = Self {
281 _file: file, mmap: Mapping::ReadOnly(mmap),
282 capacity_bytes: expected_capacity_bytes,
283 header_sidecar: subetha_core::HandshakeHeader::new(),
284 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
285 };
286 this.validate(expected_capacity_bytes)?;
287 Ok(this)
288 }
289
290 fn validate(&self, expected_capacity_bytes: usize) -> Result<(), ArenaError> {
292 let hdr = self.header();
293 if hdr.magic != ARENA_MAGIC || hdr.capacity_bytes != expected_capacity_bytes as u64 {
294 return Err(ArenaError::LayoutMismatch);
295 }
296 Ok(())
297 }
298
299 #[inline]
301 pub fn is_writable(&self) -> bool {
302 self.mmap.is_writable()
303 }
304
305 #[inline]
306 pub fn capacity_bytes(&self) -> usize { self.capacity_bytes }
307
308 #[inline]
309 pub fn used_bytes(&self) -> usize {
310 self.header().used_bytes.load(Ordering::Acquire) as usize
311 }
312
313 #[inline]
314 pub fn remaining_bytes(&self) -> usize {
315 self.capacity_bytes.saturating_sub(self.used_bytes())
316 }
317
318 fn header(&self) -> &ArenaHeader {
319 unsafe { &*(self.mmap.as_ptr() as *const ArenaHeader) }
320 }
321
322 pub fn intern(&self, s: &str) -> Result<StringRef, ArenaError> {
328 self.intern_bytes(s.as_bytes())
329 }
330
331 pub fn intern_bytes(&self, bytes: &[u8]) -> Result<StringRef, ArenaError> {
336 if !self.mmap.is_writable() {
337 return Err(ArenaError::ReadOnly);
338 }
339 let len = bytes.len() as u64;
340 if len > self.capacity_bytes as u64 {
341 self.ring_sidecar
342 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
343 return Err(ArenaError::Full);
344 }
345 let offset = self.header().used_bytes.fetch_add(len, Ordering::AcqRel);
346 if offset.saturating_add(len) > self.capacity_bytes as u64 {
347 self.header().used_bytes.fetch_sub(len, Ordering::AcqRel);
348 self.ring_sidecar
349 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
350 return Err(ArenaError::Full);
351 }
352 let dst = unsafe {
353 self.mmap.as_ptr()
354 .add(size_of::<ArenaHeader>())
355 .add(offset as usize)
356 as *mut u8
357 };
358 unsafe {
359 std::ptr::copy_nonoverlapping(bytes.as_ptr(), dst, bytes.len());
360 }
361 self.ring_sidecar
362 .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 0);
363 Ok(StringRef { offset: offset as u32, len: len as u32 })
364 }
365
366 pub fn get_bytes(&self, r: StringRef) -> Result<&[u8], ArenaError> {
369 let end = (r.offset as u64).saturating_add(r.len as u64);
370 if end > self.header().used_bytes.load(Ordering::Acquire) {
371 self.ring_sidecar
372 .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 1);
373 return Err(ArenaError::InvalidRef);
374 }
375 if end > self.capacity_bytes as u64 {
376 self.ring_sidecar
377 .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 1);
378 return Err(ArenaError::InvalidRef);
379 }
380 self.ring_sidecar
381 .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 0);
382 let base = unsafe {
383 self.mmap.as_ptr()
384 .add(size_of::<ArenaHeader>())
385 .add(r.offset as usize)
386 };
387 Ok(unsafe { std::slice::from_raw_parts(base, r.len as usize) })
388 }
389
390 pub fn get(&self, r: StringRef) -> Result<&str, ArenaError> {
394 let bytes = self.get_bytes(r)?;
395 std::str::from_utf8(bytes).map_err(|_| ArenaError::InvalidUtf8)
396 }
397
398 pub fn intern_and_get(&self, s: &str) -> Result<(StringRef, &str), ArenaError> {
401 let r = self.intern(s)?;
402 let got = self.get(r)?;
403 Ok((r, got))
404 }
405
406 pub fn clear(&self) {
411 if !self.mmap.is_writable() {
412 return;
413 }
414 self.header().used_bytes.store(0, Ordering::Release);
415 self.ring_sidecar
416 .push_op(crate::sidecar_ops::string_arena::OP_CLEAR, 0);
417 }
418
419 pub fn flush(&self) -> Result<(), ArenaError> {
420 self.mmap.flush()?;
421 Ok(())
422 }
423
424 pub fn flush_async(&self) -> Result<(), ArenaError> {
428 self.mmap.flush_async()?;
429 Ok(())
430 }
431}
432
433#[cfg(test)]
434mod tests {
435 use super::*;
436 use std::sync::Arc;
437 use std::thread;
438
439 fn tmp(name: &str) -> std::path::PathBuf {
440 let mut p = std::env::temp_dir();
441 let pid = std::process::id();
442 p.push(format!("subetha-arena-{name}-{pid}.bin"));
443 p
444 }
445
446 #[test]
449 fn second_create_attaches_and_keeps_strings() {
450 let p = tmp("attach");
451 std::fs::remove_file(&p).ok();
452 let a = SharedStringArena::create(&p, 4096).unwrap();
453 let r = a.intern("held").unwrap();
454
455 let a2 = SharedStringArena::create(&p, 4096).unwrap();
456 assert_eq!(a2.get(r).unwrap(), "held", "attach lost an interned string");
457 assert!(matches!(
458 SharedStringArena::create(&p, 2048),
459 Err(ArenaError::LayoutMismatch),
460 ));
461
462 drop(a);
465 drop(a2);
466 let fresh = SharedStringArena::reset(&p, 4096).unwrap();
467 assert_eq!(fresh.used_bytes(), 0, "reset kept interned bytes");
468 drop(fresh);
469 std::fs::remove_file(&p).ok();
470 }
471
472 #[test]
473 fn a_read_only_arena_resolves_refs_and_refuses_interning() {
474 let p = tmp("readonly");
475 let r = {
476 let w = SharedStringArena::create(&p, 4096).unwrap();
477 let r = w.intern("notepad.exe").unwrap();
478 w.flush().unwrap();
479 r
480 };
481 let ro = SharedStringArena::open_read_only(&p, 4096).unwrap();
482 assert!(!ro.is_writable());
483 assert_eq!(ro.get(r), Ok("notepad.exe"));
484 assert_eq!(ro.get_bytes(r), Ok(&b"notepad.exe"[..]));
485 assert_eq!(ro.used_bytes(), 11);
486 assert_eq!(ro.intern("more"), Err(ArenaError::ReadOnly));
487 assert_eq!(ro.intern_bytes(b"more"), Err(ArenaError::ReadOnly));
488 ro.clear();
489 assert_eq!(ro.used_bytes(), 11, "clear on a read-only arena is inert");
490 ro.flush().unwrap();
491 std::fs::remove_file(&p).ok();
492 }
493
494 #[test]
495 fn a_read_only_open_still_validates_the_header() {
496 let p = tmp("readonly-mismatch");
497 {
498 let w = SharedStringArena::create(&p, 4096).unwrap();
499 w.intern("x").unwrap();
500 w.flush().unwrap();
501 }
502 assert_eq!(
503 SharedStringArena::open_read_only(&p, 2048).err(),
504 Some(ArenaError::LayoutMismatch)
505 );
506 std::fs::remove_file(&p).ok();
507 }
508
509 #[test]
510 fn create_initial_state_is_empty() {
511 let p = tmp("init");
512 let a = SharedStringArena::create(&p, 1024).unwrap();
513 assert_eq!(a.capacity_bytes(), 1024);
514 assert_eq!(a.used_bytes(), 0);
515 assert_eq!(a.remaining_bytes(), 1024);
516 std::fs::remove_file(&p).ok();
517 }
518
519 #[test]
520 fn intern_and_get_round_trip() {
521 let p = tmp("rt");
522 let a = SharedStringArena::create(&p, 1024).unwrap();
523 let r1 = a.intern("hello").unwrap();
524 let r2 = a.intern("world").unwrap();
525 assert_eq!(a.get(r1).unwrap(), "hello");
526 assert_eq!(a.get(r2).unwrap(), "world");
527 assert_eq!(a.used_bytes(), 10);
528 std::fs::remove_file(&p).ok();
529 }
530
531 #[test]
532 fn empty_string_interns_with_zero_len() {
533 let p = tmp("empty");
534 let a = SharedStringArena::create(&p, 16).unwrap();
535 let r = a.intern("").unwrap();
536 assert_eq!(r.len, 0);
537 assert_eq!(a.get(r).unwrap(), "");
538 assert_eq!(a.used_bytes(), 0);
539 std::fs::remove_file(&p).ok();
540 }
541
542 #[test]
543 fn full_arena_returns_error() {
544 let p = tmp("full");
545 let a = SharedStringArena::create(&p, 10).unwrap();
546 a.intern("hello").unwrap();
547 a.intern("world").unwrap();
548 assert_eq!(a.intern("more").err(), Some(ArenaError::Full));
549 assert_eq!(a.used_bytes(), 10);
551 std::fs::remove_file(&p).ok();
552 }
553
554 #[test]
555 fn string_too_large_returns_full() {
556 let p = tmp("too-large");
557 let a = SharedStringArena::create(&p, 8).unwrap();
558 let big = "x".repeat(100);
559 assert_eq!(a.intern(&big).err(), Some(ArenaError::Full));
560 assert_eq!(a.used_bytes(), 0);
561 std::fs::remove_file(&p).ok();
562 }
563
564 #[test]
565 fn string_ref_packs_and_unpacks() {
566 let r = StringRef { offset: 0x1234_5678, len: 42 };
567 let packed = r.to_u64();
568 let unpacked = StringRef::from_u64(packed);
569 assert_eq!(unpacked, r);
570 }
571
572 #[test]
573 fn cross_handle_visibility() {
574 let p = tmp("cross-handle");
575 let writer = SharedStringArena::create(&p, 1024).unwrap();
576 let reader = SharedStringArena::open(&p, 1024).unwrap();
577 let r = writer.intern("cross-process").unwrap();
578 assert_eq!(reader.get(r).unwrap(), "cross-process");
579 std::fs::remove_file(&p).ok();
580 }
581
582 #[test]
583 fn invalid_ref_beyond_used_rejected() {
584 let p = tmp("invalid");
585 let a = SharedStringArena::create(&p, 1024).unwrap();
586 a.intern("hi").unwrap(); let bad = StringRef { offset: 100, len: 5 };
588 assert_eq!(a.get(bad).err(), Some(ArenaError::InvalidRef));
589 std::fs::remove_file(&p).ok();
590 }
591
592 #[test]
593 fn concurrent_interners_get_distinct_refs() {
594 let p = tmp("concurrent");
595 let a: Arc<SharedStringArena> = Arc::new(SharedStringArena::create(&p, 4096).unwrap());
596 let n_threads = 4;
597 let per_thread = 20;
598 let mut handles = vec![];
599 for t in 0..n_threads {
600 let a = a.clone();
601 handles.push(thread::spawn(move || {
602 let mut refs = vec![];
603 for i in 0..per_thread {
604 let s = format!("thread-{t}-msg-{i:03}");
605 let r = a.intern(&s).unwrap();
606 refs.push((s, r));
607 }
608 refs
609 }));
610 }
611 let all: Vec<(String, StringRef)> = handles.into_iter()
612 .flat_map(|h| h.join().unwrap())
613 .collect();
614 for (expected, r) in &all {
616 let got = a.get(*r).unwrap();
617 assert_eq!(got, expected,
618 "ref offset={} len={} should resolve to {expected}",
619 r.offset, r.len);
620 }
621 let mut refs: Vec<StringRef> = all.iter().map(|(_, r)| *r).collect();
623 refs.sort_by_key(|r| r.offset);
624 for w in refs.windows(2) {
625 let r1_end = w[0].offset + w[0].len;
626 assert!(r1_end <= w[1].offset,
627 "ref {:?} overlaps with ref {:?}", w[0], w[1]);
628 }
629 std::fs::remove_file(&p).ok();
630 }
631
632 #[test]
633 fn intern_and_get_helper_returns_both() {
634 let p = tmp("intern-and-get");
635 let a = SharedStringArena::create(&p, 1024).unwrap();
636 let (r, s) = a.intern_and_get("composite").unwrap();
637 assert_eq!(s, "composite");
638 assert_eq!(a.get(r).unwrap(), "composite");
639 std::fs::remove_file(&p).ok();
640 }
641
642 #[test]
643 fn utf8_validation_on_get() {
644 let p = tmp("utf8");
645 let a = SharedStringArena::create(&p, 128).unwrap();
646 let r = a.intern("hello").unwrap();
648 assert!(a.get(r).is_ok());
649 let r2 = a.intern_bytes(&[0xFF, 0xFE, 0xFD]).unwrap();
653 assert_eq!(a.get(r2).err(), Some(ArenaError::InvalidUtf8));
654 assert_eq!(a.get_bytes(r2).unwrap(), &[0xFF, 0xFE, 0xFD]);
655 std::fs::remove_file(&p).ok();
656 }
657
658 #[test]
659 fn clear_resets_used_bytes() {
660 let p = tmp("clear");
661 let a = SharedStringArena::create(&p, 128).unwrap();
662 a.intern("first").unwrap();
663 a.intern("second").unwrap();
664 assert!(a.used_bytes() > 0);
665 a.clear();
666 assert_eq!(a.used_bytes(), 0);
667 let r = a.intern("after-clear").unwrap();
669 assert_eq!(a.get(r).unwrap(), "after-clear");
670 assert_eq!(r.offset, 0);
671 std::fs::remove_file(&p).ok();
672 }
673
674 #[test]
675 fn disk_persistence_survives_reopen() {
676 let p = tmp("disk");
677 let r_persist;
678 {
679 let a = SharedStringArena::create(&p, 1024).unwrap();
680 r_persist = a.intern("persisted-string").unwrap();
681 a.flush().unwrap();
682 }
683 let a2 = SharedStringArena::open(&p, 1024).unwrap();
684 assert_eq!(a2.get(r_persist).unwrap(), "persisted-string");
685 let r2 = a2.intern("more-after-reopen").unwrap();
687 assert_eq!(a2.get(r2).unwrap(), "more-after-reopen");
688 std::fs::remove_file(&p).ok();
689 }
690
691 #[test]
692 fn deduplication_via_hashmap_composition() {
693 use crate::SharedHashMap;
696 use crate::shared_hash_map::fnv1a_64;
697
698 let p_arena = tmp("dedup-arena");
699 let p_index = tmp("dedup-index");
700 let arena = SharedStringArena::create(&p_arena, 256).unwrap();
701 let index: SharedHashMap<u64, u64> = SharedHashMap::create(&p_index, 32).unwrap();
702
703 let s = "deduplicate-me";
704 let h = fnv1a_64(s.as_bytes());
705
706 let r = if let Some(packed) = index.get(&h) {
708 StringRef::from_u64(packed)
709 } else {
710 let r = arena.intern(s).unwrap();
711 index.insert(h, r.to_u64()).unwrap();
712 r
713 };
714 let used_after_first = arena.used_bytes();
715
716 let r2 = if let Some(packed) = index.get(&h) {
718 StringRef::from_u64(packed)
719 } else {
720 let r = arena.intern(s).unwrap();
721 index.insert(h, r.to_u64()).unwrap();
722 r
723 };
724 assert_eq!(r, r2, "dedup should return the same ref");
725 assert_eq!(arena.used_bytes(), used_after_first,
726 "second intern should not consume more bytes");
727
728 std::fs::remove_file(&p_arena).ok();
729 std::fs::remove_file(&p_index).ok();
730 }
731}