1use std::fs::{File, OpenOptions};
41use std::mem::size_of;
42use std::path::Path;
43use std::sync::atomic::{AtomicU64, Ordering};
44
45use memmap2::{MmapMut, MmapOptions};
46
47pub const BITVEC_MAGIC: u64 = 0x4150_4256_4543_2031;
48pub const BITS_PER_WORD: usize = 64;
49
50#[repr(C, align(64))]
51pub struct BitVecHeader {
52 pub magic: u64,
53 pub capacity_bits: u64,
54 pub word_count: u64,
55 _pad: [u8; 40],
56}
57
58const _: () = {
59 assert!(size_of::<BitVecHeader>() == 64);
60};
61
62pub const fn bit_vec_file_size(capacity_bits: usize) -> usize {
63 let word_count = capacity_bits.div_ceil(BITS_PER_WORD);
64 size_of::<BitVecHeader>() + word_count * size_of::<AtomicU64>()
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq)]
68pub enum BitVecError {
69 OutOfBounds,
70 LayoutMismatch,
71 IoError(std::io::ErrorKind),
72}
73
74impl From<std::io::Error> for BitVecError {
75 fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
76}
77
78pub struct SharedBitVec {
79 _file: File,
80 mmap: MmapMut,
81 capacity_bits: usize,
82 word_count: usize,
83 header_sidecar: subetha_core::HandshakeHeader,
84 ring_sidecar: Box<subetha_core::ObservationRing>,
85}
86
87unsafe impl Send for SharedBitVec {}
88unsafe impl Sync for SharedBitVec {}
89
90impl subetha_sidecar::AdaptiveInstance for SharedBitVec {
91 fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
92 fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
93 fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
94 Box::new(subetha_sidecar::NoMigrationPolicy)
95 }
96}
97
98impl SharedBitVec {
99 pub fn create(
105 path: impl AsRef<Path>, capacity_bits: usize,
106 ) -> Result<Self, BitVecError> {
107 assert!(capacity_bits >= 1);
108 let (file, mmap) = crate::mmf_attach::create_or_attach(
109 path.as_ref(),
110 bit_vec_file_size(capacity_bits),
111 |ptr| unsafe { Self::init_region(ptr, capacity_bits) },
112 |ptr| unsafe { (*(ptr as *const BitVecHeader)).magic == BITVEC_MAGIC },
113 )?;
114 Self::from_region(file, mmap, capacity_bits)
115 }
116
117 pub fn reset(
121 path: impl AsRef<Path>, capacity_bits: usize,
122 ) -> Result<Self, BitVecError> {
123 assert!(capacity_bits >= 1);
124 let (file, mmap) = crate::mmf_attach::reset(
125 path.as_ref(),
126 bit_vec_file_size(capacity_bits),
127 |ptr| unsafe { Self::init_region(ptr, capacity_bits) },
128 )?;
129 Self::from_region(file, mmap, capacity_bits)
130 }
131
132 unsafe fn init_region(ptr: *mut u8, capacity_bits: usize) {
140 let hdr = ptr as *mut BitVecHeader;
141 unsafe {
142 (*hdr).capacity_bits = capacity_bits as u64;
143 (*hdr).word_count = capacity_bits.div_ceil(BITS_PER_WORD) as u64;
144 std::ptr::write_volatile(&raw mut (*hdr).magic, BITVEC_MAGIC);
145 }
146 }
147
148 fn from_region(
151 file: File,
152 mmap: MmapMut,
153 capacity_bits: usize,
154 ) -> Result<Self, BitVecError> {
155 let hdr = unsafe { &*(mmap.as_ptr() as *const BitVecHeader) };
156 if hdr.magic != BITVEC_MAGIC || hdr.capacity_bits != capacity_bits as u64 {
157 return Err(BitVecError::LayoutMismatch);
158 }
159 let word_count = hdr.word_count as usize;
160 Ok(Self {
161 _file: file, mmap, capacity_bits, word_count,
162 header_sidecar: subetha_core::HandshakeHeader::new(),
163 ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
164 })
165 }
166
167 pub fn open(
168 path: impl AsRef<Path>, expected_capacity_bits: usize,
169 ) -> Result<Self, BitVecError> {
170 let total = bit_vec_file_size(expected_capacity_bits);
171 let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
172 if file.metadata()?.len() < total as u64 {
173 return Err(BitVecError::LayoutMismatch);
174 }
175 let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
176 Self::from_region(file, mmap, expected_capacity_bits)
177 }
178
179 #[inline]
180 pub fn capacity_bits(&self) -> usize { self.capacity_bits }
181
182 #[inline]
183 pub fn capacity_words(&self) -> usize { self.word_count }
184
185 fn word(&self, word_idx: usize) -> &AtomicU64 {
186 let base = unsafe { self.mmap.as_ptr().add(size_of::<BitVecHeader>()) };
187 unsafe { &*(base.add(word_idx * size_of::<AtomicU64>()) as *const AtomicU64) }
188 }
189
190 #[inline]
191 fn check_bounds(&self, bit: usize) -> Result<(), BitVecError> {
192 if bit >= self.capacity_bits { Err(BitVecError::OutOfBounds) } else { Ok(()) }
193 }
194
195 pub fn set(&self, i: usize) -> Result<bool, BitVecError> {
197 if let Err(e) = self.check_bounds(i) {
198 self.ring_sidecar
199 .push_op(crate::sidecar_ops::bit_vec::OP_SET, 1);
200 return Err(e);
201 }
202 let (w, b) = (i / BITS_PER_WORD, i % BITS_PER_WORD);
203 let mask = 1u64 << b;
204 let prev = self.word(w).fetch_or(mask, Ordering::AcqRel);
205 self.ring_sidecar
206 .push_op(crate::sidecar_ops::bit_vec::OP_SET, 0);
207 Ok((prev & mask) != 0)
208 }
209
210 pub fn clear(&self, i: usize) -> Result<bool, BitVecError> {
212 if let Err(e) = self.check_bounds(i) {
213 self.ring_sidecar
214 .push_op(crate::sidecar_ops::bit_vec::OP_CLEAR, 1);
215 return Err(e);
216 }
217 let (w, b) = (i / BITS_PER_WORD, i % BITS_PER_WORD);
218 let mask = 1u64 << b;
219 let prev = self.word(w).fetch_and(!mask, Ordering::AcqRel);
220 self.ring_sidecar
221 .push_op(crate::sidecar_ops::bit_vec::OP_CLEAR, 0);
222 Ok((prev & mask) != 0)
223 }
224
225 pub fn toggle(&self, i: usize) -> Result<bool, BitVecError> {
227 if let Err(e) = self.check_bounds(i) {
228 self.ring_sidecar
229 .push_op(crate::sidecar_ops::bit_vec::OP_TOGGLE, 1);
230 return Err(e);
231 }
232 let (w, b) = (i / BITS_PER_WORD, i % BITS_PER_WORD);
233 let mask = 1u64 << b;
234 let prev = self.word(w).fetch_xor(mask, Ordering::AcqRel);
235 self.ring_sidecar
236 .push_op(crate::sidecar_ops::bit_vec::OP_TOGGLE, 0);
237 Ok((prev & mask) == 0)
239 }
240
241 pub fn get(&self, i: usize) -> Result<bool, BitVecError> {
243 if let Err(e) = self.check_bounds(i) {
244 self.ring_sidecar
245 .push_op(crate::sidecar_ops::bit_vec::OP_GET, 1);
246 return Err(e);
247 }
248 let (w, b) = (i / BITS_PER_WORD, i % BITS_PER_WORD);
249 let mask = 1u64 << b;
250 let v = (self.word(w).load(Ordering::Acquire) & mask) != 0;
251 self.ring_sidecar
252 .push_op(crate::sidecar_ops::bit_vec::OP_GET, 0);
253 Ok(v)
254 }
255
256 pub fn set_range(&self, lo: usize, hi: usize) -> Result<(), BitVecError> {
259 if lo > hi || hi > self.capacity_bits {
260 self.ring_sidecar
261 .push_op(crate::sidecar_ops::bit_vec::OP_RANGE, 1);
262 return Err(BitVecError::OutOfBounds);
263 }
264 if lo != hi { self.range_op(lo, hi, RangeOp::Set); }
265 self.ring_sidecar
266 .push_op(crate::sidecar_ops::bit_vec::OP_RANGE, 0);
267 Ok(())
268 }
269
270 pub fn clear_range(&self, lo: usize, hi: usize) -> Result<(), BitVecError> {
273 if lo > hi || hi > self.capacity_bits {
274 self.ring_sidecar
275 .push_op(crate::sidecar_ops::bit_vec::OP_RANGE, 1);
276 return Err(BitVecError::OutOfBounds);
277 }
278 if lo != hi { self.range_op(lo, hi, RangeOp::Clear); }
279 self.ring_sidecar
280 .push_op(crate::sidecar_ops::bit_vec::OP_RANGE, 0);
281 Ok(())
282 }
283
284 fn range_op(&self, lo: usize, hi: usize, op: RangeOp) {
285 let lo_word = lo / BITS_PER_WORD;
286 let hi_word = (hi - 1) / BITS_PER_WORD; let lo_bit = lo % BITS_PER_WORD;
288 let hi_bit_excl = hi - hi_word * BITS_PER_WORD;
289 if lo_word == hi_word {
290 let count = hi - lo;
292 let mask = if count == BITS_PER_WORD { u64::MAX }
293 else { ((1u64 << count) - 1) << lo_bit };
294 match op {
295 RangeOp::Set => { self.word(lo_word).fetch_or(mask, Ordering::AcqRel); }
296 RangeOp::Clear => { self.word(lo_word).fetch_and(!mask, Ordering::AcqRel); }
297 }
298 return;
299 }
300 let lo_mask = !((1u64 << lo_bit) - 1);
302 match op {
303 RangeOp::Set => { self.word(lo_word).fetch_or(lo_mask, Ordering::AcqRel); }
304 RangeOp::Clear => { self.word(lo_word).fetch_and(!lo_mask, Ordering::AcqRel); }
305 }
306 for w in (lo_word + 1)..hi_word {
308 let v = match op {
309 RangeOp::Set => u64::MAX,
310 RangeOp::Clear => 0,
311 };
312 self.word(w).store(v, Ordering::Release);
313 }
314 let hi_mask = if hi_bit_excl == BITS_PER_WORD { u64::MAX }
316 else { (1u64 << hi_bit_excl) - 1 };
317 match op {
318 RangeOp::Set => { self.word(hi_word).fetch_or(hi_mask, Ordering::AcqRel); }
319 RangeOp::Clear => { self.word(hi_word).fetch_and(!hi_mask, Ordering::AcqRel); }
320 }
321 }
322
323 pub fn count_ones(&self) -> usize {
325 let mut sum = 0usize;
326 for w in 0..self.word_count {
327 sum += self.word(w).load(Ordering::Acquire).count_ones() as usize;
328 }
329 self.ring_sidecar
330 .push_op(crate::sidecar_ops::bit_vec::OP_COUNT_ONES, 0);
331 sum
335 }
336
337 pub fn count_zeros(&self) -> usize {
339 self.capacity_bits.saturating_sub(self.count_ones())
340 }
341
342 pub fn is_all_set(&self) -> bool { self.count_ones() == self.capacity_bits }
343 pub fn is_all_clear(&self) -> bool { self.count_ones() == 0 }
344
345 pub fn set_all(&self) {
349 if self.word_count == 0 { return; }
350 let full_words = self.word_count - 1;
351 for w in 0..full_words {
352 self.word(w).store(u64::MAX, Ordering::Release);
353 }
354 let last_word_bits = self.capacity_bits - full_words * BITS_PER_WORD;
356 let last_mask = if last_word_bits == BITS_PER_WORD { u64::MAX }
357 else { (1u64 << last_word_bits) - 1 };
358 self.word(full_words).store(last_mask, Ordering::Release);
359 }
360
361 pub fn clear_all(&self) {
363 for w in 0..self.word_count {
364 self.word(w).store(0, Ordering::Release);
365 }
366 }
367
368 pub fn flush(&self) -> Result<(), BitVecError> {
369 self.mmap.flush()?;
370 Ok(())
371 }
372
373 pub fn flush_async(&self) -> Result<(), BitVecError> {
374 self.mmap.flush_async()?;
375 Ok(())
376 }
377}
378
379#[derive(Clone, Copy)]
380enum RangeOp { Set, Clear }
381
382#[cfg(test)]
383mod tests {
384 use super::*;
385 use std::sync::Arc;
386 use std::thread;
387
388 fn tmp(name: &str) -> std::path::PathBuf {
389 let mut p = std::env::temp_dir();
390 let pid = std::process::id();
391 p.push(format!("subetha-bitvec-{name}-{pid}.bin"));
392 p
393 }
394
395 #[test]
398 fn second_create_attaches_and_keeps_bits() {
399 let p = tmp("attach");
400 std::fs::remove_file(&p).ok();
401 let b = SharedBitVec::create(&p, 100).unwrap();
402 b.set(42).unwrap();
403
404 let b2 = SharedBitVec::create(&p, 100).unwrap();
405 assert!(b2.get(42).unwrap(), "attach cleared a live bit");
406 assert!(matches!(
407 SharedBitVec::create(&p, 50),
408 Err(BitVecError::LayoutMismatch),
409 ));
410
411 drop(b);
414 drop(b2);
415 let fresh = SharedBitVec::reset(&p, 100).unwrap();
416 assert!(fresh.is_all_clear(), "reset left a bit set");
417 drop(fresh);
418 std::fs::remove_file(&p).ok();
419 }
420
421 #[test]
422 fn create_initial_all_zero() {
423 let p = tmp("init");
424 let b = SharedBitVec::create(&p, 100).unwrap();
425 assert_eq!(b.capacity_bits(), 100);
426 assert_eq!(b.count_ones(), 0);
427 assert!(b.is_all_clear());
428 for i in 0..100 {
429 assert!(!b.get(i).unwrap(), "bit {i} should start clear");
430 }
431 std::fs::remove_file(&p).ok();
432 }
433
434 #[test]
435 fn set_get_round_trip() {
436 let p = tmp("rt");
437 let b = SharedBitVec::create(&p, 200).unwrap();
438 assert!(!b.set(5).unwrap());
439 assert!(!b.set(63).unwrap());
440 assert!(!b.set(64).unwrap());
441 assert!(!b.set(199).unwrap());
442 assert!(b.get(5).unwrap());
443 assert!(b.get(63).unwrap());
444 assert!(b.get(64).unwrap());
445 assert!(b.get(199).unwrap());
446 assert!(!b.get(6).unwrap());
447 assert!(!b.get(0).unwrap());
448 assert!(b.set(5).unwrap());
450 std::fs::remove_file(&p).ok();
451 }
452
453 #[test]
454 fn clear_works() {
455 let p = tmp("clear");
456 let b = SharedBitVec::create(&p, 64).unwrap();
457 b.set(10).unwrap();
458 b.set(20).unwrap();
459 assert!(b.clear(10).unwrap());
460 assert!(!b.get(10).unwrap());
461 assert!(b.get(20).unwrap());
462 assert!(!b.clear(0).unwrap());
464 std::fs::remove_file(&p).ok();
465 }
466
467 #[test]
468 fn toggle_alternates() {
469 let p = tmp("toggle");
470 let b = SharedBitVec::create(&p, 8).unwrap();
471 assert!(b.toggle(3).unwrap()); assert!(b.get(3).unwrap());
473 assert!(!b.toggle(3).unwrap()); assert!(!b.get(3).unwrap());
475 std::fs::remove_file(&p).ok();
476 }
477
478 #[test]
479 fn count_ones_accurate() {
480 let p = tmp("count");
481 let b = SharedBitVec::create(&p, 200).unwrap();
482 for i in [0, 7, 31, 63, 64, 100, 199] {
483 b.set(i).unwrap();
484 }
485 assert_eq!(b.count_ones(), 7);
486 assert_eq!(b.count_zeros(), 200 - 7);
487 std::fs::remove_file(&p).ok();
488 }
489
490 #[test]
491 fn set_range_single_word() {
492 let p = tmp("range-single");
493 let b = SharedBitVec::create(&p, 64).unwrap();
494 b.set_range(10, 20).unwrap();
495 for i in 0..64 {
496 let expected = (10..20).contains(&i);
497 assert_eq!(b.get(i).unwrap(), expected, "bit {i}");
498 }
499 std::fs::remove_file(&p).ok();
500 }
501
502 #[test]
503 fn set_range_spans_multiple_words() {
504 let p = tmp("range-multi");
505 let b = SharedBitVec::create(&p, 200).unwrap();
506 b.set_range(50, 150).unwrap();
507 for i in 0..200 {
508 let expected = (50..150).contains(&i);
509 assert_eq!(b.get(i).unwrap(), expected, "bit {i}");
510 }
511 assert_eq!(b.count_ones(), 100);
512 std::fs::remove_file(&p).ok();
513 }
514
515 #[test]
516 fn clear_range_after_set_all() {
517 let p = tmp("clear-range");
518 let b = SharedBitVec::create(&p, 200).unwrap();
519 b.set_all();
520 assert!(b.is_all_set());
521 b.clear_range(70, 130).unwrap();
522 for i in 0..200 {
523 let expected = !(70..130).contains(&i);
524 assert_eq!(b.get(i).unwrap(), expected, "bit {i}");
525 }
526 std::fs::remove_file(&p).ok();
527 }
528
529 #[test]
530 fn out_of_bounds_rejected() {
531 let p = tmp("oob");
532 let b = SharedBitVec::create(&p, 8).unwrap();
533 assert_eq!(b.set(8).err(), Some(BitVecError::OutOfBounds));
534 assert_eq!(b.set(999).err(), Some(BitVecError::OutOfBounds));
535 assert_eq!(b.get(8).err(), Some(BitVecError::OutOfBounds));
536 assert_eq!(b.set_range(0, 9).err(), Some(BitVecError::OutOfBounds));
537 std::fs::remove_file(&p).ok();
538 }
539
540 #[test]
541 fn set_all_and_clear_all() {
542 let p = tmp("all");
543 let b = SharedBitVec::create(&p, 130).unwrap();
544 b.set_all();
545 assert!(b.is_all_set());
546 for i in 0..130 { assert!(b.get(i).unwrap()); }
547 b.clear_all();
548 assert!(b.is_all_clear());
549 for i in 0..130 { assert!(!b.get(i).unwrap()); }
550 std::fs::remove_file(&p).ok();
551 }
552
553 #[test]
554 fn cross_handle_visibility() {
555 let p = tmp("cross-handle");
556 let writer = SharedBitVec::create(&p, 100).unwrap();
557 let reader = SharedBitVec::open(&p, 100).unwrap();
558 writer.set(42).unwrap();
559 writer.set(77).unwrap();
560 assert!(reader.get(42).unwrap());
561 assert!(reader.get(77).unwrap());
562 assert!(!reader.get(0).unwrap());
563 reader.clear(42).unwrap();
564 assert!(!writer.get(42).unwrap());
565 std::fs::remove_file(&p).ok();
566 }
567
568 #[test]
569 fn concurrent_setters_of_disjoint_bits_all_visible() {
570 let p = tmp("concurrent-disjoint");
572 let b = Arc::new(SharedBitVec::create(&p, 1000).unwrap());
573 let n_threads = 4;
574 let per_thread = 100;
575 let mut handles = vec![];
576 for t in 0..n_threads {
577 let b = b.clone();
578 handles.push(thread::spawn(move || {
579 for i in 0..per_thread {
580 let bit = t * per_thread + i;
581 b.set(bit).unwrap();
582 }
583 }));
584 }
585 for h in handles { h.join().unwrap(); }
586 assert_eq!(b.count_ones(), n_threads * per_thread);
587 for t in 0..n_threads {
588 for i in 0..per_thread {
589 assert!(b.get(t * per_thread + i).unwrap());
590 }
591 }
592 std::fs::remove_file(&p).ok();
593 }
594
595 #[test]
596 fn concurrent_setters_of_same_word_distinct_bits_all_visible() {
597 let p = tmp("concurrent-same-word");
600 let b = Arc::new(SharedBitVec::create(&p, 64).unwrap());
601 let mut handles = vec![];
602 for bit in 0..64 {
603 let b = b.clone();
604 handles.push(thread::spawn(move || {
605 b.set(bit).unwrap();
606 }));
607 }
608 for h in handles { h.join().unwrap(); }
609 assert!(b.is_all_set(), "every bit in the word should be set");
610 std::fs::remove_file(&p).ok();
611 }
612
613 #[test]
614 fn concurrent_setters_of_same_bit_idempotent() {
615 let p = tmp("concurrent-same-bit");
617 let b = Arc::new(SharedBitVec::create(&p, 100).unwrap());
618 let mut handles = vec![];
619 for _ in 0..16 {
620 let b = b.clone();
621 handles.push(thread::spawn(move || b.set(42).unwrap()));
622 }
623 let prev_values: Vec<bool> = handles.into_iter()
624 .map(|h| h.join().unwrap()).collect();
625 let false_count = prev_values.iter().filter(|&&v| !v).count();
627 let true_count = prev_values.iter().filter(|&&v| v).count();
628 assert_eq!(false_count, 1, "exactly one setter should see prev=false");
629 assert_eq!(true_count, 15);
630 assert!(b.get(42).unwrap());
631 assert_eq!(b.count_ones(), 1);
632 std::fs::remove_file(&p).ok();
633 }
634
635 #[test]
636 fn disk_persistence_survives_reopen() {
637 let p = tmp("disk");
638 {
639 let b = SharedBitVec::create(&p, 200).unwrap();
640 for i in [5, 50, 100, 150, 199] { b.set(i).unwrap(); }
641 b.flush().unwrap();
642 }
643 let b2 = SharedBitVec::open(&p, 200).unwrap();
644 assert_eq!(b2.count_ones(), 5);
645 for i in [5, 50, 100, 150, 199] { assert!(b2.get(i).unwrap()); }
646 for i in [0, 4, 6, 49, 51] { assert!(!b2.get(i).unwrap()); }
647 std::fs::remove_file(&p).ok();
648 }
649
650 #[test]
651 fn capacity_words_matches_packed_count() {
652 let p = tmp("words");
653 let b1 = SharedBitVec::create(&p, 64).unwrap();
654 assert_eq!(b1.capacity_words(), 1);
655 drop(b1);
656 std::fs::remove_file(&p).ok();
657
658 let p2 = tmp("words2");
659 let b2 = SharedBitVec::create(&p2, 65).unwrap();
660 assert_eq!(b2.capacity_words(), 2);
661 drop(b2);
662 std::fs::remove_file(&p2).ok();
663
664 let p3 = tmp("words3");
665 let b3 = SharedBitVec::create(&p3, 1000).unwrap();
666 assert_eq!(b3.capacity_words(), 16); drop(b3);
668 std::fs::remove_file(&p3).ok();
669 }
670
671 #[test]
672 fn allocation_bitmap_pattern() {
673 let p = tmp("alloc-pattern");
677 let b = SharedBitVec::create(&p, 1024).unwrap();
678 for i in 0..10 { assert!(!b.set(i).unwrap()); }
680 assert_eq!(b.count_ones(), 10);
681 assert!(b.clear(5).unwrap());
683 assert_eq!(b.count_ones(), 9);
684 assert!(!b.set(5).unwrap());
686 assert_eq!(b.count_ones(), 10);
687 std::fs::remove_file(&p).ok();
688 }
689}