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        if removed > 0 {
731            self.items.fetch_sub(removed, Ordering::Relaxed);
732        }
733        removed
734    }
735
736    /// Generates the fingerprint and first index for the provided key.
737    pub(crate) fn get_fingerprint<K: Hash + ?Sized>(&self, key: &K) -> Fingerprint {
738        self.get_fingerprint_and_index(key).0
739    }
740
741    fn new_data_block(&self, fp: &Fingerprint, defaults: InsertValues) -> DataBlock<Vec<u8>> {
742        let data = vec![0u8; self.configuration.data_block_size];
743        let mut cur_data_block = DataBlock::from(data);
744        cur_data_block.store_fingerprint(fp, &self.configuration);
745
746        if let Some(ttl_config) = self.configuration.ttl_field_config.as_ref() {
747            cur_data_block.set_ttl(ttl_config, defaults.ttl.unwrap_or(ttl_config.0.ttl.into()));
748        }
749        if let Some(counter_config) = self.configuration.counter_field_config.as_ref() {
750            cur_data_block.update_counter(
751                counter_config,
752                defaults
753                    .counter
754                    .unwrap_or(counter_config.0.change_on_insert),
755            );
756        }
757        if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
758            cur_data_block.init_lru_counter(lru_config);
759        }
760        cur_data_block
761    }
762
763    fn get_fingerprint_and_index<K: Hash + ?Sized>(&self, key: &K) -> (Fingerprint, u32) {
764        let result = self.build_hasher.hash_one(key);
765
766        // Fingeprint bits over 32 are definitely an overkill
767        // We can reduce number of hashes by using one hash as fingerprint and first index
768        let fingerprint = (result >> 32) as u32;
769        // Intentional truncation here
770        #[expect(clippy::cast_possible_truncation)]
771        let index = result as u32 & self.configuration.buckets_mask;
772
773        (
774            Fingerprint::new(
775                fingerprint,
776                self.configuration.fingerprint_field_config.value_mask(),
777            ),
778            index,
779        )
780    }
781
782    // Intentional truncation here
783    #[expect(clippy::cast_possible_truncation)]
784    fn alt_index(&self, fingerprint: &Fingerprint, index: u32) -> u32 {
785        let result = self.build_hasher.hash_one(fingerprint);
786
787        (index ^ ((result as u32) & self.configuration.buckets_mask))
788            & self.configuration.buckets_mask
789    }
790
791    #[expect(clippy::unwrap_used)]
792    fn lock_bucket(&self, index: usize) -> MutexGuard<'_, Bucket> {
793        // Any panic while lock is held should come from this library
794        // Any panic produced while the lock is held is a bug in the library!
795        self.buckets[index].lock().unwrap()
796    }
797}
798
799#[cfg(test)]
800#[expect(clippy::unwrap_used)]
801mod tests {
802    use std::{
803        collections::{HashSet, VecDeque},
804        hash::Hasher,
805        ops::Range,
806    };
807
808    use crate::config::{CounterConfig, LruConfig, TtlConfig};
809
810    use super::*;
811
812    fn get_words(range: Range<usize>) -> Vec<String> {
813        std::fs::read_to_string("/usr/share/dict/words")
814            .unwrap()
815            .split("\n")
816            .skip(range.start)
817            .take(range.len())
818            .map(ToString::to_string)
819            .collect()
820    }
821
822    #[test]
823    fn basic_insertion() {
824        let filter = CuckooFilter::new_random(CuckooConfiguration::builder(1000).build().unwrap());
825
826        filter.insert("basic");
827
828        assert!(filter.contains("basic"));
829    }
830
831    #[test]
832    fn basic_removal() {
833        let filter = CuckooFilter::new_random(CuckooConfiguration::builder(1000).build().unwrap());
834
835        filter.insert("basic");
836
837        assert!(filter.contains("basic"));
838
839        filter.remove("basic");
840
841        assert!(!filter.contains("basic"));
842    }
843
844    struct PredefinedBucketItem(u64);
845    struct TestHasher(u64);
846    impl BuildHasher for TestHasher {
847        type Hasher = TestHasher;
848
849        fn build_hasher(&self) -> Self::Hasher {
850            TestHasher(0)
851        }
852    }
853    impl Hasher for TestHasher {
854        fn finish(&self) -> u64 {
855            self.0
856        }
857
858        fn write(&mut self, bytes: &[u8]) {
859            if bytes.len() == 8 {
860                self.0 = u64::from_ne_bytes(bytes.try_into().unwrap());
861            } else {
862                // Shift fingeprint hashes a bit, to allow control
863                self.0 = 1 - (u32::from_ne_bytes(bytes.try_into().unwrap()) as u64 % 2);
864            }
865        }
866    }
867    impl Hash for PredefinedBucketItem {
868        fn hash<H: Hasher>(&self, state: &mut H) {
869            state.write_u64(self.0);
870        }
871    }
872
873    #[test]
874    fn lru_insertion() {
875        let filter = CuckooFilter::new(
876            CuckooConfiguration::builder(1000)
877                .bucket_size(2.try_into().unwrap())
878                .with_lru(LruConfig {
879                    counter_bits: 8.try_into().unwrap(),
880                    ..Default::default()
881                })
882                .build()
883                .unwrap(),
884            TestHasher(0),
885        );
886
887        let test_item = PredefinedBucketItem(2 << 32);
888        filter.insert(&test_item);
889        filter.contains(&test_item); // Make it more used than others
890
891        let test_item_2 = PredefinedBucketItem(4 << 32);
892        filter.insert(&test_item_2); // Sharing the same bucket as "test", but less used
893
894        let test_item_3 = PredefinedBucketItem((3 << 32) + 2);
895        filter.insert(&test_item_3); // Another bucket, but also valid for "test" bucket
896        filter.contains(&test_item_3); // Make it more used
897
898        let test_item_4 = PredefinedBucketItem((5 << 32) + 2);
899        filter.insert(&test_item_4); // Takes bucket of "test_item_3", but less used
900
901        // Everything fits now
902        assert!(filter.contains(&test_item));
903        assert!(filter.contains(&test_item_2));
904        assert!(filter.contains(&test_item_3));
905        assert!(filter.contains(&test_item_4));
906
907        let test_item_5 = PredefinedBucketItem((1 << 32) + 2);
908        // Insert a new item which has to take one of the 2 fully occupied buckets
909        filter.insert(&test_item_5);
910
911        assert!(filter.contains(&test_item_2));
912        assert!(filter.contains(&test_item));
913        assert!(filter.contains(&test_item_3));
914
915        assert!(
916            !filter.contains(&test_item_5) || !filter.contains(&test_item_4),
917            "No inserted items are missing, but filter can't hold them all"
918        );
919
920        // Insert both of these items again and confirm the more used ones are still there
921        filter.insert(&test_item_5);
922        filter.insert(&test_item_4);
923        assert!(filter.contains(&test_item));
924        assert!(filter.contains(&test_item_3));
925    }
926
927    #[test]
928    fn alt_index() {
929        let words = get_words(0..200_000);
930        let filter = CuckooFilter::new_random(
931            CuckooConfiguration::builder(200_000)
932                .fingerprint_bits(32.try_into().unwrap())
933                .build()
934                .unwrap(),
935        );
936
937        for word in words {
938            let (fp, index) = filter.get_fingerprint_and_index(&word);
939            let alt_index = filter.alt_index(&fp, index);
940            assert_eq!(index, filter.alt_index(&fp, alt_index));
941        }
942    }
943
944    #[test]
945    fn random_kicks() {
946        let filter = CuckooFilter::new(
947            CuckooConfiguration::builder(1000)
948                .bucket_size(2.try_into().unwrap())
949                .build()
950                .unwrap(),
951            TestHasher(0),
952        );
953
954        let test_item = PredefinedBucketItem(2 << 32);
955        filter.insert(&test_item);
956
957        let test_item_2 = PredefinedBucketItem(4 << 32);
958        filter.insert(&test_item_2); // Sharing the same bucket as "test"
959
960        let test_item_3 = PredefinedBucketItem((3 << 32) + 2);
961        filter.insert(&test_item_3); // Another bucket, but also valid for "test" bucket
962
963        let test_item_4 = PredefinedBucketItem((5 << 32) + 2);
964        filter.insert(&test_item_4); // Takes bucket of "test_item_3"
965
966        // This one should not be kicked, because it takes an unrelated bucket
967        let test_item_unrelated = PredefinedBucketItem((10 << 32) + 10);
968        filter.insert(&test_item_unrelated);
969
970        // Everything fits now
971        assert!(filter.contains(&test_item));
972        assert!(filter.contains(&test_item_2));
973        assert!(filter.contains(&test_item_3));
974        assert!(filter.contains(&test_item_4));
975
976        let test_item_5 = PredefinedBucketItem((1 << 32) + 2);
977        // Insert a new item which has to take one of the 2 fully occupied buckets
978        let kicked = filter.insert(&test_item_5);
979        assert!(kicked.is_some(), "An item had to be kicked");
980        assert!(filter.contains(&test_item_5));
981        assert!(filter.contains(&test_item_unrelated));
982
983        for item in [&test_item, &test_item_2, &test_item_3, &test_item_4]
984            .iter()
985            .filter(|i| !kicked.as_ref().unwrap().matches_key(i, &filter))
986        {
987            assert!(filter.contains(item), "Only one item should be kicked");
988        }
989    }
990
991    #[test]
992    fn overriding_defaults() {
993        let filter = CuckooFilter::new_random(
994            CuckooConfiguration::builder(1000)
995                .with_ttl(TtlConfig {
996                    ttl: 30.try_into().unwrap(),
997                    ttl_bits: 8.try_into().unwrap(),
998                })
999                .with_counter(CounterConfig::default())
1000                .build()
1001                .unwrap(),
1002        );
1003
1004        filter.insert_with_defaults(
1005            "basic",
1006            InsertValues {
1007                ttl: Some(50),
1008                counter: Some(10),
1009            },
1010        );
1011
1012        assert!(filter.contains("basic"));
1013        assert_eq!(
1014            filter
1015                .get_associated_data("basic")
1016                .unwrap()
1017                .get_stored_ttl_value()
1018                .unwrap(),
1019            50
1020        );
1021        assert_eq!(
1022            filter
1023                .get_associated_data("basic")
1024                .unwrap()
1025                .get_counter()
1026                .unwrap(),
1027            13 // initial 10 + 1 on contains + 2x1 on get_associated_data
1028        );
1029    }
1030
1031    #[test]
1032    fn overriding_updates() {
1033        let filter = CuckooFilter::new_random(
1034            CuckooConfiguration::builder(1000)
1035                .with_ttl(TtlConfig {
1036                    ttl: 30.try_into().unwrap(),
1037                    ttl_bits: 8.try_into().unwrap(),
1038                })
1039                .with_counter(CounterConfig::default())
1040                .build()
1041                .unwrap(),
1042        );
1043
1044        filter.insert_with_defaults(
1045            "basic",
1046            InsertValues {
1047                ttl: Some(5),
1048                counter: Some(1),
1049            },
1050        );
1051
1052        assert!(filter.contains_with_update(
1053            "basic",
1054            LookupValues {
1055                ttl: Some(50),
1056                counter_diff: Some(10),
1057            },
1058        ));
1059        assert_eq!(
1060            filter
1061                .get_associated_data("basic")
1062                .unwrap()
1063                .get_stored_ttl_value()
1064                .unwrap(),
1065            50
1066        );
1067        assert_eq!(
1068            filter
1069                .get_associated_data("basic")
1070                .unwrap()
1071                .get_counter()
1072                .unwrap(),
1073            13 // initial 1 + 10 on contains + 2x1 on get_associated_data
1074        );
1075    }
1076
1077    #[test]
1078    fn scan_and_update_full() {
1079        let words = get_words(0..100_000);
1080        let filter = CuckooFilter::new_random(
1081            CuckooConfiguration::builder(100_000)
1082                .fingerprint_bits(32.try_into().unwrap())
1083                .with_lru(LruConfig::default())
1084                .with_ttl(TtlConfig {
1085                    ttl: 3.try_into().unwrap(),
1086                    ttl_bits: 2.try_into().unwrap(),
1087                })
1088                .build()
1089                .unwrap(),
1090        );
1091
1092        assert_eq!(filter.get_item_count(), 0);
1093
1094        let mut stored_words = HashSet::new();
1095
1096        for (index, word) in words.iter().enumerate() {
1097            stored_words.insert(word);
1098            if let Some(evicted_fp) = filter.insert(word) {
1099                words[0..=index]
1100                    .iter()
1101                    .filter(|w| evicted_fp.matches_key(w, &filter))
1102                    .for_each(|evicted_word| {
1103                        stored_words.remove(evicted_word);
1104                    });
1105            }
1106        }
1107
1108        assert_eq!(filter.get_item_count(), stored_words.len());
1109
1110        for _ in 0..2 {
1111            assert_eq!(filter.scan_and_update_full(), 0);
1112        }
1113
1114        assert_eq!(filter.get_item_count(), stored_words.len());
1115        for word in stored_words.iter() {
1116            assert!(
1117                filter.contains(word),
1118                "Word: {word} expected in the filter, but not found"
1119            );
1120        }
1121
1122        // TTL should remove all entries now
1123        assert_eq!(filter.scan_and_update_full(), stored_words.len());
1124        for word in &words {
1125            assert!(
1126                !filter.contains(word),
1127                "Filter contained {word}, but shouldn't have"
1128            );
1129        }
1130        assert_eq!(filter.get_item_count(), 0);
1131    }
1132
1133    #[test]
1134    fn scan_and_update_lru_deletion() {
1135        let words = get_words(0..100_000);
1136        let filter = CuckooFilter::new_random(
1137            CuckooConfiguration::builder(100_000)
1138                .fingerprint_bits(32.try_into().unwrap())
1139                .with_lru(LruConfig {
1140                    counter_bits: 2.try_into().unwrap(),
1141                    remove_on_zero: true,
1142                    ..Default::default()
1143                })
1144                .build()
1145                .unwrap(),
1146        );
1147
1148        assert_eq!(filter.get_item_count(), 0);
1149
1150        let mut stored_words = HashSet::new();
1151
1152        for (index, word) in words.iter().enumerate() {
1153            stored_words.insert(word);
1154            if let Some(evicted_fp) = filter.insert_if_not_present(word) {
1155                words[0..=index]
1156                    .iter()
1157                    .filter(|w| evicted_fp.matches_key(w, &filter))
1158                    .for_each(|evicted_word| {
1159                        stored_words.remove(evicted_word);
1160                    });
1161            }
1162        }
1163
1164        let mut kept_words = HashSet::new();
1165        for word in stored_words.clone().into_iter() {
1166            kept_words.insert(word);
1167            if let Some(evicted_fp) = filter.insert_if_not_present(word) {
1168                words
1169                    .iter()
1170                    .filter(|w| evicted_fp.matches_key(w, &filter))
1171                    .for_each(|evicted_word| {
1172                        stored_words.remove(evicted_word);
1173                        kept_words.remove(evicted_word);
1174                    });
1175            }
1176        }
1177
1178        assert_eq!(filter.get_item_count(), stored_words.len());
1179
1180        for _ in 0..2 {
1181            assert_eq!(filter.get_item_count(), stored_words.len());
1182            assert_eq!(filter.scan_and_update_full(), 0);
1183        }
1184
1185        assert_eq!(filter.get_item_count(), kept_words.len());
1186        for word in kept_words.iter() {
1187            assert!(
1188                filter.contains(word),
1189                "Word: {word} expected in the filter, but not found"
1190            );
1191        }
1192        for word in stored_words.difference(&kept_words) {
1193            assert!(
1194                !filter.contains(word),
1195                "Filter contained {word}, but shouldn't have"
1196            );
1197        }
1198
1199        // Contains check above incremented LRU counters again, lets bring them back to 0
1200        assert_eq!(filter.scan_and_update_full(), 0);
1201
1202        // LRU should remove all entries now
1203        assert_eq!(filter.scan_and_update_full(), kept_words.len());
1204        for word in &words {
1205            assert!(
1206                !filter.contains(word),
1207                "Filter contained {word}, but shouldn't have"
1208            );
1209        }
1210        assert_eq!(filter.get_item_count(), 0);
1211    }
1212
1213    #[test]
1214    fn scan_and_update_lru_deletion_decrement_strategy() {
1215        let words = get_words(0..100_000);
1216        let filter = CuckooFilter::new_random(
1217            CuckooConfiguration::builder(100_000)
1218                .fingerprint_bits(32.try_into().unwrap())
1219                .with_lru(LruConfig {
1220                    counter_bits: 2.try_into().unwrap(),
1221                    starting_value: 3,
1222                    aging_strategy: crate::config::LruAgingStrategy::Decrement(1),
1223                    remove_on_zero: true,
1224                    ..Default::default()
1225                })
1226                .build()
1227                .unwrap(),
1228        );
1229
1230        assert_eq!(filter.get_item_count(), 0);
1231
1232        let mut stored_words = HashSet::new();
1233
1234        for (index, word) in words.iter().enumerate() {
1235            stored_words.insert(word);
1236            if let Some(evicted_fp) = filter.insert_if_not_present(word) {
1237                words[0..=index]
1238                    .iter()
1239                    .filter(|w| evicted_fp.matches_key(w, &filter))
1240                    .for_each(|evicted_word| {
1241                        stored_words.remove(evicted_word);
1242                    });
1243            }
1244        }
1245
1246        assert_eq!(filter.get_item_count(), stored_words.len());
1247
1248        for _ in 0..3 {
1249            assert_eq!(filter.get_item_count(), stored_words.len());
1250            assert_eq!(filter.scan_and_update_full(), 0);
1251        }
1252
1253        assert_eq!(filter.get_item_count(), stored_words.len());
1254        for word in stored_words.iter() {
1255            assert!(
1256                filter.contains(word),
1257                "Word: {word} expected in the filter, but not found"
1258            );
1259        }
1260
1261        // Contains check above incremented LRU counters again, lets bring them back to 0
1262        assert_eq!(filter.scan_and_update_full(), 0);
1263
1264        // LRU should remove all entries now
1265        assert_eq!(filter.scan_and_update_full(), stored_words.len());
1266        for word in &words {
1267            assert!(
1268                !filter.contains(word),
1269                "Filter contained {word}, but shouldn't have"
1270            );
1271        }
1272        assert_eq!(filter.get_item_count(), 0);
1273    }
1274
1275    #[test]
1276    fn export_import() {
1277        let words = get_words(0..100_000);
1278        let filter = CuckooFilter::new_random_exportable(
1279            CuckooConfiguration::builder(100_000)
1280                .fingerprint_bits(32.try_into().unwrap())
1281                .with_lru(LruConfig::default())
1282                .with_ttl(TtlConfig {
1283                    ttl: 3.try_into().unwrap(),
1284                    ttl_bits: 2.try_into().unwrap(),
1285                })
1286                .build()
1287                .unwrap(),
1288        );
1289
1290        assert_eq!(filter.get_item_count(), 0);
1291
1292        let mut stored_words = HashSet::new();
1293
1294        for (index, word) in words.iter().enumerate() {
1295            stored_words.insert(word);
1296            if let Some(evicted_fp) = filter.insert(word) {
1297                words[0..=index]
1298                    .iter()
1299                    .filter(|w| evicted_fp.matches_key(w, &filter))
1300                    .for_each(|evicted_word| {
1301                        stored_words.remove(evicted_word);
1302                    });
1303            }
1304        }
1305
1306        assert_eq!(filter.get_item_count(), stored_words.len());
1307
1308        let exported_buf = filter.exporter().snapshot().unwrap();
1309        let mut readable_buf = VecDeque::from(exported_buf);
1310
1311        let imported_filter = CuckooFilter::import_random_exportable(&mut readable_buf).unwrap();
1312
1313        assert_eq!(
1314            imported_filter.get_configuration(),
1315            filter.get_configuration()
1316        );
1317
1318        assert_eq!(imported_filter.get_item_count(), stored_words.len());
1319        for word in stored_words.iter() {
1320            assert!(
1321                imported_filter.contains(word),
1322                "Word: {word} expected in the filter, but not found"
1323            );
1324        }
1325    }
1326}
cuckoo_clock/filter.rs

cuckoo_clock/
filter.rs
