Skip to main content

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, ExportableBuildHasher, ExportableRandomState, import_config,
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 = import_config(&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        for _ in 0..self.configuration.max_kicks {
343            {
344                let mut bucket = self.lock_bucket(cur_index as usize);
345                // Replace a random item first
346                if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
347                    if !bucket.kick_lru(&mut cur_data_block, &self.configuration, lru_config) {
348                        return Some(cur_data_block.get_fingerprint(&self.configuration));
349                    }
350                } else {
351                    bucket.kick_random(&mut cur_data_block, &self.configuration);
352                }
353                cur_index = self.alt_index(
354                    &cur_data_block.get_fingerprint(&self.configuration),
355                    cur_index,
356                );
357            }
358
359            if self
360                .lock_bucket(cur_index as usize)
361                .insert(&cur_data_block, &self.configuration)
362            {
363                self.items.fetch_add(1, Ordering::Relaxed);
364                // Found an alternative spot for evicted item, done with kicks
365                return None;
366            }
367        }
368
369        // Filter is full
370        Some(cur_data_block.get_fingerprint(&self.configuration))
371    }
372
373    /// Inserts a new item into the filter.
374    ///
375    /// If both target buckets for this item are full, random item is kicked out of one of these 2
376    /// buckets and moved into its alternate bucket, starting a recursive kicking process, which
377    /// stops once an empty slot is found in alternate bucket of a kicked item, or
378    /// [`crate::config::CuckooConfigurationBuilder::max_kicks`] is reached.
379    ///
380    /// Returns fingerprint of the item that was evicted from the filter, if eviction had to take
381    /// place to finalize the insertion. It is possible that the item that was just inserted gets
382    /// evicted in random kicking process. That can be confirmed using
383    /// [`Fingerprint::matches_key`].
384    pub fn insert<K: Hash + ?Sized>(&self, key: &K) -> Option<Fingerprint> {
385        self.insert_with_defaults(key, InsertValues::default())
386    }
387
388    /// Inserts a new item into the filter, with defined defaults for associated data.
389    ///
390    /// This is similar to [`CuckooFilter::insert`], but allows overrides for associated data
391    /// defaults.
392    pub fn insert_with_defaults<K: Hash + ?Sized>(
393        &self,
394        key: &K,
395        default: InsertValues,
396    ) -> Option<Fingerprint> {
397        let (fp, i1) = self.get_fingerprint_and_index(key);
398        let mut cur_data_block = self.new_data_block(&fp, default);
399
400        let inserted = self
401            .lock_bucket(i1 as usize)
402            .insert(&cur_data_block, &self.configuration);
403
404        if inserted {
405            self.items.fetch_add(1, Ordering::Relaxed);
406            return None;
407        }
408
409        let i2 = self.alt_index(&fp, i1);
410
411        let inserted = self
412            .lock_bucket(i2 as usize)
413            .insert(&cur_data_block, &self.configuration);
414
415        if inserted {
416            self.items.fetch_add(1, Ordering::Relaxed);
417            return None;
418        }
419
420        let mut cur_index = i1;
421        for _ in 0..self.configuration.max_kicks {
422            {
423                let mut bucket = self.lock_bucket(cur_index as usize);
424                // Replace a random item first
425                if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
426                    if !bucket.kick_lru(&mut cur_data_block, &self.configuration, lru_config) {
427                        return Some(cur_data_block.get_fingerprint(&self.configuration));
428                    }
429                } else {
430                    // TODO: this can even kick the newest item, which is not ideal
431                    bucket.kick_random(&mut cur_data_block, &self.configuration);
432                }
433                cur_index = self.alt_index(
434                    &cur_data_block.get_fingerprint(&self.configuration),
435                    cur_index,
436                );
437            }
438
439            if self
440                .lock_bucket(cur_index as usize)
441                .insert(&cur_data_block, &self.configuration)
442            {
443                self.items.fetch_add(1, Ordering::Relaxed);
444                // Found an alternative spot for evicted item, done with kicks
445                return None;
446            }
447        }
448
449        // Filter is full
450        Some(cur_data_block.get_fingerprint(&self.configuration))
451    }
452
453    /// Check if this key is stored in the filter and applies the provided [`LookupValues`].
454    ///
455    /// This is similar to [`CuckooFilter::contains`], but allows overrides for updates on
456    /// successful lookup.
457    pub fn contains_with_update<K: Hash + ?Sized>(&self, key: &K, update: LookupValues) -> bool {
458        let (fp, i1) = self.get_fingerprint_and_index(key);
459
460        let mut contains =
461            self.lock_bucket(i1 as usize)
462                .contains(&fp, &self.configuration, &update);
463
464        if !contains {
465            let i2 = self.alt_index(&fp, i1);
466            contains = self
467                .lock_bucket(i2 as usize)
468                .contains(&fp, &self.configuration, &update);
469        }
470
471        contains
472    }
473
474    /// Check if this key is stored in the filter.
475    ///
476    /// Returns true if this key might be present in the filter. If false is returned, then the key
477    /// is definitely not present.
478    pub fn contains<K: Hash + ?Sized>(&self, key: &K) -> bool {
479        self.contains_with_update(key, LookupValues::default())
480    }
481
482    /// Loads associated data of a key stored in the filter.
483    ///
484    /// Returns None if this filter is not present in the filter. Returns associated data for the
485    /// first item with the fingerprint matching this key's fingerprint. Note that it is
486    /// recommended to use [`CuckooFilter::insert_if_not_present`] if consistent [`AssociatedData`]
487    /// is required.
488    pub fn get_associated_data<K: Hash + ?Sized>(&self, key: &K) -> Option<AssociatedData> {
489        self.get_associated_data_with_update(key, LookupValues::default())
490    }
491
492    /// Loads associated data of a key stored in the filter and applies the provided [`LookupValues`].
493    ///
494    /// This is similar to [`CuckooFilter::get_associated_data`], but allows overrides for updates on
495    /// successful lookup.
496    pub fn get_associated_data_with_update<K: Hash + ?Sized>(
497        &self,
498        key: &K,
499        update: LookupValues,
500    ) -> Option<AssociatedData> {
501        let (fp, i1) = self.get_fingerprint_and_index(key);
502
503        let mut contains =
504            self.lock_bucket(i1 as usize)
505                .get_associated_data(&fp, &self.configuration, &update);
506
507        if contains.is_none() {
508            let i2 = self.alt_index(&fp, i1);
509            contains = self.lock_bucket(i2 as usize).get_associated_data(
510                &fp,
511                &self.configuration,
512                &update,
513            );
514        }
515
516        contains
517    }
518
519    /// Removes this key from the filter, if present.
520    ///
521    /// Returns true if the key was present in the filter.
522    pub fn remove<K: Hash + ?Sized>(&self, key: &K) -> bool {
523        let (fp, i1) = self.get_fingerprint_and_index(key);
524
525        let mut removed = self
526            .lock_bucket(i1 as usize)
527            .remove(&fp, &self.configuration);
528
529        if !removed {
530            let i2 = self.alt_index(&fp, i1);
531            removed = self
532                .lock_bucket(i2 as usize)
533                .remove(&fp, &self.configuration);
534        }
535
536        if removed {
537            self.items.fetch_sub(1, Ordering::Relaxed);
538        }
539
540        removed
541    }
542
543    /// Scans all buckets of this filter and reduces TTL and LRU counters.
544    ///
545    /// If LRU and/or TTL features are used, this must be called periodically.
546    /// Each call to this function will age all the LRU and TTL counters. The frequency of calls
547    /// will affect both LRU and TTL in different ways:
548    /// - TTL will get reduced by 1 on each call, meaning that scanning each second indirectly sets
549    ///   the unit of TTL field to be seconds.
550    /// - LRU will get halved on each call. By scanning more frequently, items will require more
551    ///   frequent usage to stay in the filter.
552    ///
553    /// This is a no-op if both LRU and TTL are disabled.
554    ///
555    /// # Examples
556    ///
557    /// ```
558    /// use cuckoo_clock::{CuckooFilter, config::{CuckooConfiguration, CounterConfig, TtlConfig}};
559    ///
560    /// let filter = CuckooFilter::new_random(
561    ///     CuckooConfiguration::builder(10_000)
562    ///         .with_ttl(TtlConfig {
563    ///             ttl: 3.try_into()?,
564    ///             ttl_bits: 2.try_into()?
565    ///         })
566    ///         .build()?
567    /// );
568    ///
569    /// filter.insert("example_data");
570    ///
571    /// assert!(filter.contains("example_data"));
572    ///
573    /// filter.scan_and_update_full();
574    /// assert!(filter.contains("example_data"));
575    ///
576    /// filter.scan_and_update_full();
577    /// assert!(filter.contains("example_data"));
578    ///
579    /// // The item will get removed now, due to expired TTL
580    /// filter.scan_and_update_full();
581    /// assert!(!filter.contains("example_data"));
582    ///
583    /// # Ok::<(), Box<dyn std::error::Error>>(())
584    /// ```
585    pub fn scan_and_update_full(&self) -> usize {
586        #[expect(clippy::unwrap_used)]
587        self.scan_and_update_full_partition(NonZeroUsize::new(1).unwrap(), 0)
588    }
589
590    /// Scans a single group of buckets of this filter and reduces TTL and LRU counters.
591    ///
592    /// This is the same as [`CuckooFilter::scan_and_update_full`], but it more suitable for
593    /// parallelization, by splitting buckets into partitions to process in parallel.
594    pub fn scan_and_update_full_partition(
595        &self,
596        total_partitions: NonZeroUsize,
597        partition_index: usize,
598    ) -> usize {
599        if self.configuration.lru_field_config.is_none()
600            && self.configuration.ttl_field_config.is_none()
601        {
602            return 0;
603        }
604
605        let mut removed = 0;
606        let part_size = self.buckets.len().div_ceil(total_partitions.get());
607        if (partition_index * part_size) >= self.buckets.len() {
608            return 0;
609        }
610        for b in self.buckets
611            [partition_index * part_size..self.buckets.len().min((partition_index + 1) * part_size)]
612            .iter()
613        {
614            #[expect(clippy::unwrap_used)]
615            let mut bucket = b.lock().unwrap();
616            if let Some(lru_config) = &self.configuration.lru_field_config {
617                bucket.age_lru_counters(&self.configuration, lru_config);
618            }
619            if let Some(ttl_config) = &self.configuration.ttl_field_config {
620                removed += bucket.age_ttl_counters(&self.configuration, ttl_config);
621            }
622        }
623
624        if removed > 0 {
625            self.items.fetch_sub(removed, Ordering::Relaxed);
626        }
627        removed
628    }
629
630    /// Scans all buckets of this filter and reduces TTL counters.
631    ///
632    /// Similar to [`CuckooFilter::scan_and_update_full`], but updates only TTL counters. This
633    /// allows more control, enabling different update frequency for TTL and LRU.
634    ///
635    /// This is a no-op if TTL is disabled.
636    pub fn scan_and_update_ttl(&self) -> usize {
637        #[expect(clippy::unwrap_used)]
638        self.scan_and_update_ttl_partition(NonZeroUsize::new(1).unwrap(), 0)
639    }
640
641    /// Scans a single group of buckets of this filter and reduces TTL counters.
642    ///
643    /// This is the same as [`CuckooFilter::scan_and_update_ttl`], but it more suitable for
644    /// parallelization, by splitting buckets into partitions to process in parallel.
645    pub fn scan_and_update_ttl_partition(
646        &self,
647        total_partitions: NonZeroUsize,
648        partition_index: usize,
649    ) -> usize {
650        if self.configuration.ttl_field_config.is_none() {
651            return 0;
652        }
653
654        let mut removed = 0;
655        let part_size = self.buckets.len().div_ceil(total_partitions.get());
656        if (partition_index * part_size) >= self.buckets.len() {
657            return 0;
658        }
659        for b in self.buckets
660            [partition_index * part_size..self.buckets.len().min((partition_index + 1) * part_size)]
661            .iter()
662        {
663            #[expect(clippy::unwrap_used)]
664            let mut bucket = b.lock().unwrap();
665            if let Some(ttl_config) = &self.configuration.ttl_field_config {
666                removed += bucket.age_ttl_counters(&self.configuration, ttl_config);
667            }
668        }
669
670        if removed > 0 {
671            self.items.fetch_sub(removed, Ordering::Relaxed);
672        }
673        removed
674    }
675
676    /// Scans all buckets of this filter and reduces LRU counters.
677    ///
678    /// Similar to [`CuckooFilter::scan_and_update_full`], but updates only LRU counters. This
679    /// allows more control, enabling different update frequency for TTL and LRU.
680    ///
681    /// This is a no-op if LRU is disabled.
682    pub fn scan_and_update_lru(&self) {
683        #[expect(clippy::unwrap_used)]
684        self.scan_and_update_lru_partition(NonZeroUsize::new(1).unwrap(), 0)
685    }
686
687    /// Scans a single group of buckets of this filter and reduces LRU counters.
688    ///
689    /// This is the same as [`CuckooFilter::scan_and_update_lru`], but it more suitable for
690    /// parallelization, by splitting buckets into partitions to process in parallel.
691    pub fn scan_and_update_lru_partition(
692        &self,
693        total_partitions: NonZeroUsize,
694        partition_index: usize,
695    ) {
696        if self.configuration.lru_field_config.is_none() {
697            return;
698        }
699
700        let part_size = self.buckets.len().div_ceil(total_partitions.get());
701        if (partition_index * part_size) >= self.buckets.len() {
702            return;
703        }
704        for b in self.buckets
705            [partition_index * part_size..self.buckets.len().min((partition_index + 1) * part_size)]
706            .iter()
707        {
708            #[expect(clippy::unwrap_used)]
709            let mut bucket = b.lock().unwrap();
710            if let Some(lru_config) = &self.configuration.lru_field_config {
711                bucket.age_lru_counters(&self.configuration, lru_config);
712            }
713        }
714    }
715
716    /// Generates the fingerprint and first index for the provided key.
717    pub(crate) fn get_fingerprint<K: Hash + ?Sized>(&self, key: &K) -> Fingerprint {
718        self.get_fingerprint_and_index(key).0
719    }
720
721    fn new_data_block(&self, fp: &Fingerprint, defaults: InsertValues) -> DataBlock<Vec<u8>> {
722        let data = vec![0u8; self.configuration.data_block_size];
723        let mut cur_data_block = DataBlock::from(data);
724        cur_data_block.store_fingerprint(fp, &self.configuration);
725
726        if let Some(ttl_config) = self.configuration.ttl_field_config.as_ref() {
727            cur_data_block.set_ttl(ttl_config, defaults.ttl.unwrap_or(ttl_config.0.ttl.into()));
728        }
729        if let Some(counter_config) = self.configuration.counter_field_config.as_ref() {
730            cur_data_block.update_counter(
731                counter_config,
732                defaults
733                    .counter
734                    .unwrap_or(counter_config.0.change_on_insert),
735            );
736        }
737        if let Some(lru_config) = self.configuration.lru_field_config.as_ref() {
738            cur_data_block.inc_lru_counter(lru_config);
739        }
740        cur_data_block
741    }
742
743    fn get_fingerprint_and_index<K: Hash + ?Sized>(&self, key: &K) -> (Fingerprint, u32) {
744        let result = self.build_hasher.hash_one(key);
745
746        // Fingeprint bits over 32 are definitely an overkill
747        // We can reduce number of hashes by using one hash as fingerprint and first index
748        let fingerprint = (result >> 32) as u32;
749        // Intentional truncation here
750        #[expect(clippy::cast_possible_truncation)]
751        let index = result as u32 & self.configuration.buckets_mask;
752
753        (
754            Fingerprint::new(
755                fingerprint,
756                self.configuration.fingerprint_field_config.value_mask(),
757            ),
758            index,
759        )
760    }
761
762    // Intentional truncation here
763    #[expect(clippy::cast_possible_truncation)]
764    fn alt_index(&self, fingerprint: &Fingerprint, index: u32) -> u32 {
765        let result = self.build_hasher.hash_one(fingerprint);
766
767        (index ^ ((result as u32) & self.configuration.buckets_mask))
768            & self.configuration.buckets_mask
769    }
770
771    #[expect(clippy::unwrap_used)]
772    fn lock_bucket(&self, index: usize) -> MutexGuard<'_, Bucket> {
773        // Any panic while lock is held should come from this library
774        // Any panic produced while the lock is held is a bug in the library!
775        self.buckets[index].lock().unwrap()
776    }
777}
778
779#[cfg(test)]
780#[expect(clippy::unwrap_used)]
781mod tests {
782    use std::{
783        collections::{HashSet, VecDeque},
784        hash::Hasher,
785        ops::Range,
786    };
787
788    use crate::config::{CounterConfig, LruConfig, TtlConfig};
789
790    use super::*;
791
792    fn get_words(range: Range<usize>) -> Vec<String> {
793        std::fs::read_to_string("/usr/share/dict/words")
794            .unwrap()
795            .split("\n")
796            .skip(range.start)
797            .take(range.len())
798            .map(ToString::to_string)
799            .collect()
800    }
801
802    #[test]
803    fn basic_insertion() {
804        let filter = CuckooFilter::new_random(CuckooConfiguration::builder(1000).build().unwrap());
805
806        filter.insert("basic");
807
808        assert!(filter.contains("basic"));
809    }
810
811    #[test]
812    fn basic_removal() {
813        let filter = CuckooFilter::new_random(CuckooConfiguration::builder(1000).build().unwrap());
814
815        filter.insert("basic");
816
817        assert!(filter.contains("basic"));
818
819        filter.remove("basic");
820
821        assert!(!filter.contains("basic"));
822    }
823
824    struct PredefinedBucketItem(u64);
825    struct TestHasher(u64);
826    impl BuildHasher for TestHasher {
827        type Hasher = TestHasher;
828
829        fn build_hasher(&self) -> Self::Hasher {
830            TestHasher(0)
831        }
832    }
833    impl Hasher for TestHasher {
834        fn finish(&self) -> u64 {
835            self.0
836        }
837
838        fn write(&mut self, bytes: &[u8]) {
839            if bytes.len() == 8 {
840                self.0 = u64::from_ne_bytes(bytes.try_into().unwrap());
841            } else {
842                // Shift fingeprint hashes a bit, to allow control
843                self.0 = 1 - (u32::from_ne_bytes(bytes.try_into().unwrap()) as u64 % 2);
844            }
845        }
846    }
847    impl Hash for PredefinedBucketItem {
848        fn hash<H: Hasher>(&self, state: &mut H) {
849            state.write_u64(self.0);
850        }
851    }
852
853    #[test]
854    fn lru_insertion() {
855        let filter = CuckooFilter::new(
856            CuckooConfiguration::builder(1000)
857                .bucket_size(2.try_into().unwrap())
858                .with_lru(LruConfig {
859                    counter_bits: 8.try_into().unwrap(),
860                })
861                .build()
862                .unwrap(),
863            TestHasher(0),
864        );
865
866        let test_item = PredefinedBucketItem(2 << 32);
867        filter.insert(&test_item);
868        filter.contains(&test_item); // Make it more used than others
869
870        let test_item_2 = PredefinedBucketItem(4 << 32);
871        filter.insert(&test_item_2); // Sharing the same bucket as "test", but less used
872
873        let test_item_3 = PredefinedBucketItem((3 << 32) + 2);
874        filter.insert(&test_item_3); // Another bucket, but also valid for "test" bucket
875        filter.contains(&test_item_3); // Make it more used
876
877        let test_item_4 = PredefinedBucketItem((5 << 32) + 2);
878        filter.insert(&test_item_4); // Takes bucket of "test_item_3", but less used
879
880        // Everything fits now
881        assert!(filter.contains(&test_item));
882        assert!(filter.contains(&test_item_2));
883        assert!(filter.contains(&test_item_3));
884        assert!(filter.contains(&test_item_4));
885
886        let test_item_5 = PredefinedBucketItem((1 << 32) + 2);
887        // Insert a new item which has to take one of the 2 fully occupied buckets
888        filter.insert(&test_item_5);
889
890        assert!(filter.contains(&test_item_2));
891        assert!(filter.contains(&test_item));
892        assert!(filter.contains(&test_item_3));
893
894        assert!(
895            !filter.contains(&test_item_5) || !filter.contains(&test_item_4),
896            "No inserted items are missing, but filter can't hold them all"
897        );
898
899        // Insert both of these items again and confirm the more used ones are still there
900        filter.insert(&test_item_5);
901        filter.insert(&test_item_4);
902        assert!(filter.contains(&test_item));
903        assert!(filter.contains(&test_item_3));
904    }
905
906    #[test]
907    fn alt_index() {
908        let words = get_words(0..200_000);
909        let filter = CuckooFilter::new_random(
910            CuckooConfiguration::builder(200_000)
911                .fingerprint_bits(32.try_into().unwrap())
912                .build()
913                .unwrap(),
914        );
915
916        for word in words {
917            let (fp, index) = filter.get_fingerprint_and_index(&word);
918            let alt_index = filter.alt_index(&fp, index);
919            assert_eq!(index, filter.alt_index(&fp, alt_index));
920        }
921    }
922
923    #[test]
924    fn random_kicks() {
925        let filter = CuckooFilter::new(
926            CuckooConfiguration::builder(1000)
927                .bucket_size(2.try_into().unwrap())
928                .build()
929                .unwrap(),
930            TestHasher(0),
931        );
932
933        let test_item = PredefinedBucketItem(2 << 32);
934        filter.insert(&test_item);
935
936        let test_item_2 = PredefinedBucketItem(4 << 32);
937        filter.insert(&test_item_2); // Sharing the same bucket as "test"
938
939        let test_item_3 = PredefinedBucketItem((3 << 32) + 2);
940        filter.insert(&test_item_3); // Another bucket, but also valid for "test" bucket
941
942        let test_item_4 = PredefinedBucketItem((5 << 32) + 2);
943        filter.insert(&test_item_4); // Takes bucket of "test_item_3"
944
945        // This one should not be kicked, because it takes an unrelated bucket
946        let test_item_unrelated = PredefinedBucketItem((10 << 32) + 10);
947        filter.insert(&test_item_unrelated);
948
949        // Everything fits now
950        assert!(filter.contains(&test_item));
951        assert!(filter.contains(&test_item_2));
952        assert!(filter.contains(&test_item_3));
953        assert!(filter.contains(&test_item_4));
954
955        let test_item_5 = PredefinedBucketItem((1 << 32) + 2);
956        // Insert a new item which has to take one of the 2 fully occupied buckets
957        let kicked = filter.insert(&test_item_5);
958        assert!(kicked.is_some(), "An item had to be kicked");
959        assert!(filter.contains(&test_item_5));
960        assert!(filter.contains(&test_item_unrelated));
961
962        for item in [&test_item, &test_item_2, &test_item_3, &test_item_4]
963            .iter()
964            .filter(|i| !kicked.as_ref().unwrap().matches_key(i, &filter))
965        {
966            assert!(filter.contains(item), "Only one item should be kicked");
967        }
968    }
969
970    #[test]
971    fn overriding_defaults() {
972        let filter = CuckooFilter::new_random(
973            CuckooConfiguration::builder(1000)
974                .with_ttl(TtlConfig {
975                    ttl: 30.try_into().unwrap(),
976                    ttl_bits: 8.try_into().unwrap(),
977                })
978                .with_counter(CounterConfig::default())
979                .build()
980                .unwrap(),
981        );
982
983        filter.insert_with_defaults(
984            "basic",
985            InsertValues {
986                ttl: Some(50),
987                counter: Some(10),
988            },
989        );
990
991        assert!(filter.contains("basic"));
992        assert_eq!(
993            filter
994                .get_associated_data("basic")
995                .unwrap()
996                .get_stored_ttl_value()
997                .unwrap(),
998            50
999        );
1000        assert_eq!(
1001            filter
1002                .get_associated_data("basic")
1003                .unwrap()
1004                .get_counter()
1005                .unwrap(),
1006            13 // initial 10 + 1 on contains + 2x1 on get_associated_data
1007        );
1008    }
1009
1010    #[test]
1011    fn overriding_updates() {
1012        let filter = CuckooFilter::new_random(
1013            CuckooConfiguration::builder(1000)
1014                .with_ttl(TtlConfig {
1015                    ttl: 30.try_into().unwrap(),
1016                    ttl_bits: 8.try_into().unwrap(),
1017                })
1018                .with_counter(CounterConfig::default())
1019                .build()
1020                .unwrap(),
1021        );
1022
1023        filter.insert_with_defaults(
1024            "basic",
1025            InsertValues {
1026                ttl: Some(5),
1027                counter: Some(1),
1028            },
1029        );
1030
1031        assert!(filter.contains_with_update(
1032            "basic",
1033            LookupValues {
1034                ttl: Some(50),
1035                counter_diff: Some(10),
1036            },
1037        ));
1038        assert_eq!(
1039            filter
1040                .get_associated_data("basic")
1041                .unwrap()
1042                .get_stored_ttl_value()
1043                .unwrap(),
1044            50
1045        );
1046        assert_eq!(
1047            filter
1048                .get_associated_data("basic")
1049                .unwrap()
1050                .get_counter()
1051                .unwrap(),
1052            13 // initial 1 + 10 on contains + 2x1 on get_associated_data
1053        );
1054    }
1055
1056    #[test]
1057    fn scan_and_update_full() {
1058        let words = get_words(0..100_000);
1059        let filter = CuckooFilter::new_random(
1060            CuckooConfiguration::builder(100_000)
1061                .fingerprint_bits(32.try_into().unwrap())
1062                .with_lru(LruConfig::default())
1063                .with_ttl(TtlConfig {
1064                    ttl: 3.try_into().unwrap(),
1065                    ttl_bits: 2.try_into().unwrap(),
1066                })
1067                .build()
1068                .unwrap(),
1069        );
1070
1071        assert_eq!(filter.get_item_count(), 0);
1072
1073        let mut stored_words = HashSet::new();
1074
1075        for (index, word) in words.iter().enumerate() {
1076            stored_words.insert(word);
1077            if let Some(evicted_fp) = filter.insert(word) {
1078                words[0..=index]
1079                    .iter()
1080                    .filter(|w| evicted_fp.matches_key(w, &filter))
1081                    .for_each(|evicted_word| {
1082                        stored_words.remove(evicted_word);
1083                    });
1084            }
1085        }
1086
1087        assert_eq!(filter.get_item_count(), stored_words.len());
1088
1089        for _ in 0..2 {
1090            assert_eq!(filter.scan_and_update_full(), 0);
1091        }
1092
1093        assert_eq!(filter.get_item_count(), stored_words.len());
1094        for word in stored_words.iter() {
1095            assert!(
1096                filter.contains(word),
1097                "Word: {word} expected in the filter, but not found"
1098            );
1099        }
1100
1101        // TTL should remove all entries now
1102        assert_eq!(filter.scan_and_update_full(), stored_words.len());
1103        for word in &words {
1104            assert!(
1105                !filter.contains(word),
1106                "Filter contained {word}, but shouldn't have"
1107            );
1108        }
1109        assert_eq!(filter.get_item_count(), 0);
1110    }
1111
1112    #[test]
1113    fn export_import() {
1114        let words = get_words(0..100_000);
1115        let filter = CuckooFilter::new_random_exportable(
1116            CuckooConfiguration::builder(100_000)
1117                .fingerprint_bits(32.try_into().unwrap())
1118                .with_lru(LruConfig::default())
1119                .with_ttl(TtlConfig {
1120                    ttl: 3.try_into().unwrap(),
1121                    ttl_bits: 2.try_into().unwrap(),
1122                })
1123                .build()
1124                .unwrap(),
1125        );
1126
1127        assert_eq!(filter.get_item_count(), 0);
1128
1129        let mut stored_words = HashSet::new();
1130
1131        for (index, word) in words.iter().enumerate() {
1132            stored_words.insert(word);
1133            if let Some(evicted_fp) = filter.insert(word) {
1134                words[0..=index]
1135                    .iter()
1136                    .filter(|w| evicted_fp.matches_key(w, &filter))
1137                    .for_each(|evicted_word| {
1138                        stored_words.remove(evicted_word);
1139                    });
1140            }
1141        }
1142
1143        assert_eq!(filter.get_item_count(), stored_words.len());
1144
1145        let exported_buf = filter.exporter().snapshot().unwrap();
1146        let mut readable_buf = VecDeque::from(exported_buf);
1147
1148        let imported_filter = CuckooFilter::import_random_exportable(&mut readable_buf).unwrap();
1149
1150        assert_eq!(
1151            imported_filter.get_configuration(),
1152            filter.get_configuration()
1153        );
1154
1155        assert_eq!(imported_filter.get_item_count(), stored_words.len());
1156        for word in stored_words.iter() {
1157            assert!(
1158                imported_filter.contains(word),
1159                "Word: {word} expected in the filter, but not found"
1160            );
1161        }
1162    }
1163}