cuckoo_clock/
filter.rs

1use std::{
2    hash::{BuildHasher, Hash, RandomState},
3    io::Read,
4    iter::repeat_with,
5    num::NonZeroUsize,
6    sync::{
7        Arc, Mutex, MutexGuard,
8        atomic::{AtomicUsize, Ordering},
9    },
10};
11
12use crate::{
13    associated_data::AssociatedData,
14    bucket::{Bucket, InsertValues, LookupValues},
15    config::CuckooConfiguration,
16    data_block::{DataBlock, Fingerprint},
17    exporter::{
18        CuckooFilterExporter, Exportable, ExportableBuildHasher, ExportableRandomState,
19        read_hasher_from,
20    },
21};
22
23/// Thread-safe cuckoo filter, with support for TTL, LRU and custom counters associated with the
24/// stored data.
25///
26/// Instances of [`CuckooFilter`] can be cloned and used across different threads. To ensure thread
27/// safety, locks are used, but locking is done per bucket, meaning that 2 separate threads can
28/// freely access different buckets without conflicts. In most cases locks shouldn't block, because
29/// optimal cuckoo filter configuration will have a large number of buckets, reducing the change of
30/// concurrent access to the same bucket.
31///
32/// Instances of [`CuckooFilter`] build using [`CuckooFilter::new_random_exportable`] or with a
33/// [`BuildHasher`] that also implements [`ExportableBuildHasher`] can be exported and imported,
34/// using [`CuckooFilter::exporter`] and [`CuckooFilter::import`]. Note that configuration is
35/// stored in the exported data too and can't be changed, because any changes to the configuration
36/// data would invalidate all of the stored data.
37///
38/// # Examples
39///
40/// Basic cuckoo filter with default configuration
41/// ```
42/// use cuckoo_clock::{CuckooFilter, config::CuckooConfiguration};
43///
44/// let filter = CuckooFilter::new_random(CuckooConfiguration::builder(100_000).build()?);
45///
46/// // None returned from insertion means no entry was evicted
47/// assert!(filter.insert("example_data").is_none());
48///
49/// // Insertion must have been successful
50/// assert!(filter.contains("example_data"));
51///
52/// // Deletion must have been successful
53/// assert!(filter.remove("example_data"));
54/// assert!(!filter.contains("example_data"));
55///
56/// # Ok::<(), Box<dyn std::error::Error>>(())
57/// ```
58///
59/// More complex use-case, with additional options
60/// ```
61/// use cuckoo_clock::{CuckooFilter, config::{CuckooConfiguration, CounterConfig, TtlConfig}};
62///
63/// let filter = CuckooFilter::new_random(
64///     CuckooConfiguration::builder(10_000_000)
65///         .fingerprint_bits(18.try_into()?)
66///         .bucket_size(8.try_into()?)
67///         .with_counter(CounterConfig {
68///             counter_bits: 4.try_into()?,
69///             ..Default::default()
70///         })
71///         .with_ttl(TtlConfig {
72///             ttl: 600.try_into()?,
73///             ttl_bits: 10.try_into()?
74///         })
75///         .build()?
76/// );
77///
78/// // In this case, we use `insert_if_not_present` to ensure no duplicates, because we care about
79/// // the counter
80/// // None returned from insertion means no entry was evicted
81/// assert!(filter.insert_if_not_present("example_data").is_none());
82/// assert!(filter.insert_if_not_present("example_data").is_none());
83///
84/// // Insertion must have been successful
85/// assert!(filter.contains("example_data"));
86///
87/// // Counter should be 4 now.
88/// // We have accessed this item 3 times, but `get_associated_data` also counts as an access.
89/// assert_eq!(filter.get_associated_data("example_data").unwrap().get_counter()?, 4);
90///
91/// # Ok::<(), Box<dyn std::error::Error>>(())
92/// ```
93///
94/// Export/import
95/// ```
96/// use std::{collections::VecDeque, io::Read};
97/// use cuckoo_clock::{CuckooFilter, config::CuckooConfiguration};
98///
99/// let filter = CuckooFilter::new_random_exportable(CuckooConfiguration::builder(100_000).build()?);
100///
101/// // None returned from insertion means no entry was evicted
102/// assert!(filter.insert("example_data").is_none());
103///
104/// let mut buf = Vec::new();
105/// filter.exporter().write_to(&mut buf)?;
106///
107/// let mut buf = VecDeque::from(buf);
108/// let imported_filter = CuckooFilter::import_random_exportable(&mut buf)?;
109///
110/// // The inserted data is available in the imported filter
111/// assert!(filter.contains("example_data"));
112///
113/// # Ok::<(), Box<dyn std::error::Error>>(())
114/// ```
115///
116#[derive(Clone)]
117pub struct CuckooFilter<H: BuildHasher> {
118    configuration: CuckooConfiguration,
119    buckets: Arc<Vec<Mutex<Bucket>>>,
120    build_hasher: H,
121    items: Arc<AtomicUsize>,
122}
123
124impl CuckooFilter<RandomState> {
125    /// Creates a new instance of [`CuckooFilter`], using [`RandomState`] as its [`BuildHasher`].
126    ///
127    /// # Panics
128    ///
129    /// Panics if allocation of buckets fails (if too much memory was requested).
130    #[must_use]
131    pub fn new_random(configuration: CuckooConfiguration) -> Self {
132        Self::new(configuration, RandomState::new())
133    }
134}
135
136impl CuckooFilter<ExportableRandomState> {
137    /// Creates a new instance of [`CuckooFilter`], using [`ExportableBuildHasher`] based on
138    /// [`RandomState`] as its [`BuildHasher`].
139    /// This instance supports export using [`CuckooFilter::export`].
140    ///
141    /// # Panics
142    ///
143    /// Panics if allocation of buckets fails (if too much memory was requested).
144    #[must_use]
145    pub fn new_random_exportable(configuration: CuckooConfiguration) -> Self {
146        Self::new(configuration, ExportableRandomState::new_random())
147    }
148
149    /// Creates a new instance of [`CuckooFilter`], using [`ExportableBuildHasher`] based on
150    /// exported data.
151    ///
152    /// # Panics
153    ///
154    /// Panics if allocation of buckets fails (if too much memory was requested).
155    pub fn import_random_exportable(reader: impl Read) -> Result<Self, crate::ImportError> {
156        Self::import(reader)
157    }
158}
159
160impl<H: ExportableBuildHasher + BuildHasher> CuckooFilter<H> {
161    /// Creates a cuckoo filter from its exported state.
162    ///
163    /// # Panics
164    ///
165    /// Panics if allocation of buckets fails (if too much memory was requested).
166    pub fn import(mut reader: impl Read) -> Result<Self, crate::ImportError> {
167        let (hasher, configuration) = Self::import_config(&mut reader)?;
168        Self::import_state(hasher, configuration, reader)
169    }
170
171    /// Creates a cuckoo filter configuration from its exported state - skips the actual state.
172    pub fn import_config(
173        mut reader: impl Read,
174    ) -> Result<(H, CuckooConfiguration), crate::ImportError> {
175        let hasher = read_hasher_from::<H>(&mut reader)?;
176        let config = CuckooConfiguration::read_from(&mut reader)?;
177        Ok((hasher, config))
178    }
179
180    /// Creates a cuckoo filter from its exported state and already read hasher and configuration.
181    /// This assumes that these 2 were already read from the provided reader.
182    ///
183    /// # Panics
184    ///
185    /// Panics if allocation of buckets fails (if too much memory was requested).
186    pub fn import_state(
187        hasher: H,
188        configuration: CuckooConfiguration,
189        mut reader: impl Read,
190    ) -> Result<Self, crate::ImportError> {
191        let mut buckets = Vec::with_capacity(configuration.bucket_count);
192
193        let mut item_count = 0;
194        for _ in 0..configuration.bucket_count {
195            let bucket = Bucket::take_from(&mut reader, &configuration)?;
196            item_count += bucket.occupied_count(&configuration);
197            buckets.push(Mutex::new(bucket));
198        }
199
200        Ok(Self {
201            configuration,
202            buckets: Arc::new(buckets),
203            build_hasher: hasher,
204            items: Arc::new(AtomicUsize::new(item_count)),
205        })
206    }
207
208    /// Prepares an exporter for this [`CuckooFilter`], enabling to persist it and import it later
209    /// using [`CuckooFilter::import`].
210    pub fn exporter<'a>(&'a self) -> CuckooFilterExporter<'a, H> {
211        CuckooFilterExporter::new(&self.build_hasher, &self.buckets, &self.configuration)
212    }
213}
214
215impl<H: BuildHasher> CuckooFilter<H> {
216    /// Creates a new instance of [`CuckooFilter`], using provided [`BuildHasher`].
217    ///
218    /// # Panics
219    ///
220    /// Panics if allocation of buckets fails (if too much memory was requested).
221    pub fn new(configuration: CuckooConfiguration, build_hasher: H) -> Self {
222        Self {
223            configuration: configuration.clone(),
224            buckets: repeat_with(|| Bucket::new(&configuration).into())
225                .take(configuration.bucket_count)
226                .collect::<Vec<_>>()
227                .into(),
228            build_hasher,
229            items: Arc::new(AtomicUsize::new(0)),
230        }
231    }
232
233    /// Returns the actual bucket count for this [`CuckooFilter`].
234    ///
235    /// Bucket count is calculated as first next power of two of capacity / bucket_size.
236    /// This means that the actual capacity of the filter is usually bigger than the requested
237    /// capacity.
238    pub const fn get_bucket_count(&self) -> usize {
239        self.configuration.bucket_count
240    }
241
242    /// Returns the actual number of items currently stored in this [`CuckooFilter`].
243    pub fn get_item_count(&self) -> usize {
244        self.items.load(Ordering::Relaxed)
245    }
246
247    /// Returns the configuration for this [`CuckooFilter`].
248    pub fn get_configuration(&self) -> CuckooConfiguration {
249        self.configuration.clone()
250    }
251
252    /// Returns the memory usage of this filter in bytes.
253    pub fn get_memory_usage(&self) -> usize {
254        size_of::<Self>()
255            + size_of::<AtomicUsize>()
256            + size_of::<Vec<Mutex<Bucket>>>()
257            + size_of::<Mutex<Bucket>>() * self.buckets.len()
258            + self.configuration.bucket_byte_size * self.buckets.len()
259    }
260
261    /// Returns the expected memory usage of a filter created with provided parameters.
262    pub(crate) const fn get_expected_memory_usage(
263        bucket_byte_size: usize,
264        buckets: usize,
265    ) -> usize {
266        size_of::<Self>()
267            + size_of::<AtomicUsize>()
268            + size_of::<Vec<Mutex<Bucket>>>()
269            + size_of::<Mutex<Bucket>>() * buckets
270            + bucket_byte_size * buckets
271    }
272
273    /// Inserts a new item into the filter, only if the filter doesn't contain it already.
274    ///
275    /// This is slower than [`CuckooFilter::insert`], but it ensures that no duplicates are present
276    /// in the filter. That can be useful when [`AssociatedData`] is used, to ensure consistent
277    /// results.
278    ///
279    /// Returns fingerprint of the item that was evicted from the filter, if eviction had to take
280    /// place to finalize the insertion. It is possible that the item that was just inserted gets
281    /// evicted in random kicking process. That can be confirmed using
282    /// [`Fingerprint::matches_key`].
283    pub fn insert_if_not_present<K: Hash + ?Sized>(&self, key: &K) -> Option<Fingerprint> {
284        self.insert_if_not_present_with_update(
285            key,
286            InsertValues::default(),
287            LookupValues::default(),
288        )
289    }
290
291    /// Inserts a new item into the filter, only if the filter doesn't contain it already. Also
292    /// applies provided updates.
293    ///
294    /// This is similar to [`CuckooFilter::insert_if_not_present`], but it also updates found
295    /// values, or starts off values with different values.
296    pub fn insert_if_not_present_with_update<K: Hash + ?Sized>(
297        &self,
298        key: &K,
299        insert_values: InsertValues,
300        lookup_update: LookupValues,
301    ) -> Option<Fingerprint> {
302        let (fp, i1) = self.get_fingerprint_and_index(key);
303
304        let mut contains =
305            self.lock_bucket(i1 as usize)
306                .contains(&fp, &self.configuration, &lookup_update);
307
308        if contains {
309            return None;
310        }
311
312        let i2 = self.alt_index(&fp, i1);
313        contains = self
314            .lock_bucket(i2 as usize)
315            .contains(&fp, &self.configuration, &lookup_update);
316
317        if contains {
318            return None;
319        }
320
321        let mut cur_data_block = self.new_data_block(&fp, insert_values);
322
323        let inserted = self
324            .lock_bucket(i1 as usize)
325            .insert(&cur_data_block, &self.configuration);
326
327        if inserted {
328            self.items.fetch_add(1, Ordering::Relaxed);
329            return None;
330        }
331
332        let inserted = self
333            .lock_bucket(i2 as usize)
334            .insert(&cur_data_block, &self.configuration);
335
336        if inserted {
337            self.items.fetch_add(1, Ordering::Relaxed);
338            return None;
339        }
340
341        let mut cur_index = if rand::random::<bool>() { i1 } else { i2 };
342        let mut new_item = true;
343        for _ in 0..self.configuration.max_kicks {
344            {
345                let mut bucket = self.lock_bucket(cur_index as usize);
346                // Replace a random item first
347                if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
348                    if !bucket.kick_lru(
349                        &mut cur_data_block,
350                        &self.configuration,
351                        lru_config,
352                        new_item,
353                    ) {
354                        return Some(cur_data_block.get_fingerprint(&self.configuration));
355                    }
356                    new_item = false;
357                } else {
358                    bucket.kick_random(&mut cur_data_block, &self.configuration);
359                }
360                cur_index = self.alt_index(
361                    &cur_data_block.get_fingerprint(&self.configuration),
362                    cur_index,
363                );
364            }
365
366            if self
367                .lock_bucket(cur_index as usize)
368                .insert(&cur_data_block, &self.configuration)
369            {
370                self.items.fetch_add(1, Ordering::Relaxed);
371                // Found an alternative spot for evicted item, done with kicks
372                return None;
373            }
374        }
375
376        // Filter is full
377        Some(cur_data_block.get_fingerprint(&self.configuration))
378    }
379
380    /// Inserts a new item into the filter.
381    ///
382    /// If both target buckets for this item are full, random item is kicked out of one of these 2
383    /// buckets and moved into its alternate bucket, starting a recursive kicking process, which
384    /// stops once an empty slot is found in alternate bucket of a kicked item, or
385    /// [`crate::config::CuckooConfigurationBuilder::max_kicks`] is reached.
386    ///
387    /// Returns fingerprint of the item that was evicted from the filter, if eviction had to take
388    /// place to finalize the insertion. It is possible that the item that was just inserted gets
389    /// evicted in random kicking process. That can be confirmed using
390    /// [`Fingerprint::matches_key`].
391    pub fn insert<K: Hash + ?Sized>(&self, key: &K) -> Option<Fingerprint> {
392        self.insert_with_defaults(key, InsertValues::default())
393    }
394
395    /// Inserts a new item into the filter, with defined defaults for associated data.
396    ///
397    /// This is similar to [`CuckooFilter::insert`], but allows overrides for associated data
398    /// defaults.
399    pub fn insert_with_defaults<K: Hash + ?Sized>(
400        &self,
401        key: &K,
402        default: InsertValues,
403    ) -> Option<Fingerprint> {
404        let (fp, i1) = self.get_fingerprint_and_index(key);
405        let mut cur_data_block = self.new_data_block(&fp, default);
406
407        let inserted = self
408            .lock_bucket(i1 as usize)
409            .insert(&cur_data_block, &self.configuration);
410
411        if inserted {
412            self.items.fetch_add(1, Ordering::Relaxed);
413            return None;
414        }
415
416        let i2 = self.alt_index(&fp, i1);
417
418        let inserted = self
419            .lock_bucket(i2 as usize)
420            .insert(&cur_data_block, &self.configuration);
421
422        if inserted {
423            self.items.fetch_add(1, Ordering::Relaxed);
424            return None;
425        }
426
427        let mut cur_index = i1;
428        let mut new_item = true;
429        for _ in 0..self.configuration.max_kicks {
430            {
431                let mut bucket = self.lock_bucket(cur_index as usize);
432                // Replace a random item first
433                if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
434                    if !bucket.kick_lru(
435                        &mut cur_data_block,
436                        &self.configuration,
437                        lru_config,
438                        new_item,
439                    ) {
440                        return Some(cur_data_block.get_fingerprint(&self.configuration));
441                    }
442                    new_item = false;
443                } else {
444                    // TODO: this can even kick the newest item, which is not ideal
445                    bucket.kick_random(&mut cur_data_block, &self.configuration);
446                }
447                cur_index = self.alt_index(
448                    &cur_data_block.get_fingerprint(&self.configuration),
449                    cur_index,
450                );
451            }
452
453            if self
454                .lock_bucket(cur_index as usize)
455                .insert(&cur_data_block, &self.configuration)
456            {
457                self.items.fetch_add(1, Ordering::Relaxed);
458                // Found an alternative spot for evicted item, done with kicks
459                return None;
460            }
461        }
462
463        // Filter is full
464        Some(cur_data_block.get_fingerprint(&self.configuration))
465    }
466
467    /// Check if this key is stored in the filter and applies the provided [`LookupValues`].
468    ///
469    /// This is similar to [`CuckooFilter::contains`], but allows overrides for updates on
470    /// successful lookup.
471    pub fn contains_with_update<K: Hash + ?Sized>(&self, key: &K, update: LookupValues) -> bool {
472        let (fp, i1) = self.get_fingerprint_and_index(key);
473
474        let mut contains =
475            self.lock_bucket(i1 as usize)
476                .contains(&fp, &self.configuration, &update);
477
478        if !contains {
479            let i2 = self.alt_index(&fp, i1);
480            contains = self
481                .lock_bucket(i2 as usize)
482                .contains(&fp, &self.configuration, &update);
483        }
484
485        contains
486    }
487
488    /// Check if this key is stored in the filter.
489    ///
490    /// Returns true if this key might be present in the filter. If false is returned, then the key
491    /// is definitely not present.
492    pub fn contains<K: Hash + ?Sized>(&self, key: &K) -> bool {
493        self.contains_with_update(key, LookupValues::default())
494    }
495
496    /// Loads associated data of a key stored in the filter.
497    ///
498    /// Returns None if this filter is not present in the filter. Returns associated data for the
499    /// first item with the fingerprint matching this key's fingerprint. Note that it is
500    /// recommended to use [`CuckooFilter::insert_if_not_present`] if consistent [`AssociatedData`]
501    /// is required.
502    pub fn get_associated_data<K: Hash + ?Sized>(&self, key: &K) -> Option<AssociatedData> {
503        self.get_associated_data_with_update(key, LookupValues::default())
504    }
505
506    /// Loads associated data of a key stored in the filter and applies the provided [`LookupValues`].
507    ///
508    /// This is similar to [`CuckooFilter::get_associated_data`], but allows overrides for updates on
509    /// successful lookup.
510    pub fn get_associated_data_with_update<K: Hash + ?Sized>(
511        &self,
512        key: &K,
513        update: LookupValues,
514    ) -> Option<AssociatedData> {
515        let (fp, i1) = self.get_fingerprint_and_index(key);
516
517        let mut contains =
518            self.lock_bucket(i1 as usize)
519                .get_associated_data(&fp, &self.configuration, &update);
520
521        if contains.is_none() {
522            let i2 = self.alt_index(&fp, i1);
523            contains = self.lock_bucket(i2 as usize).get_associated_data(
524                &fp,
525                &self.configuration,
526                &update,
527            );
528        }
529
530        contains
531    }
532
533    /// Removes this key from the filter, if present.
534    ///
535    /// Returns true if the key was present in the filter.
536    pub fn remove<K: Hash + ?Sized>(&self, key: &K) -> bool {
537        let (fp, i1) = self.get_fingerprint_and_index(key);
538
539        let mut removed = self
540            .lock_bucket(i1 as usize)
541            .remove(&fp, &self.configuration);
542
543        if !removed {
544            let i2 = self.alt_index(&fp, i1);
545            removed = self
546                .lock_bucket(i2 as usize)
547                .remove(&fp, &self.configuration);
548        }
549
550        if removed {
551            self.items.fetch_sub(1, Ordering::Relaxed);
552        }
553
554        removed
555    }
556
557    /// Scans all buckets of this filter and reduces TTL and LRU counters.
558    ///
559    /// If LRU and/or TTL features are used, this must be called periodically.
560    /// Each call to this function will age all the LRU and TTL counters. The frequency of calls
561    /// will affect both LRU and TTL in different ways:
562    /// - TTL will get reduced by 1 on each call, meaning that scanning each second indirectly sets
563    ///   the unit of TTL field to be seconds.
564    /// - LRU will get halved on each call. By scanning more frequently, items will require more
565    ///   frequent usage to stay in the filter.
566    ///
567    /// This is a no-op if both LRU and TTL are disabled.
568    ///
569    /// # Examples
570    ///
571    /// ```
572    /// use cuckoo_clock::{CuckooFilter, config::{CuckooConfiguration, CounterConfig, TtlConfig}};
573    ///
574    /// let filter = CuckooFilter::new_random(
575    ///     CuckooConfiguration::builder(10_000)
576    ///         .with_ttl(TtlConfig {
577    ///             ttl: 3.try_into()?,
578    ///             ttl_bits: 2.try_into()?
579    ///         })
580    ///         .build()?
581    /// );
582    ///
583    /// filter.insert("example_data");
584    ///
585    /// assert!(filter.contains("example_data"));
586    ///
587    /// filter.scan_and_update_full();
588    /// assert!(filter.contains("example_data"));
589    ///
590    /// filter.scan_and_update_full();
591    /// assert!(filter.contains("example_data"));
592    ///
593    /// // The item will get removed now, due to expired TTL
594    /// filter.scan_and_update_full();
595    /// assert!(!filter.contains("example_data"));
596    ///
597    /// # Ok::<(), Box<dyn std::error::Error>>(())
598    /// ```
599    pub fn scan_and_update_full(&self) -> usize {
600        #[expect(clippy::unwrap_used)]
601        self.scan_and_update_full_partition(NonZeroUsize::new(1).unwrap(), 0)
602    }
603
604    /// Scans a single group of buckets of this filter and reduces TTL and LRU counters.
605    ///
606    /// This is the same as [`CuckooFilter::scan_and_update_full`], but it more suitable for
607    /// parallelization, by splitting buckets into partitions to process in parallel.
608    pub fn scan_and_update_full_partition(
609        &self,
610        total_partitions: NonZeroUsize,
611        partition_index: usize,
612    ) -> usize {
613        if self.configuration.lru_field_config.is_none()
614            && self.configuration.ttl_field_config.is_none()
615        {
616            return 0;
617        }
618
619        let mut removed = 0;
620        let part_size = self.buckets.len().div_ceil(total_partitions.get());
621        if (partition_index * part_size) >= self.buckets.len() {
622            return 0;
623        }
624        for b in self.buckets
625            [partition_index * part_size..self.buckets.len().min((partition_index + 1) * part_size)]
626            .iter()
627        {
628            #[expect(clippy::unwrap_used)]
629            let mut bucket = b.lock().unwrap();
630            if let Some(lru_config) = &self.configuration.lru_field_config {
631                removed += bucket.age_lru_counters(&self.configuration, lru_config)
632            }
633            if let Some(ttl_config) = &self.configuration.ttl_field_config {
634                removed += bucket.age_ttl_counters(&self.configuration, ttl_config);
635            }
636        }
637
638        if removed > 0 {
639            self.items.fetch_sub(removed, Ordering::Relaxed);
640        }
641        removed
642    }
643
644    /// Scans all buckets of this filter and reduces TTL counters.
645    ///
646    /// Similar to [`CuckooFilter::scan_and_update_full`], but updates only TTL counters. This
647    /// allows more control, enabling different update frequency for TTL and LRU.
648    ///
649    /// This is a no-op if TTL is disabled.
650    pub fn scan_and_update_ttl(&self) -> usize {
651        #[expect(clippy::unwrap_used)]
652        self.scan_and_update_ttl_partition(NonZeroUsize::new(1).unwrap(), 0)
653    }
654
655    /// Scans a single group of buckets of this filter and reduces TTL counters.
656    ///
657    /// This is the same as [`CuckooFilter::scan_and_update_ttl`], but it more suitable for
658    /// parallelization, by splitting buckets into partitions to process in parallel.
659    pub fn scan_and_update_ttl_partition(
660        &self,
661        total_partitions: NonZeroUsize,
662        partition_index: usize,
663    ) -> usize {
664        if self.configuration.ttl_field_config.is_none() {
665            return 0;
666        }
667
668        let mut removed = 0;
669        let part_size = self.buckets.len().div_ceil(total_partitions.get());
670        if (partition_index * part_size) >= self.buckets.len() {
671            return 0;
672        }
673        for b in self.buckets
674            [partition_index * part_size..self.buckets.len().min((partition_index + 1) * part_size)]
675            .iter()
676        {
677            #[expect(clippy::unwrap_used)]
678            let mut bucket = b.lock().unwrap();
679            if let Some(ttl_config) = &self.configuration.ttl_field_config {
680                removed += bucket.age_ttl_counters(&self.configuration, ttl_config);
681            }
682        }
683
684        if removed > 0 {
685            self.items.fetch_sub(removed, Ordering::Relaxed);
686        }
687        removed
688    }
689
690    /// Scans all buckets of this filter and reduces LRU counters.
691    ///
692    /// Similar to [`CuckooFilter::scan_and_update_full`], but updates only LRU counters. This
693    /// allows more control, enabling different update frequency for TTL and LRU.
694    ///
695    /// This is a no-op if LRU is disabled.
696    pub fn scan_and_update_lru(&self) -> usize {
697        #[expect(clippy::unwrap_used)]
698        self.scan_and_update_lru_partition(NonZeroUsize::new(1).unwrap(), 0)
699    }
700
701    /// Scans a single group of buckets of this filter and reduces LRU counters.
702    ///
703    /// This is the same as [`CuckooFilter::scan_and_update_lru`], but it more suitable for
704    /// parallelization, by splitting buckets into partitions to process in parallel.
705    pub fn scan_and_update_lru_partition(
706        &self,
707        total_partitions: NonZeroUsize,
708        partition_index: usize,
709    ) -> usize {
710        if self.configuration.lru_field_config.is_none() {
711            return 0;
712        }
713
714        let mut removed = 0;
715        let part_size = self.buckets.len().div_ceil(total_partitions.get());
716        if (partition_index * part_size) >= self.buckets.len() {
717            return removed;
718        }
719        for b in self.buckets
720            [partition_index * part_size..self.buckets.len().min((partition_index + 1) * part_size)]
721            .iter()
722        {
723            #[expect(clippy::unwrap_used)]
724            let mut bucket = b.lock().unwrap();
725            if let Some(lru_config) = &self.configuration.lru_field_config {
726                removed += bucket.age_lru_counters(&self.configuration, lru_config);
727            }
728        }
729
730        removed
731    }
732
733    /// Generates the fingerprint and first index for the provided key.
734    pub(crate) fn get_fingerprint<K: Hash + ?Sized>(&self, key: &K) -> Fingerprint {
735        self.get_fingerprint_and_index(key).0
736    }
737
738    fn new_data_block(&self, fp: &Fingerprint, defaults: InsertValues) -> DataBlock<Vec<u8>> {
739        let data = vec![0u8; self.configuration.data_block_size];
740        let mut cur_data_block = DataBlock::from(data);
741        cur_data_block.store_fingerprint(fp, &self.configuration);
742
743        if let Some(ttl_config) = self.configuration.ttl_field_config.as_ref() {
744            cur_data_block.set_ttl(ttl_config, defaults.ttl.unwrap_or(ttl_config.0.ttl.into()));
745        }
746        if let Some(counter_config) = self.configuration.counter_field_config.as_ref() {
747            cur_data_block.update_counter(
748                counter_config,
749                defaults
750                    .counter
751                    .unwrap_or(counter_config.0.change_on_insert),
752            );
753        }
754        if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
755            cur_data_block.init_lru_counter(lru_config);
756        }
757        cur_data_block
758    }
759
760    fn get_fingerprint_and_index<K: Hash + ?Sized>(&self, key: &K) -> (Fingerprint, u32) {
761        let result = self.build_hasher.hash_one(key);
762
763        // Fingeprint bits over 32 are definitely an overkill
764        // We can reduce number of hashes by using one hash as fingerprint and first index
765        let fingerprint = (result >> 32) as u32;
766        // Intentional truncation here
767        #[expect(clippy::cast_possible_truncation)]
768        let index = result as u32 & self.configuration.buckets_mask;
769
770        (
771            Fingerprint::new(
772                fingerprint,
773                self.configuration.fingerprint_field_config.value_mask(),
774            ),
775            index,
776        )
777    }
778
779    // Intentional truncation here
780    #[expect(clippy::cast_possible_truncation)]
781    fn alt_index(&self, fingerprint: &Fingerprint, index: u32) -> u32 {
782        let result = self.build_hasher.hash_one(fingerprint);
783
784        (index ^ ((result as u32) & self.configuration.buckets_mask))
785            & self.configuration.buckets_mask
786    }
787
788    #[expect(clippy::unwrap_used)]
789    fn lock_bucket(&self, index: usize) -> MutexGuard<'_, Bucket> {
790        // Any panic while lock is held should come from this library
791        // Any panic produced while the lock is held is a bug in the library!
792        self.buckets[index].lock().unwrap()
793    }
794}
795
796#[cfg(test)]
797#[expect(clippy::unwrap_used)]
798mod tests {
799    use std::{
800        collections::{HashSet, VecDeque},
801        hash::Hasher,
802        ops::Range,
803    };
804
805    use crate::config::{CounterConfig, LruConfig, TtlConfig};
806
807    use super::*;
808
809    fn get_words(range: Range<usize>) -> Vec<String> {
810        std::fs::read_to_string("/usr/share/dict/words")
811            .unwrap()
812            .split("\n")
813            .skip(range.start)
814            .take(range.len())
815            .map(ToString::to_string)
816            .collect()
817    }
818
819    #[test]
820    fn basic_insertion() {
821        let filter = CuckooFilter::new_random(CuckooConfiguration::builder(1000).build().unwrap());
822
823        filter.insert("basic");
824
825        assert!(filter.contains("basic"));
826    }
827
828    #[test]
829    fn basic_removal() {
830        let filter = CuckooFilter::new_random(CuckooConfiguration::builder(1000).build().unwrap());
831
832        filter.insert("basic");
833
834        assert!(filter.contains("basic"));
835
836        filter.remove("basic");
837
838        assert!(!filter.contains("basic"));
839    }
840
841    struct PredefinedBucketItem(u64);
842    struct TestHasher(u64);
843    impl BuildHasher for TestHasher {
844        type Hasher = TestHasher;
845
846        fn build_hasher(&self) -> Self::Hasher {
847            TestHasher(0)
848        }
849    }
850    impl Hasher for TestHasher {
851        fn finish(&self) -> u64 {
852            self.0
853        }
854
855        fn write(&mut self, bytes: &[u8]) {
856            if bytes.len() == 8 {
857                self.0 = u64::from_ne_bytes(bytes.try_into().unwrap());
858            } else {
859                // Shift fingeprint hashes a bit, to allow control
860                self.0 = 1 - (u32::from_ne_bytes(bytes.try_into().unwrap()) as u64 % 2);
861            }
862        }
863    }
864    impl Hash for PredefinedBucketItem {
865        fn hash<H: Hasher>(&self, state: &mut H) {
866            state.write_u64(self.0);
867        }
868    }
869
870    #[test]
871    fn lru_insertion() {
872        let filter = CuckooFilter::new(
873            CuckooConfiguration::builder(1000)
874                .bucket_size(2.try_into().unwrap())
875                .with_lru(LruConfig {
876                    counter_bits: 8.try_into().unwrap(),
877                    ..Default::default()
878                })
879                .build()
880                .unwrap(),
881            TestHasher(0),
882        );
883
884        let test_item = PredefinedBucketItem(2 << 32);
885        filter.insert(&test_item);
886        filter.contains(&test_item); // Make it more used than others
887
888        let test_item_2 = PredefinedBucketItem(4 << 32);
889        filter.insert(&test_item_2); // Sharing the same bucket as "test", but less used
890
891        let test_item_3 = PredefinedBucketItem((3 << 32) + 2);
892        filter.insert(&test_item_3); // Another bucket, but also valid for "test" bucket
893        filter.contains(&test_item_3); // Make it more used
894
895        let test_item_4 = PredefinedBucketItem((5 << 32) + 2);
896        filter.insert(&test_item_4); // Takes bucket of "test_item_3", but less used
897
898        // Everything fits now
899        assert!(filter.contains(&test_item));
900        assert!(filter.contains(&test_item_2));
901        assert!(filter.contains(&test_item_3));
902        assert!(filter.contains(&test_item_4));
903
904        let test_item_5 = PredefinedBucketItem((1 << 32) + 2);
905        // Insert a new item which has to take one of the 2 fully occupied buckets
906        filter.insert(&test_item_5);
907
908        assert!(filter.contains(&test_item_2));
909        assert!(filter.contains(&test_item));
910        assert!(filter.contains(&test_item_3));
911
912        assert!(
913            !filter.contains(&test_item_5) || !filter.contains(&test_item_4),
914            "No inserted items are missing, but filter can't hold them all"
915        );
916
917        // Insert both of these items again and confirm the more used ones are still there
918        filter.insert(&test_item_5);
919        filter.insert(&test_item_4);
920        assert!(filter.contains(&test_item));
921        assert!(filter.contains(&test_item_3));
922    }
923
924    #[test]
925    fn alt_index() {
926        let words = get_words(0..200_000);
927        let filter = CuckooFilter::new_random(
928            CuckooConfiguration::builder(200_000)
929                .fingerprint_bits(32.try_into().unwrap())
930                .build()
931                .unwrap(),
932        );
933
934        for word in words {
935            let (fp, index) = filter.get_fingerprint_and_index(&word);
936            let alt_index = filter.alt_index(&fp, index);
937            assert_eq!(index, filter.alt_index(&fp, alt_index));
938        }
939    }
940
941    #[test]
942    fn random_kicks() {
943        let filter = CuckooFilter::new(
944            CuckooConfiguration::builder(1000)
945                .bucket_size(2.try_into().unwrap())
946                .build()
947                .unwrap(),
948            TestHasher(0),
949        );
950
951        let test_item = PredefinedBucketItem(2 << 32);
952        filter.insert(&test_item);
953
954        let test_item_2 = PredefinedBucketItem(4 << 32);
955        filter.insert(&test_item_2); // Sharing the same bucket as "test"
956
957        let test_item_3 = PredefinedBucketItem((3 << 32) + 2);
958        filter.insert(&test_item_3); // Another bucket, but also valid for "test" bucket
959
960        let test_item_4 = PredefinedBucketItem((5 << 32) + 2);
961        filter.insert(&test_item_4); // Takes bucket of "test_item_3"
962
963        // This one should not be kicked, because it takes an unrelated bucket
964        let test_item_unrelated = PredefinedBucketItem((10 << 32) + 10);
965        filter.insert(&test_item_unrelated);
966
967        // Everything fits now
968        assert!(filter.contains(&test_item));
969        assert!(filter.contains(&test_item_2));
970        assert!(filter.contains(&test_item_3));
971        assert!(filter.contains(&test_item_4));
972
973        let test_item_5 = PredefinedBucketItem((1 << 32) + 2);
974        // Insert a new item which has to take one of the 2 fully occupied buckets
975        let kicked = filter.insert(&test_item_5);
976        assert!(kicked.is_some(), "An item had to be kicked");
977        assert!(filter.contains(&test_item_5));
978        assert!(filter.contains(&test_item_unrelated));
979
980        for item in [&test_item, &test_item_2, &test_item_3, &test_item_4]
981            .iter()
982            .filter(|i| !kicked.as_ref().unwrap().matches_key(i, &filter))
983        {
984            assert!(filter.contains(item), "Only one item should be kicked");
985        }
986    }
987
988    #[test]
989    fn overriding_defaults() {
990        let filter = CuckooFilter::new_random(
991            CuckooConfiguration::builder(1000)
992                .with_ttl(TtlConfig {
993                    ttl: 30.try_into().unwrap(),
994                    ttl_bits: 8.try_into().unwrap(),
995                })
996                .with_counter(CounterConfig::default())
997                .build()
998                .unwrap(),
999        );
1000
1001        filter.insert_with_defaults(
1002            "basic",
1003            InsertValues {
1004                ttl: Some(50),
1005                counter: Some(10),
1006            },
1007        );
1008
1009        assert!(filter.contains("basic"));
1010        assert_eq!(
1011            filter
1012                .get_associated_data("basic")
1013                .unwrap()
1014                .get_stored_ttl_value()
1015                .unwrap(),
1016            50
1017        );
1018        assert_eq!(
1019            filter
1020                .get_associated_data("basic")
1021                .unwrap()
1022                .get_counter()
1023                .unwrap(),
1024            13 // initial 10 + 1 on contains + 2x1 on get_associated_data
1025        );
1026    }
1027
1028    #[test]
1029    fn overriding_updates() {
1030        let filter = CuckooFilter::new_random(
1031            CuckooConfiguration::builder(1000)
1032                .with_ttl(TtlConfig {
1033                    ttl: 30.try_into().unwrap(),
1034                    ttl_bits: 8.try_into().unwrap(),
1035                })
1036                .with_counter(CounterConfig::default())
1037                .build()
1038                .unwrap(),
1039        );
1040
1041        filter.insert_with_defaults(
1042            "basic",
1043            InsertValues {
1044                ttl: Some(5),
1045                counter: Some(1),
1046            },
1047        );
1048
1049        assert!(filter.contains_with_update(
1050            "basic",
1051            LookupValues {
1052                ttl: Some(50),
1053                counter_diff: Some(10),
1054            },
1055        ));
1056        assert_eq!(
1057            filter
1058                .get_associated_data("basic")
1059                .unwrap()
1060                .get_stored_ttl_value()
1061                .unwrap(),
1062            50
1063        );
1064        assert_eq!(
1065            filter
1066                .get_associated_data("basic")
1067                .unwrap()
1068                .get_counter()
1069                .unwrap(),
1070            13 // initial 1 + 10 on contains + 2x1 on get_associated_data
1071        );
1072    }
1073
1074    #[test]
1075    fn scan_and_update_full() {
1076        let words = get_words(0..100_000);
1077        let filter = CuckooFilter::new_random(
1078            CuckooConfiguration::builder(100_000)
1079                .fingerprint_bits(32.try_into().unwrap())
1080                .with_lru(LruConfig::default())
1081                .with_ttl(TtlConfig {
1082                    ttl: 3.try_into().unwrap(),
1083                    ttl_bits: 2.try_into().unwrap(),
1084                })
1085                .build()
1086                .unwrap(),
1087        );
1088
1089        assert_eq!(filter.get_item_count(), 0);
1090
1091        let mut stored_words = HashSet::new();
1092
1093        for (index, word) in words.iter().enumerate() {
1094            stored_words.insert(word);
1095            if let Some(evicted_fp) = filter.insert(word) {
1096                words[0..=index]
1097                    .iter()
1098                    .filter(|w| evicted_fp.matches_key(w, &filter))
1099                    .for_each(|evicted_word| {
1100                        stored_words.remove(evicted_word);
1101                    });
1102            }
1103        }
1104
1105        assert_eq!(filter.get_item_count(), stored_words.len());
1106
1107        for _ in 0..2 {
1108            assert_eq!(filter.scan_and_update_full(), 0);
1109        }
1110
1111        assert_eq!(filter.get_item_count(), stored_words.len());
1112        for word in stored_words.iter() {
1113            assert!(
1114                filter.contains(word),
1115                "Word: {word} expected in the filter, but not found"
1116            );
1117        }
1118
1119        // TTL should remove all entries now
1120        assert_eq!(filter.scan_and_update_full(), stored_words.len());
1121        for word in &words {
1122            assert!(
1123                !filter.contains(word),
1124                "Filter contained {word}, but shouldn't have"
1125            );
1126        }
1127        assert_eq!(filter.get_item_count(), 0);
1128    }
1129
1130    #[test]
1131    fn scan_and_update_lru_deletion() {
1132        let words = get_words(0..100_000);
1133        let filter = CuckooFilter::new_random(
1134            CuckooConfiguration::builder(100_000)
1135                .fingerprint_bits(32.try_into().unwrap())
1136                .with_lru(LruConfig {
1137                    counter_bits: 2.try_into().unwrap(),
1138                    remove_on_zero: true,
1139                    ..Default::default()
1140                })
1141                .build()
1142                .unwrap(),
1143        );
1144
1145        assert_eq!(filter.get_item_count(), 0);
1146
1147        let mut stored_words = HashSet::new();
1148
1149        for (index, word) in words.iter().enumerate() {
1150            stored_words.insert(word);
1151            if let Some(evicted_fp) = filter.insert_if_not_present(word) {
1152                words[0..=index]
1153                    .iter()
1154                    .filter(|w| evicted_fp.matches_key(w, &filter))
1155                    .for_each(|evicted_word| {
1156                        stored_words.remove(evicted_word);
1157                    });
1158            }
1159        }
1160
1161        let mut kept_words = HashSet::new();
1162        for word in stored_words.clone().into_iter() {
1163            kept_words.insert(word);
1164            if let Some(evicted_fp) = filter.insert_if_not_present(word) {
1165                words
1166                    .iter()
1167                    .filter(|w| evicted_fp.matches_key(w, &filter))
1168                    .for_each(|evicted_word| {
1169                        stored_words.remove(evicted_word);
1170                        kept_words.remove(evicted_word);
1171                    });
1172            }
1173        }
1174
1175        assert_eq!(filter.get_item_count(), stored_words.len());
1176
1177        for _ in 0..2 {
1178            assert_eq!(filter.get_item_count(), stored_words.len());
1179            assert_eq!(filter.scan_and_update_full(), 0);
1180        }
1181
1182        assert_eq!(filter.get_item_count(), kept_words.len());
1183        for word in kept_words.iter() {
1184            assert!(
1185                filter.contains(word),
1186                "Word: {word} expected in the filter, but not found"
1187            );
1188        }
1189        for word in stored_words.difference(&kept_words) {
1190            assert!(
1191                !filter.contains(word),
1192                "Filter contained {word}, but shouldn't have"
1193            );
1194        }
1195
1196        // Contains check above incremented LRU counters again, lets bring them back to 0
1197        assert_eq!(filter.scan_and_update_full(), 0);
1198
1199        // LRU should remove all entries now
1200        assert_eq!(filter.scan_and_update_full(), kept_words.len());
1201        for word in &words {
1202            assert!(
1203                !filter.contains(word),
1204                "Filter contained {word}, but shouldn't have"
1205            );
1206        }
1207        assert_eq!(filter.get_item_count(), 0);
1208    }
1209
1210    #[test]
1211    fn scan_and_update_lru_deletion_decrement_strategy() {
1212        let words = get_words(0..100_000);
1213        let filter = CuckooFilter::new_random(
1214            CuckooConfiguration::builder(100_000)
1215                .fingerprint_bits(32.try_into().unwrap())
1216                .with_lru(LruConfig {
1217                    counter_bits: 2.try_into().unwrap(),
1218                    starting_value: 3,
1219                    aging_strategy: crate::config::LruAgingStrategy::Decrement(1),
1220                    remove_on_zero: true,
1221                    ..Default::default()
1222                })
1223                .build()
1224                .unwrap(),
1225        );
1226
1227        assert_eq!(filter.get_item_count(), 0);
1228
1229        let mut stored_words = HashSet::new();
1230
1231        for (index, word) in words.iter().enumerate() {
1232            stored_words.insert(word);
1233            if let Some(evicted_fp) = filter.insert_if_not_present(word) {
1234                words[0..=index]
1235                    .iter()
1236                    .filter(|w| evicted_fp.matches_key(w, &filter))
1237                    .for_each(|evicted_word| {
1238                        stored_words.remove(evicted_word);
1239                    });
1240            }
1241        }
1242
1243        assert_eq!(filter.get_item_count(), stored_words.len());
1244
1245        for _ in 0..3 {
1246            assert_eq!(filter.get_item_count(), stored_words.len());
1247            assert_eq!(filter.scan_and_update_full(), 0);
1248        }
1249
1250        assert_eq!(filter.get_item_count(), stored_words.len());
1251        for word in stored_words.iter() {
1252            assert!(
1253                filter.contains(word),
1254                "Word: {word} expected in the filter, but not found"
1255            );
1256        }
1257
1258        // Contains check above incremented LRU counters again, lets bring them back to 0
1259        assert_eq!(filter.scan_and_update_full(), 0);
1260
1261        // LRU should remove all entries now
1262        assert_eq!(filter.scan_and_update_full(), stored_words.len());
1263        for word in &words {
1264            assert!(
1265                !filter.contains(word),
1266                "Filter contained {word}, but shouldn't have"
1267            );
1268        }
1269        assert_eq!(filter.get_item_count(), 0);
1270    }
1271
1272    #[test]
1273    fn export_import() {
1274        let words = get_words(0..100_000);
1275        let filter = CuckooFilter::new_random_exportable(
1276            CuckooConfiguration::builder(100_000)
1277                .fingerprint_bits(32.try_into().unwrap())
1278                .with_lru(LruConfig::default())
1279                .with_ttl(TtlConfig {
1280                    ttl: 3.try_into().unwrap(),
1281                    ttl_bits: 2.try_into().unwrap(),
1282                })
1283                .build()
1284                .unwrap(),
1285        );
1286
1287        assert_eq!(filter.get_item_count(), 0);
1288
1289        let mut stored_words = HashSet::new();
1290
1291        for (index, word) in words.iter().enumerate() {
1292            stored_words.insert(word);
1293            if let Some(evicted_fp) = filter.insert(word) {
1294                words[0..=index]
1295                    .iter()
1296                    .filter(|w| evicted_fp.matches_key(w, &filter))
1297                    .for_each(|evicted_word| {
1298                        stored_words.remove(evicted_word);
1299                    });
1300            }
1301        }
1302
1303        assert_eq!(filter.get_item_count(), stored_words.len());
1304
1305        let exported_buf = filter.exporter().snapshot().unwrap();
1306        let mut readable_buf = VecDeque::from(exported_buf);
1307
1308        let imported_filter = CuckooFilter::import_random_exportable(&mut readable_buf).unwrap();
1309
1310        assert_eq!(
1311            imported_filter.get_configuration(),
1312            filter.get_configuration()
1313        );
1314
1315        assert_eq!(imported_filter.get_item_count(), stored_words.len());
1316        for word in stored_words.iter() {
1317            assert!(
1318                imported_filter.contains(word),
1319                "Word: {word} expected in the filter, but not found"
1320            );
1321        }
1322    }
1323}
cuckoo_clock/filter.rs

cuckoo_clock/
filter.rs
