subetha_cxc/
shared_bloom_filter.rs

1//! `SharedBloomFilter` - cross-process probabilistic set membership.
2//!
3//! Composite primitive: [`SharedBitVec`] +
4//! `k` hash functions. Insert hashes the input `k` times and sets
5//! those `k` bits; `contains` returns true if and only if all `k`
6//! bits are set. No false negatives; false positives are possible
7//! with a tunable rate.
8//!
9//! # Sizing rules of thumb
10//!
11//! For `n` distinct items and target false-positive rate `p`:
12//! - Optimal `n_bits` = `-(n * ln(p)) / (ln(2)^2)`
13//! - Optimal `n_hashes` = `(n_bits / n) * ln(2)`
14//!
15//! For example, n=10_000 items with p=0.01 (1% FPR):
16//! `n_bits ~= 95_851`, `n_hashes ~= 7`.
17//!
18//! Use [`suggest_config`](SharedBloomFilter::suggest_config) to
19//! compute these.
20//!
21//! # Cross-process angle
22//!
23//! Just a SharedBitVec wrapper. The underlying bit array is the
24//! shared state; n_bits and n_hashes are header-stored config so
25//! cross-handle opens verify they match.
26//!
27//! # Hash function: double-hashing FNV-1a
28//!
29//! We compute two FNV-1a hashes with different seeds, then derive
30//! the k hash positions via `(h1 + i * h2) mod n_bits` for
31//! i in 0..k. This is the standard Kirsch-Mitzenmacher
32//! double-hashing technique that gives k effectively-independent
33//! hash positions from only two underlying hash computations.
34
35use std::fs::{File, OpenOptions};
36use std::path::{Path, PathBuf};
37
38use memmap2::{MmapMut, MmapOptions};
39
40use crate::shared_bit_vec::{BitVecError, SharedBitVec};
41
42pub const BLOOM_MAGIC: u64 = 0x4150_424C_4F4F_4D31;
43
44#[repr(C, align(64))]
45pub struct BloomHeader {
46    pub magic: u64,
47    pub n_bits: u64,
48    pub n_hashes: u32,
49    _pad: [u8; 44],
50}
51
52const _: () = {
53    assert!(std::mem::size_of::<BloomHeader>() == 64);
54};
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub enum BloomError {
58    BitVec(BitVecError),
59    LayoutMismatch,
60    InvalidConfig,
61    IoError(std::io::ErrorKind),
62}
63
64impl From<BitVecError> for BloomError {
65    fn from(e: BitVecError) -> Self { Self::BitVec(e) }
66}
67impl From<std::io::Error> for BloomError {
68    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
69}
70
71fn header_path(base: &Path) -> PathBuf {
72    let mut p = base.to_path_buf();
73    let stem = p.file_name().unwrap().to_string_lossy().to_string();
74    p.set_file_name(format!("{stem}.bloom.bin"));
75    p
76}
77fn bits_path(base: &Path) -> PathBuf {
78    let mut p = base.to_path_buf();
79    let stem = p.file_name().unwrap().to_string_lossy().to_string();
80    p.set_file_name(format!("{stem}.bits.bin"));
81    p
82}
83
84pub struct SharedBloomFilter {
85    _file: File,
86    _mmap: MmapMut,
87    bits: SharedBitVec,
88    n_bits: u64,
89    n_hashes: u32,
90    header_sidecar: subetha_core::HandshakeHeader,
91    ring_sidecar: Box<subetha_core::ObservationRing>,
92}
93
94unsafe impl Send for SharedBloomFilter {}
95unsafe impl Sync for SharedBloomFilter {}
96
97impl subetha_sidecar::AdaptiveInstance for SharedBloomFilter {
98    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
99    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
100    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
101        Box::new(subetha_sidecar::NoMigrationPolicy)
102    }
103}
104
105const FNV_OFFSET_BASIS_1: u64 = 0xcbf2_9ce4_8422_2325;
106const FNV_OFFSET_BASIS_2: u64 = 0x84222325_cbf29ce4;
107const FNV_PRIME: u64 = 0x100_0000_01b3;
108
109#[inline]
110fn fnv1a_seeded(bytes: &[u8], offset_basis: u64) -> u64 {
111    let mut h = offset_basis;
112    for &b in bytes {
113        h ^= b as u64;
114        h = h.wrapping_mul(FNV_PRIME);
115    }
116    h
117}
118
119/// MurmurHash3 64-bit finalizer (avalanche). FNV-1a alone has weak
120/// high-bit diffusion; the position mapping uses Lemire fastrange, which
121/// keys on the high bits, so the seed hashes are avalanched here first.
122/// This also tightens the achieved false-positive rate toward the
123/// configured target.
124#[inline]
125fn fmix64(mut h: u64) -> u64 {
126    h ^= h >> 33;
127    h = h.wrapping_mul(0xff51_afd7_ed55_8ccd);
128    h ^= h >> 33;
129    h = h.wrapping_mul(0xc4ce_b9fe_1a85_ec53);
130    h ^= h >> 33;
131    h
132}
133
134impl SharedBloomFilter {
135    /// Suggest n_bits and n_hashes for a target false-positive rate
136    /// `p` and expected item count `n`. Both rounded up.
137    pub fn suggest_config(n_items: usize, p: f64) -> (usize, u32) {
138        let n = n_items as f64;
139        let n_bits = (-n * p.ln() / (std::f64::consts::LN_2 * std::f64::consts::LN_2)).ceil() as usize;
140        let n_hashes = ((n_bits as f64 / n) * std::f64::consts::LN_2).round() as u32;
141        (n_bits.max(64), n_hashes.max(1))
142    }
143
144    /// Obtain the filter at `base_path`, initializing an empty one if
145    /// its files do not yet exist and attaching to them if they do.
146    /// Attaching leaves inserted members in place; a region built with
147    /// a different `n_bits` or `n_hashes` is a `LayoutMismatch`.
148    /// [`reset`](Self::reset) reinitializes.
149    pub fn create(
150        base_path: impl AsRef<Path>, n_bits: usize, n_hashes: u32,
151    ) -> Result<Self, BloomError> {
152        if n_bits == 0 || n_hashes == 0 {
153            return Err(BloomError::InvalidConfig);
154        }
155        let base = base_path.as_ref();
156        let (file, mmap) = crate::mmf_attach::create_or_attach(
157            &header_path(base),
158            std::mem::size_of::<BloomHeader>(),
159            |ptr| unsafe { Self::init_region(ptr, n_bits, n_hashes) },
160            |ptr| unsafe { (*(ptr as *const BloomHeader)).magic == BLOOM_MAGIC },
161        )?;
162        Self::check_header(&mmap, n_bits, n_hashes)?;
163        let bits = SharedBitVec::create(bits_path(base), n_bits)?;
164        Ok(Self::assemble(file, mmap, bits, n_bits, n_hashes))
165    }
166
167    /// Truncate both of the filter's files at `base_path` and
168    /// initialize an empty one, discarding every member live peers
169    /// share. For a caller that knows it owns the path.
170    pub fn reset(
171        base_path: impl AsRef<Path>, n_bits: usize, n_hashes: u32,
172    ) -> Result<Self, BloomError> {
173        if n_bits == 0 || n_hashes == 0 {
174            return Err(BloomError::InvalidConfig);
175        }
176        let base = base_path.as_ref();
177        let (file, mmap) = crate::mmf_attach::reset(
178            &header_path(base),
179            std::mem::size_of::<BloomHeader>(),
180            |ptr| unsafe { Self::init_region(ptr, n_bits, n_hashes) },
181        )?;
182        let bits = SharedBitVec::reset(bits_path(base), n_bits)?;
183        Ok(Self::assemble(file, mmap, bits, n_bits, n_hashes))
184    }
185
186    /// Lay out the config header: sizes first, magic last, because
187    /// attachers spin on it.
188    ///
189    /// # Safety
190    /// `ptr` addresses at least `size_of::<BloomHeader>()` writable
191    /// zeroed bytes.
192    unsafe fn init_region(ptr: *mut u8, n_bits: usize, n_hashes: u32) {
193        let hdr = ptr as *mut BloomHeader;
194        unsafe {
195            (*hdr).n_bits = n_bits as u64;
196            (*hdr).n_hashes = n_hashes;
197            std::ptr::write_volatile(&raw mut (*hdr).magic, BLOOM_MAGIC);
198        }
199    }
200
201    /// Refuse a header built with a different config.
202    fn check_header(mmap: &MmapMut, n_bits: usize, n_hashes: u32) -> Result<(), BloomError> {
203        let hdr = unsafe { &*(mmap.as_ptr() as *const BloomHeader) };
204        if hdr.magic != BLOOM_MAGIC
205            || hdr.n_bits != n_bits as u64
206            || hdr.n_hashes != n_hashes
207        {
208            return Err(BloomError::LayoutMismatch);
209        }
210        Ok(())
211    }
212
213    fn assemble(
214        file: File,
215        mmap: MmapMut,
216        bits: SharedBitVec,
217        n_bits: usize,
218        n_hashes: u32,
219    ) -> Self {
220        Self {
221            _file: file, _mmap: mmap, bits,
222            n_bits: n_bits as u64, n_hashes,
223            header_sidecar: subetha_core::HandshakeHeader::new(),
224            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
225        }
226    }
227
228    pub fn open(
229        base_path: impl AsRef<Path>, n_bits: usize, n_hashes: u32,
230    ) -> Result<Self, BloomError> {
231        let base = base_path.as_ref();
232        let hpath = header_path(base);
233        let file = OpenOptions::new().read(true).write(true).open(&hpath)?;
234        if file.metadata()?.len() < std::mem::size_of::<BloomHeader>() as u64 {
235            return Err(BloomError::LayoutMismatch);
236        }
237        let mmap = unsafe {
238            MmapOptions::new().len(std::mem::size_of::<BloomHeader>()).map_mut(&file)?
239        };
240        Self::check_header(&mmap, n_bits, n_hashes)?;
241        let bits = SharedBitVec::open(bits_path(base), n_bits)?;
242        Ok(Self::assemble(file, mmap, bits, n_bits, n_hashes))
243    }
244
245    #[inline]
246    pub fn n_bits(&self) -> u64 { self.n_bits }
247    #[inline]
248    pub fn n_hashes(&self) -> u32 { self.n_hashes }
249
250    /// Insert an item. Sets `n_hashes` bits in the underlying
251    /// vector. Idempotent: re-inserting the same item is a no-op
252    /// for membership purposes (the bits stay set).
253    /// Map the i-th Kirsch-Mitzenmacher double-hash to a bit index in
254    /// `[0, n_bits)` via Lemire's fastrange (multiply-shift) instead of
255    /// `% n_bits`: replaces a hardware DIV per hash with a multiply +
256    /// shift and needs no power-of-two `n_bits`. insert and contains
257    /// share this mapping, so membership stays consistent.
258    #[inline]
259    fn position(&self, h1: u64, h2: u64, i: u64) -> usize {
260        let h = h1.wrapping_add(i.wrapping_mul(h2));
261        ((h as u128 * self.n_bits as u128) >> 64) as usize
262    }
263
264    pub fn insert(&self, item: &[u8]) -> Result<(), BloomError> {
265        let h1 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_1));
266        let h2 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_2));
267        for i in 0..self.n_hashes as u64 {
268            self.bits.set(self.position(h1, h2, i))?;
269        }
270        self.ring_sidecar
271            .push_op(crate::sidecar_ops::sketch::OP_INSERT, 0);
272        Ok(())
273    }
274
275    /// True if `item` MIGHT be in the set; false if definitely not.
276    /// False positives are possible; false negatives are NOT.
277    pub fn contains(&self, item: &[u8]) -> Result<bool, BloomError> {
278        let h1 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_1));
279        let h2 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_2));
280        for i in 0..self.n_hashes as u64 {
281            if !self.bits.get(self.position(h1, h2, i))? {
282                self.ring_sidecar
283                    .push_op(crate::sidecar_ops::sketch::OP_QUERY, 2); // absent
284                return Ok(false);
285            }
286        }
287        self.ring_sidecar
288            .push_op(crate::sidecar_ops::sketch::OP_QUERY, 0);
289        Ok(true)
290    }
291
292    /// Clear all bits (resets the filter to empty).
293    pub fn clear(&self) {
294        self.bits.clear_all();
295        self.ring_sidecar
296            .push_op(crate::sidecar_ops::sketch::OP_CLEAR, 0);
297    }
298
299    /// Estimate the current false-positive rate from the bit-set
300    /// density. Returns 0.0 for an empty filter; approaches 1.0
301    /// as the filter saturates.
302    pub fn estimated_false_positive_rate(&self) -> f64 {
303        let fill = self.bits.count_ones() as f64;
304        let n = self.n_bits as f64;
305        let k = self.n_hashes as f64;
306        if fill == 0.0 { return 0.0; }
307        (fill / n).powf(k)
308    }
309
310    /// Estimate the number of distinct inserted items based on bit
311    /// density. Formula: `-n_bits / n_hashes * ln(1 - fill / n_bits)`.
312    pub fn estimated_insert_count(&self) -> u64 {
313        let fill = self.bits.count_ones() as f64;
314        let n = self.n_bits as f64;
315        let k = self.n_hashes as f64;
316        if fill == 0.0 { return 0; }
317        if fill >= n { return u64::MAX; }
318        ((-n / k) * (1.0 - fill / n).ln()).round() as u64
319    }
320
321    pub fn flush(&self) -> Result<(), BloomError> {
322        Ok(self.bits.flush()?)
323    }
324
325    pub fn flush_async(&self) -> Result<(), BloomError> {
326        Ok(self.bits.flush_async()?)
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333
334    fn tmp_base(name: &str) -> PathBuf {
335        let mut p = std::env::temp_dir();
336        let pid = std::process::id();
337        p.push(format!("subetha-bloom-{name}-{pid}"));
338        p
339    }
340
341    fn cleanup(base: &Path) {
342        std::fs::remove_file(header_path(base)).ok();
343        std::fs::remove_file(bits_path(base)).ok();
344    }
345
346    /// A second create attaches with inserted members in place; reset
347    /// is what strips them.
348    #[test]
349    fn second_create_attaches_and_keeps_members() {
350        let base = tmp_base("attach");
351        cleanup(&base);
352        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
353        b.insert(b"member").unwrap();
354
355        let b2 = SharedBloomFilter::create(&base, 1024, 3).unwrap();
356        assert!(b2.contains(b"member").unwrap(), "attach lost a member");
357        assert!(matches!(
358            SharedBloomFilter::create(&base, 512, 3),
359            Err(BloomError::LayoutMismatch),
360        ));
361
362        // Windows refuses to truncate a mapped file, so every handle goes
363        // before the reset.
364        drop(b);
365        drop(b2);
366        let fresh = SharedBloomFilter::reset(&base, 1024, 3).unwrap();
367        assert!(!fresh.contains(b"member").unwrap(), "reset kept a member");
368        drop(fresh);
369        cleanup(&base);
370    }
371
372    #[test]
373    fn empty_filter_contains_nothing() {
374        let base = tmp_base("empty");
375        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
376        assert!(!b.contains(b"anything").unwrap());
377        assert!(!b.contains(b"").unwrap());
378        cleanup(&base);
379    }
380
381    #[test]
382    fn insert_then_contains_returns_true_no_false_negatives() {
383        let base = tmp_base("no-fn");
384        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
385        let items: &[&[u8]] = &[
386            b"hello", b"world", b"adaptive-prims", b"42",
387            b"the quick brown fox", b"", b"a",
388        ];
389        for &item in items {
390            b.insert(item).unwrap();
391        }
392        for &item in items {
393            assert!(b.contains(item).unwrap(),
394                "false negative for item: {item:?}");
395        }
396        cleanup(&base);
397    }
398
399    #[test]
400    fn invalid_config_rejected() {
401        let base = tmp_base("invalid");
402        assert_eq!(
403            SharedBloomFilter::create(&base, 0, 3).err(),
404            Some(BloomError::InvalidConfig)
405        );
406        assert_eq!(
407            SharedBloomFilter::create(&base, 1024, 0).err(),
408            Some(BloomError::InvalidConfig)
409        );
410        cleanup(&base);
411    }
412
413    #[test]
414    fn false_positive_rate_within_bounds_at_target_load() {
415        let (n_bits, n_hashes) = SharedBloomFilter::suggest_config(1000, 0.01);
416        let base = tmp_base("fpr");
417        let b = SharedBloomFilter::create(&base, n_bits, n_hashes).unwrap();
418        for i in 0..1000u32 {
419            b.insert(format!("item-{i:04}").as_bytes()).unwrap();
420        }
421        let mut fp = 0u32;
422        for i in 10_000u32..20_000 {
423            if b.contains(format!("query-{i:05}").as_bytes()).unwrap() {
424                fp += 1;
425            }
426        }
427        let observed_fpr = fp as f64 / 10_000.0;
428        // Target was 0.01; allow stochastic slack up to 0.03.
429        assert!(observed_fpr < 0.03,
430            "observed FPR {observed_fpr} should be below 0.03 (target was 0.01)");
431        cleanup(&base);
432    }
433
434    #[test]
435    fn clear_resets_filter() {
436        let base = tmp_base("clear");
437        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
438        b.insert(b"foo").unwrap();
439        b.insert(b"bar").unwrap();
440        assert!(b.contains(b"foo").unwrap());
441        b.clear();
442        assert!(!b.contains(b"foo").unwrap());
443        assert!(!b.contains(b"bar").unwrap());
444        cleanup(&base);
445    }
446
447    #[test]
448    fn cross_handle_visibility() {
449        let base = tmp_base("cross-handle");
450        let writer = SharedBloomFilter::create(&base, 1024, 3).unwrap();
451        let reader = SharedBloomFilter::open(&base, 1024, 3).unwrap();
452        writer.insert(b"cross-process").unwrap();
453        assert!(reader.contains(b"cross-process").unwrap());
454        assert!(!reader.contains(b"not-inserted").unwrap());
455        cleanup(&base);
456    }
457
458    #[test]
459    fn config_mismatch_at_open_rejected() {
460        let base = tmp_base("mismatch");
461        let _w = SharedBloomFilter::create(&base, 1024, 3).unwrap();
462        assert!(matches!(
463            SharedBloomFilter::open(&base, 2048, 3),
464            Err(BloomError::LayoutMismatch)
465        ));
466        assert!(matches!(
467            SharedBloomFilter::open(&base, 1024, 5),
468            Err(BloomError::LayoutMismatch)
469        ));
470        cleanup(&base);
471    }
472
473    #[test]
474    fn estimated_count_tracks_real_inserts() {
475        let base = tmp_base("count-est");
476        let (n_bits, n_hashes) = SharedBloomFilter::suggest_config(1000, 0.01);
477        let b = SharedBloomFilter::create(&base, n_bits, n_hashes).unwrap();
478        for i in 0..500u32 {
479            b.insert(format!("k{i}").as_bytes()).unwrap();
480        }
481        let est = b.estimated_insert_count();
482        // Roughly 500, within 25% slack for stochastic distribution.
483        assert!(est > 350 && est < 650,
484            "estimated insert count {est} should be near 500");
485        cleanup(&base);
486    }
487
488    #[test]
489    fn suggest_config_returns_sensible_values() {
490        let (n_bits, n_hashes) = SharedBloomFilter::suggest_config(10_000, 0.01);
491        assert!(n_bits > 90_000 && n_bits < 100_000,
492            "n_bits {n_bits} should be ~95k");
493        assert!((6..=8).contains(&n_hashes),
494            "n_hashes {n_hashes} should be ~7");
495    }
496
497    #[test]
498    fn disk_persistence_survives_reopen() {
499        let base = tmp_base("disk");
500        {
501            let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
502            b.insert(b"persisted").unwrap();
503            b.insert(b"also-persisted").unwrap();
504            b.flush().unwrap();
505        }
506        let b2 = SharedBloomFilter::open(&base, 1024, 3).unwrap();
507        assert!(b2.contains(b"persisted").unwrap());
508        assert!(b2.contains(b"also-persisted").unwrap());
509        assert!(!b2.contains(b"not-there").unwrap());
510        cleanup(&base);
511    }
512
513    #[test]
514    fn concurrent_inserters_no_lost_updates() {
515        use std::sync::Arc;
516        use std::thread;
517        let base = tmp_base("concurrent");
518        let b: Arc<SharedBloomFilter>
519            = Arc::new(SharedBloomFilter::create(&base, 8192, 5).unwrap());
520        let n_threads = 4;
521        let per_thread = 100;
522        let mut handles = vec![];
523        for t in 0..n_threads {
524            let b = b.clone();
525            handles.push(thread::spawn(move || {
526                for i in 0..per_thread {
527                    b.insert(format!("t{t}-i{i}").as_bytes()).unwrap();
528                }
529            }));
530        }
531        for h in handles { h.join().unwrap(); }
532        for t in 0..n_threads {
533            for i in 0..per_thread {
534                let item = format!("t{t}-i{i}");
535                assert!(b.contains(item.as_bytes()).unwrap(),
536                    "missing item {item}");
537            }
538        }
539        cleanup(&base);
540    }
541}
subetha_cxc/shared_bloom_filter.rs

subetha_cxc/
shared_bloom_filter.rs
