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    pub fn create(
145        base_path: impl AsRef<Path>, n_bits: usize, n_hashes: u32,
146    ) -> Result<Self, BloomError> {
147        if n_bits == 0 || n_hashes == 0 {
148            return Err(BloomError::InvalidConfig);
149        }
150        let base = base_path.as_ref();
151        let hpath = header_path(base);
152        let file = OpenOptions::new()
153            .read(true).write(true).create(true).truncate(true)
154            .open(&hpath)?;
155        file.set_len(std::mem::size_of::<BloomHeader>() as u64)?;
156        let mut mmap = unsafe {
157            MmapOptions::new().len(std::mem::size_of::<BloomHeader>()).map_mut(&file)?
158        };
159        let hdr = mmap.as_mut_ptr() as *mut BloomHeader;
160        unsafe {
161            std::ptr::write_bytes(hdr as *mut u8, 0, std::mem::size_of::<BloomHeader>());
162            (*hdr).magic = BLOOM_MAGIC;
163            (*hdr).n_bits = n_bits as u64;
164            (*hdr).n_hashes = n_hashes;
165        }
166        let bits = SharedBitVec::create(bits_path(base), n_bits)?;
167        Ok(Self {
168            _file: file, _mmap: mmap, bits,
169            n_bits: n_bits as u64, n_hashes,
170            header_sidecar: subetha_core::HandshakeHeader::new(),
171            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
172        })
173    }
174
175    pub fn open(
176        base_path: impl AsRef<Path>, n_bits: usize, n_hashes: u32,
177    ) -> Result<Self, BloomError> {
178        let base = base_path.as_ref();
179        let hpath = header_path(base);
180        let file = OpenOptions::new().read(true).write(true).open(&hpath)?;
181        if file.metadata()?.len() < std::mem::size_of::<BloomHeader>() as u64 {
182            return Err(BloomError::LayoutMismatch);
183        }
184        let mmap = unsafe {
185            MmapOptions::new().len(std::mem::size_of::<BloomHeader>()).map_mut(&file)?
186        };
187        let hdr = unsafe { &*(mmap.as_ptr() as *const BloomHeader) };
188        if hdr.magic != BLOOM_MAGIC
189            || hdr.n_bits != n_bits as u64
190            || hdr.n_hashes != n_hashes
191        {
192            return Err(BloomError::LayoutMismatch);
193        }
194        let bits = SharedBitVec::open(bits_path(base), n_bits)?;
195        Ok(Self {
196            _file: file, _mmap: mmap, bits,
197            n_bits: n_bits as u64, n_hashes,
198            header_sidecar: subetha_core::HandshakeHeader::new(),
199            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
200        })
201    }
202
203    #[inline]
204    pub fn n_bits(&self) -> u64 { self.n_bits }
205    #[inline]
206    pub fn n_hashes(&self) -> u32 { self.n_hashes }
207
208    /// Insert an item. Sets `n_hashes` bits in the underlying
209    /// vector. Idempotent: re-inserting the same item is a no-op
210    /// for membership purposes (the bits stay set).
211    /// Map the i-th Kirsch-Mitzenmacher double-hash to a bit index in
212    /// `[0, n_bits)` via Lemire's fastrange (multiply-shift) instead of
213    /// `% n_bits`: replaces a hardware DIV per hash with a multiply +
214    /// shift and needs no power-of-two `n_bits`. insert and contains
215    /// share this mapping, so membership stays consistent.
216    #[inline]
217    fn position(&self, h1: u64, h2: u64, i: u64) -> usize {
218        let h = h1.wrapping_add(i.wrapping_mul(h2));
219        ((h as u128 * self.n_bits as u128) >> 64) as usize
220    }
221
222    pub fn insert(&self, item: &[u8]) -> Result<(), BloomError> {
223        let h1 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_1));
224        let h2 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_2));
225        for i in 0..self.n_hashes as u64 {
226            self.bits.set(self.position(h1, h2, i))?;
227        }
228        self.ring_sidecar
229            .push_op(crate::sidecar_ops::sketch::OP_INSERT, 0);
230        Ok(())
231    }
232
233    /// True if `item` MIGHT be in the set; false if definitely not.
234    /// False positives are possible; false negatives are NOT.
235    pub fn contains(&self, item: &[u8]) -> Result<bool, BloomError> {
236        let h1 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_1));
237        let h2 = fmix64(fnv1a_seeded(item, FNV_OFFSET_BASIS_2));
238        for i in 0..self.n_hashes as u64 {
239            if !self.bits.get(self.position(h1, h2, i))? {
240                self.ring_sidecar
241                    .push_op(crate::sidecar_ops::sketch::OP_QUERY, 2); // absent
242                return Ok(false);
243            }
244        }
245        self.ring_sidecar
246            .push_op(crate::sidecar_ops::sketch::OP_QUERY, 0);
247        Ok(true)
248    }
249
250    /// Clear all bits (resets the filter to empty).
251    pub fn clear(&self) {
252        self.bits.clear_all();
253        self.ring_sidecar
254            .push_op(crate::sidecar_ops::sketch::OP_CLEAR, 0);
255    }
256
257    /// Estimate the current false-positive rate from the bit-set
258    /// density. Returns 0.0 for an empty filter; approaches 1.0
259    /// as the filter saturates.
260    pub fn estimated_false_positive_rate(&self) -> f64 {
261        let fill = self.bits.count_ones() as f64;
262        let n = self.n_bits as f64;
263        let k = self.n_hashes as f64;
264        if fill == 0.0 { return 0.0; }
265        (fill / n).powf(k)
266    }
267
268    /// Estimate the number of distinct inserted items based on bit
269    /// density. Formula: `-n_bits / n_hashes * ln(1 - fill / n_bits)`.
270    pub fn estimated_insert_count(&self) -> u64 {
271        let fill = self.bits.count_ones() as f64;
272        let n = self.n_bits as f64;
273        let k = self.n_hashes as f64;
274        if fill == 0.0 { return 0; }
275        if fill >= n { return u64::MAX; }
276        ((-n / k) * (1.0 - fill / n).ln()).round() as u64
277    }
278
279    pub fn flush(&self) -> Result<(), BloomError> {
280        Ok(self.bits.flush()?)
281    }
282
283    pub fn flush_async(&self) -> Result<(), BloomError> {
284        Ok(self.bits.flush_async()?)
285    }
286}
287
288#[cfg(test)]
289mod tests {
290    use super::*;
291
292    fn tmp_base(name: &str) -> PathBuf {
293        let mut p = std::env::temp_dir();
294        let pid = std::process::id();
295        p.push(format!("subetha-bloom-{name}-{pid}"));
296        p
297    }
298
299    fn cleanup(base: &Path) {
300        std::fs::remove_file(header_path(base)).ok();
301        std::fs::remove_file(bits_path(base)).ok();
302    }
303
304    #[test]
305    fn empty_filter_contains_nothing() {
306        let base = tmp_base("empty");
307        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
308        assert!(!b.contains(b"anything").unwrap());
309        assert!(!b.contains(b"").unwrap());
310        cleanup(&base);
311    }
312
313    #[test]
314    fn insert_then_contains_returns_true_no_false_negatives() {
315        let base = tmp_base("no-fn");
316        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
317        let items: &[&[u8]] = &[
318            b"hello", b"world", b"adaptive-prims", b"42",
319            b"the quick brown fox", b"", b"a",
320        ];
321        for &item in items {
322            b.insert(item).unwrap();
323        }
324        for &item in items {
325            assert!(b.contains(item).unwrap(),
326                "false negative for item: {item:?}");
327        }
328        cleanup(&base);
329    }
330
331    #[test]
332    fn invalid_config_rejected() {
333        let base = tmp_base("invalid");
334        assert_eq!(
335            SharedBloomFilter::create(&base, 0, 3).err(),
336            Some(BloomError::InvalidConfig)
337        );
338        assert_eq!(
339            SharedBloomFilter::create(&base, 1024, 0).err(),
340            Some(BloomError::InvalidConfig)
341        );
342        cleanup(&base);
343    }
344
345    #[test]
346    fn false_positive_rate_within_bounds_at_target_load() {
347        let (n_bits, n_hashes) = SharedBloomFilter::suggest_config(1000, 0.01);
348        let base = tmp_base("fpr");
349        let b = SharedBloomFilter::create(&base, n_bits, n_hashes).unwrap();
350        for i in 0..1000u32 {
351            b.insert(format!("item-{i:04}").as_bytes()).unwrap();
352        }
353        let mut fp = 0u32;
354        for i in 10_000u32..20_000 {
355            if b.contains(format!("query-{i:05}").as_bytes()).unwrap() {
356                fp += 1;
357            }
358        }
359        let observed_fpr = fp as f64 / 10_000.0;
360        // Target was 0.01; allow stochastic slack up to 0.03.
361        assert!(observed_fpr < 0.03,
362            "observed FPR {observed_fpr} should be below 0.03 (target was 0.01)");
363        cleanup(&base);
364    }
365
366    #[test]
367    fn clear_resets_filter() {
368        let base = tmp_base("clear");
369        let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
370        b.insert(b"foo").unwrap();
371        b.insert(b"bar").unwrap();
372        assert!(b.contains(b"foo").unwrap());
373        b.clear();
374        assert!(!b.contains(b"foo").unwrap());
375        assert!(!b.contains(b"bar").unwrap());
376        cleanup(&base);
377    }
378
379    #[test]
380    fn cross_handle_visibility() {
381        let base = tmp_base("cross-handle");
382        let writer = SharedBloomFilter::create(&base, 1024, 3).unwrap();
383        let reader = SharedBloomFilter::open(&base, 1024, 3).unwrap();
384        writer.insert(b"cross-process").unwrap();
385        assert!(reader.contains(b"cross-process").unwrap());
386        assert!(!reader.contains(b"not-inserted").unwrap());
387        cleanup(&base);
388    }
389
390    #[test]
391    fn config_mismatch_at_open_rejected() {
392        let base = tmp_base("mismatch");
393        let _w = SharedBloomFilter::create(&base, 1024, 3).unwrap();
394        assert!(matches!(
395            SharedBloomFilter::open(&base, 2048, 3),
396            Err(BloomError::LayoutMismatch)
397        ));
398        assert!(matches!(
399            SharedBloomFilter::open(&base, 1024, 5),
400            Err(BloomError::LayoutMismatch)
401        ));
402        cleanup(&base);
403    }
404
405    #[test]
406    fn estimated_count_tracks_real_inserts() {
407        let base = tmp_base("count-est");
408        let (n_bits, n_hashes) = SharedBloomFilter::suggest_config(1000, 0.01);
409        let b = SharedBloomFilter::create(&base, n_bits, n_hashes).unwrap();
410        for i in 0..500u32 {
411            b.insert(format!("k{i}").as_bytes()).unwrap();
412        }
413        let est = b.estimated_insert_count();
414        // Roughly 500, within 25% slack for stochastic distribution.
415        assert!(est > 350 && est < 650,
416            "estimated insert count {est} should be near 500");
417        cleanup(&base);
418    }
419
420    #[test]
421    fn suggest_config_returns_sensible_values() {
422        let (n_bits, n_hashes) = SharedBloomFilter::suggest_config(10_000, 0.01);
423        assert!(n_bits > 90_000 && n_bits < 100_000,
424            "n_bits {n_bits} should be ~95k");
425        assert!((6..=8).contains(&n_hashes),
426            "n_hashes {n_hashes} should be ~7");
427    }
428
429    #[test]
430    fn disk_persistence_survives_reopen() {
431        let base = tmp_base("disk");
432        {
433            let b = SharedBloomFilter::create(&base, 1024, 3).unwrap();
434            b.insert(b"persisted").unwrap();
435            b.insert(b"also-persisted").unwrap();
436            b.flush().unwrap();
437        }
438        let b2 = SharedBloomFilter::open(&base, 1024, 3).unwrap();
439        assert!(b2.contains(b"persisted").unwrap());
440        assert!(b2.contains(b"also-persisted").unwrap());
441        assert!(!b2.contains(b"not-there").unwrap());
442        cleanup(&base);
443    }
444
445    #[test]
446    fn concurrent_inserters_no_lost_updates() {
447        use std::sync::Arc;
448        use std::thread;
449        let base = tmp_base("concurrent");
450        let b: Arc<SharedBloomFilter>
451            = Arc::new(SharedBloomFilter::create(&base, 8192, 5).unwrap());
452        let n_threads = 4;
453        let per_thread = 100;
454        let mut handles = vec![];
455        for t in 0..n_threads {
456            let b = b.clone();
457            handles.push(thread::spawn(move || {
458                for i in 0..per_thread {
459                    b.insert(format!("t{t}-i{i}").as_bytes()).unwrap();
460                }
461            }));
462        }
463        for h in handles { h.join().unwrap(); }
464        for t in 0..n_threads {
465            for i in 0..per_thread {
466                let item = format!("t{t}-i{i}");
467                assert!(b.contains(item.as_bytes()).unwrap(),
468                    "missing item {item}");
469            }
470        }
471        cleanup(&base);
472    }
473}
subetha_cxc/shared_bloom_filter.rs

subetha_cxc/
shared_bloom_filter.rs
