From cf85f629165ab04db953a7709f8c3d9770df4754 Mon Sep 17 00:00:00 2001 From: "Gregory Meyer (gregjm)" Date: Tue, 10 Mar 2020 02:52:34 -0400 Subject: [PATCH] 0.4.0 (#16) * add run_deferred calls after all test cases additionally, run_deferred now allocates a pretty massive amount of memory in the hopes of filling up the thread and global caches. it seems to work pretty reliably, but it would be better if crossbeam-epoch exposed an API for directly clearing out the thread and global caches. * update copyright year in LICENSE to 2020 * add StripedHashMap (#14) * move all HashMap code of note into BucketArrayRef BucketArrayRef abstracts away the necessary parts of a hash map behind references. This will make implementing the striped table extremely easy! * add striped::map::HashMap, alias as StripedHashMap StripedHashMap partitions a hash map into submaps, where each submap is allocated separately. This is an effort to further spread entries around in memory and reduce contention. * impl Drop for striped::map::HashMap whoops! * add #[inline] to all important table operations all of these just dispatch out to another member function or to a BucketArrayRef function. kind of wasteful to not inline them! * reorder arguments to StripedHashMap constructors to match the names of the functions * correct feature names * memset+Vec::set_len to write a lot of Shared::null this provides a measurable boost to the performance of functions like with_capacity. yes, unsafe, but also yes, really fast. * double default minimum stripe amount empirically provides the best performance * reorganize tests into an icky macro invocation * bump version to 0.4.0 * refer to segments as segments, not stripes in the context of a hash table, striping refers to using a (relatively) small set of locks per bucket. segmenting, the correct term, refers to composing a large hash map out of several smaller hash maps (segments). --- Cargo.toml | 7 +- LICENSE | 2 +- src/lib.rs | 6 + src/map.rs | 338 ++----- src/map/bucket.rs | 17 +- src/map/bucket_array_ref.rs | 356 ++++++++ src/map/tests.rs | 903 ------------------ src/segment.rs | 38 + src/segment/map.rs | 1116 +++++++++++++++++++++++ src/{map/tests/util.rs => test_util.rs} | 11 +- src/test_util/tests.rs | 957 +++++++++++++++++++ 11 files changed, 2577 insertions(+), 1174 deletions(-) create mode 100644 src/map/bucket_array_ref.rs delete mode 100644 src/map/tests.rs create mode 100644 src/segment.rs create mode 100644 src/segment/map.rs rename src/{map/tests/util.rs => test_util.rs} (93%) create mode 100644 src/test_util/tests.rs diff --git a/Cargo.toml b/Cargo.toml index 98263b1..c865d64 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cht" -version = "0.3.0" +version = "0.4.0" authors = ["Gregory Meyer "] edition = "2018" description = "Lockfree resizeable concurrent hash table." @@ -8,9 +8,14 @@ repository = "https://github.com/Gregory-Meyer/cht" readme = "README.md" license = "MIT" +[features] +default = ["num-cpus"] +num-cpus = ["num_cpus"] + [dependencies] ahash = "^0.3.2" crossbeam-epoch = "^0.8.2" +num_cpus = { version = "^1.12.0", optional = true } [dev-dependencies] criterion = "^0.3.1" diff --git a/LICENSE b/LICENSE index fcfaa17..9545bb8 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2019 Gregory Meyer +Copyright (c) 2020 Gregory Meyer Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/src/lib.rs b/src/lib.rs index a8a8925..371b50b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,5 +32,11 @@ //! [Junction]: https://github.com/preshing/junction pub mod map; +pub mod segment; + +#[cfg(test)] +#[macro_use] +pub(crate) mod test_util; pub use map::HashMap; +pub use segment::HashMap as SegmentedHashMap; diff --git a/src/map.rs b/src/map.rs index e0e7118..5e8ba88 100644 --- a/src/map.rs +++ b/src/map.rs @@ -25,12 +25,11 @@ //! A lockfree concurrent hash map implemented with open addressing and linear //! probing. -mod bucket; +pub(crate) mod bucket; +pub(crate) mod bucket_array_ref; -#[cfg(test)] -mod tests; - -use bucket::{Bucket, BucketArray, InsertOrModifyState, KeyOrOwnedBucket}; +use bucket::BucketArray; +use bucket_array_ref::BucketArrayRef; use std::{ borrow::Borrow, @@ -39,7 +38,7 @@ use std::{ }; use ahash::RandomState; -use crossbeam_epoch::{self, Atomic, CompareAndSetError, Guard, Owned, Shared}; +use crossbeam_epoch::{self, Atomic}; /// Default hasher for `HashMap`. /// @@ -84,13 +83,13 @@ pub type DefaultHashBuilder = RandomState; /// [Junction]: https://github.com/preshing/junction /// [this blog post]: https://preshing.com/20160222/a-resizable-concurrent-map/ #[derive(Default)] -pub struct HashMap { +pub struct HashMap { bucket_array: Atomic>, build_hasher: S, len: AtomicUsize, } -impl HashMap { +impl HashMap { /// Creates an empty `HashMap`. /// /// The hash map is created with a capacity of 0 and will not allocate any @@ -108,7 +107,7 @@ impl HashMap { } } -impl HashMap { +impl HashMap { /// Creates an empty `HashMap` that will use `build_hasher` to hash keys. /// /// The created map will have a capacity of 0 and as such will not have any @@ -125,7 +124,10 @@ impl HashMap { let bucket_array = if capacity == 0 { Atomic::null() } else { - Atomic::new(BucketArray::with_capacity(0, capacity)) + Atomic::new(BucketArray::with_length( + 0, + (capacity * 2).next_power_of_two(), + )) }; Self { @@ -169,7 +171,9 @@ impl HashMap { .map(BucketArray::capacity) .unwrap_or(0) } +} +impl HashMap { /// Returns a copy of the value corresponding to `key`. /// /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` @@ -181,6 +185,7 @@ impl HashMap { /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html /// [`get_and`]: #method.get_and + #[inline] pub fn get(&self, key: &Q) -> Option where K: Borrow, @@ -200,6 +205,7 @@ impl HashMap { /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html /// [`get_key_value_and`]: #method.get_key_value_and + #[inline] pub fn get_key_value(&self, key: &Q) -> Option<(K, V)> where K: Borrow + Clone, @@ -218,6 +224,7 @@ impl HashMap { /// /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] pub fn get_and T, T>( &self, key: &Q, @@ -240,6 +247,7 @@ impl HashMap { /// /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] pub fn get_key_value_and T, T>( &self, key: &Q, @@ -248,40 +256,10 @@ impl HashMap { where K: Borrow, { - let guard = &crossbeam_epoch::pin(); - let current_ref = self.bucket_array(guard); - let mut bucket_array_ref = current_ref; - let hash = bucket::hash(&self.build_hasher, key); - - let result; - - loop { - match bucket_array_ref - .get(guard, hash, key) - .map(|p| unsafe { p.as_ref() }) - { - Ok(Some(Bucket { - key, - maybe_value: value, - })) => { - result = Some(with_entry(key, unsafe { &*value.as_ptr() })); - - break; - } - Ok(None) => { - result = None; - - break; - } - Err(_) => { - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - } - } - - self.swing_bucket_array(guard, current_ref, bucket_array_ref); + let hash = bucket::hash(&self.build_hasher, &key); - result + self.bucket_array_ref() + .get_key_value_and(key, hash, with_entry) } /// Inserts a key-value pair, then returns a copy of the value previously @@ -295,6 +273,7 @@ impl HashMap { /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn insert(&self, key: K, value: V) -> Option where V: Clone, @@ -312,6 +291,7 @@ impl HashMap { /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn insert_entry(&self, key: K, value: V) -> Option<(K, V)> where K: Clone, @@ -327,6 +307,7 @@ impl HashMap { /// returned and `with_previous_value` is not invoked. /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] pub fn insert_and T, T>( &self, key: K, @@ -343,57 +324,17 @@ impl HashMap { /// returned and `with_previous_entry` is not invoked. /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] pub fn insert_entry_and T, T>( &self, key: K, value: V, with_previous_entry: F, ) -> Option { - let guard = &crossbeam_epoch::pin(); - let current_ref = self.bucket_array(guard); - let mut bucket_array_ref = current_ref; let hash = bucket::hash(&self.build_hasher, &key); - let mut bucket_ptr = Owned::new(Bucket::new(key, value)); - - let result; - - loop { - while self.len.load(Ordering::Relaxed) > bucket_array_ref.capacity() { - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - - match bucket_array_ref.insert(guard, hash, bucket_ptr) { - Ok(previous_bucket_ptr) => { - if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { - if previous_bucket_ptr.tag() & bucket::TOMBSTONE_TAG != 0 { - self.len.fetch_add(1, Ordering::Relaxed); - result = None; - } else { - let Bucket { - key, - maybe_value: value, - } = previous_bucket_ref; - result = Some(with_previous_entry(key, unsafe { &*value.as_ptr() })); - } - unsafe { bucket::defer_destroy_bucket(guard, previous_bucket_ptr) }; - } else { - self.len.fetch_add(1, Ordering::Relaxed); - result = None; - } - - break; - } - Err(p) => { - bucket_ptr = p; - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - } - } - - self.swing_bucket_array(guard, current_ref, bucket_array_ref); - - result + self.bucket_array_ref() + .insert_entry_and(key, hash, value, with_previous_entry) } /// If there is a value associated with `key`, remove and return a copy of @@ -406,6 +347,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn remove(&self, key: &Q) -> Option where K: Borrow, @@ -424,6 +366,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn remove_entry(&self, key: &Q) -> Option<(K, V)> where K: Borrow + Clone, @@ -440,6 +383,7 @@ impl HashMap { /// /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] pub fn remove_and T, T>( &self, key: &Q, @@ -459,6 +403,7 @@ impl HashMap { /// /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] pub fn remove_entry_and T, T>( &self, key: &Q, @@ -484,6 +429,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn remove_if bool>( &self, key: &Q, @@ -509,6 +455,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn remove_entry_if bool>( &self, key: &Q, @@ -535,6 +482,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] pub fn remove_if_and bool, G: FnOnce(&V) -> T, T>( &self, key: &Q, @@ -561,6 +509,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] pub fn remove_entry_if_and< Q: Hash + Eq + ?Sized, F: FnMut(&K, &V) -> bool, @@ -569,47 +518,16 @@ impl HashMap { >( &self, key: &Q, - mut condition: F, + condition: F, with_previous_entry: G, ) -> Option where K: Borrow, { - let guard = &crossbeam_epoch::pin(); - let current_ref = self.bucket_array(guard); - let mut bucket_array_ref = current_ref; let hash = bucket::hash(&self.build_hasher, &key); - let result; - - loop { - match bucket_array_ref.remove_if(guard, hash, key, condition) { - Ok(previous_bucket_ptr) => { - if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { - let Bucket { - key, - maybe_value: value, - } = previous_bucket_ref; - self.len.fetch_sub(1, Ordering::Relaxed); - result = Some(with_previous_entry(key, unsafe { &*value.as_ptr() })); - - unsafe { bucket::defer_destroy_tombstone(guard, previous_bucket_ptr) }; - } else { - result = None; - } - - break; - } - Err(c) => { - condition = c; - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - } - } - - self.swing_bucket_array(guard, current_ref, bucket_array_ref); - - result + self.bucket_array_ref() + .remove_entry_if_and(key, hash, condition, with_previous_entry) } /// Insert a value if none is associated with `key`. Otherwise, replace the @@ -624,6 +542,7 @@ impl HashMap { /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn insert_or_modify V>( &self, key: K, @@ -648,6 +567,7 @@ impl HashMap { /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn insert_or_modify_entry V>( &self, key: K, @@ -682,6 +602,7 @@ impl HashMap { /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn insert_with_or_modify V, G: FnMut(&K, &V) -> V>( &self, key: K, @@ -710,6 +631,7 @@ impl HashMap { /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn insert_with_or_modify_entry V, G: FnMut(&K, &V) -> V>( &self, key: K, @@ -735,6 +657,7 @@ impl HashMap { /// multiple times, even if [`None`] is returned. /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] pub fn insert_or_modify_and V, G: FnOnce(&V) -> T, T>( &self, key: K, @@ -760,6 +683,7 @@ impl HashMap { /// multiple times, even if [`None`] is returned. /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] pub fn insert_or_modify_entry_and V, G: FnOnce(&K, &V) -> T, T>( &self, key: K, @@ -782,6 +706,7 @@ impl HashMap { /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some + #[inline] pub fn insert_with_or_modify_and< F: FnOnce() -> V, G: FnMut(&K, &V) -> V, @@ -811,6 +736,7 @@ impl HashMap { /// /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some + #[inline] pub fn insert_with_or_modify_entry_and< F: FnOnce() -> V, G: FnMut(&K, &V) -> V, @@ -820,55 +746,18 @@ impl HashMap { &self, key: K, on_insert: F, - mut on_modify: G, + on_modify: G, with_old_entry: H, ) -> Option { - let guard = &crossbeam_epoch::pin(); - let current_ref = self.bucket_array(guard); - let mut bucket_array_ref = current_ref; let hash = bucket::hash(&self.build_hasher, &key); - let mut state = InsertOrModifyState::New(key, on_insert); - - let result; - - loop { - while self.len.load(Ordering::Relaxed) > bucket_array_ref.capacity() { - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - - match bucket_array_ref.insert_or_modify(guard, hash, state, on_modify) { - Ok(previous_bucket_ptr) => { - if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { - if previous_bucket_ptr.tag() & bucket::TOMBSTONE_TAG != 0 { - self.len.fetch_add(1, Ordering::Relaxed); - result = None; - } else { - let Bucket { - key, - maybe_value: value, - } = previous_bucket_ref; - result = Some(with_old_entry(key, unsafe { &*value.as_ptr() })); - } - unsafe { bucket::defer_destroy_bucket(guard, previous_bucket_ptr) }; - } else { - self.len.fetch_add(1, Ordering::Relaxed); - result = None; - } - - break; - } - Err((s, f)) => { - state = s; - on_modify = f; - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - } - } - - self.swing_bucket_array(guard, current_ref, bucket_array_ref); - - result + self.bucket_array_ref().insert_with_or_modify_entry_and( + key, + hash, + on_insert, + on_modify, + with_old_entry, + ) } /// If there is a value associated with `key`, replace it with the result of @@ -886,6 +775,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn modify V>(&self, key: K, on_modify: F) -> Option where V: Clone, @@ -908,6 +798,7 @@ impl HashMap { /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] pub fn modify_entry V>(&self, key: K, on_modify: F) -> Option<(K, V)> where K: Clone, @@ -931,6 +822,7 @@ impl HashMap { /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] pub fn modify_and V, G: FnOnce(&V) -> T, T>( &self, key: K, @@ -954,115 +846,32 @@ impl HashMap { /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] pub fn modify_entry_and V, G: FnOnce(&K, &V) -> T, T>( &self, key: K, - mut on_modify: F, + on_modify: F, with_old_entry: G, ) -> Option { - let guard = &crossbeam_epoch::pin(); - let current_ref = self.bucket_array(guard); - let mut bucket_array_ref = current_ref; let hash = bucket::hash(&self.build_hasher, &key); - let mut key_or_owned_bucket = KeyOrOwnedBucket::Key(key); - let result; - - loop { - match bucket_array_ref.modify(guard, hash, key_or_owned_bucket, on_modify) { - Ok(previous_bucket_ptr) => { - if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { - let Bucket { - key, - maybe_value: value, - } = previous_bucket_ref; - result = Some(with_old_entry(key, unsafe { &*value.as_ptr() })); - - unsafe { bucket::defer_destroy_bucket(guard, previous_bucket_ptr) }; - } else { - result = None; - } - - break; - } - Err((kb, f)) => { - key_or_owned_bucket = kb; - on_modify = f; - bucket_array_ref = bucket_array_ref.rehash(guard, &self.build_hasher); - } - } - } - - self.swing_bucket_array(guard, current_ref, bucket_array_ref); - - result + self.bucket_array_ref() + .modify_entry_and(key, hash, on_modify, with_old_entry) } } -impl HashMap { - fn bucket_array<'g>(&self, guard: &'g Guard) -> &'g BucketArray { - const DEFAULT_CAPACITY: usize = 64; - - let mut maybe_new_bucket_array = None; - - loop { - let bucket_array_ptr = self.bucket_array.load_consume(guard); - - if let Some(bucket_array_ref) = unsafe { bucket_array_ptr.as_ref() } { - return bucket_array_ref; - } - - let new_bucket_array = maybe_new_bucket_array - .unwrap_or_else(|| Owned::new(BucketArray::with_capacity(0, DEFAULT_CAPACITY))); - - match self.bucket_array.compare_and_set_weak( - Shared::null(), - new_bucket_array, - (Ordering::Release, Ordering::Relaxed), - guard, - ) { - Ok(b) => return unsafe { b.as_ref() }.unwrap(), - Err(CompareAndSetError { new, .. }) => maybe_new_bucket_array = Some(new), - } - } - } - - fn swing_bucket_array<'g>( - &self, - guard: &'g Guard, - mut current_ref: &'g BucketArray, - min_ref: &'g BucketArray, - ) { - let min_epoch = min_ref.epoch; - - let mut current_ptr = (current_ref as *const BucketArray).into(); - let min_ptr: Shared<'g, _> = (min_ref as *const BucketArray).into(); - - loop { - if current_ref.epoch >= min_epoch { - return; - } - - match self.bucket_array.compare_and_set_weak( - current_ptr, - min_ptr, - (Ordering::Release, Ordering::Relaxed), - guard, - ) { - Ok(_) => unsafe { bucket::defer_acquire_destroy(guard, current_ptr) }, - Err(_) => { - let new_ptr = self.bucket_array.load_consume(guard); - assert!(!new_ptr.is_null()); - - current_ptr = new_ptr; - current_ref = unsafe { new_ptr.as_ref() }.unwrap(); - } - } +impl HashMap { + #[inline] + fn bucket_array_ref(&'_ self) -> BucketArrayRef<'_, K, V, S> { + BucketArrayRef { + bucket_array: &self.bucket_array, + build_hasher: &self.build_hasher, + len: &self.len, } } } -impl Drop for HashMap { +impl Drop for HashMap { fn drop(&mut self) { let guard = unsafe { &crossbeam_epoch::unprotected() }; atomic::fence(Ordering::Acquire); @@ -1092,3 +901,12 @@ impl Drop for HashMap { } } } + +#[cfg(test)] +mod tests { + use crate::write_test_cases_for_me; + + use super::*; + + write_test_cases_for_me!(HashMap); +} diff --git a/src/map/bucket.rs b/src/map/bucket.rs index 8d09d53..a214883 100644 --- a/src/map/bucket.rs +++ b/src/map/bucket.rs @@ -39,12 +39,13 @@ pub(crate) struct BucketArray { } impl BucketArray { - pub(crate) fn with_capacity(epoch: usize, capacity: usize) -> Self { - let real_capacity = (capacity * 2).next_power_of_two(); - let mut buckets = Vec::with_capacity(real_capacity); + pub(crate) fn with_length(epoch: usize, length: usize) -> Self { + assert!(length.is_power_of_two()); + let mut buckets = Vec::with_capacity(length); - for _ in 0..real_capacity { - buckets.push(Atomic::null()); + unsafe { + ptr::write_bytes(buckets.as_mut_ptr(), 0, length); + buckets.set_len(length); } let buckets = buckets.into_boxed_slice(); @@ -473,9 +474,9 @@ impl<'g, K: 'g, V: 'g> BucketArray { } let new_next = maybe_new_next.unwrap_or_else(|| { - Owned::new(BucketArray::with_capacity( + Owned::new(BucketArray::with_length( self.epoch + 1, - self.buckets.len(), + self.buckets.len() * 2, )) }); @@ -709,7 +710,7 @@ mod tests { #[test] fn get_insert_remove() { let build_hasher = RandomState::new(); - let buckets = BucketArray::with_capacity(0, 8); + let buckets = BucketArray::with_length(0, 16); let guard = unsafe { &crossbeam_epoch::unprotected() }; let k1 = "foo"; diff --git a/src/map/bucket_array_ref.rs b/src/map/bucket_array_ref.rs new file mode 100644 index 0000000..ef7ced5 --- /dev/null +++ b/src/map/bucket_array_ref.rs @@ -0,0 +1,356 @@ +// MIT License +// +// Copyright (c) 2020 Gregory Meyer +// +// Permission is hereby granted, free of charge, to any person +// obtaining a copy of this software and associated documentation files +// (the "Software"), to deal in the Software without restriction, +// including without limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of the Software, +// and to permit persons to whom the Software is furnished to do so, +// subject to the following conditions: +// +// The above copyright notice and this permission notice shall be +// included in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS +// BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN +// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +use super::bucket::{self, Bucket, BucketArray, InsertOrModifyState, KeyOrOwnedBucket}; + +use std::{ + borrow::Borrow, + hash::{BuildHasher, Hash}, + sync::atomic::{AtomicUsize, Ordering}, +}; + +use crossbeam_epoch::{Atomic, CompareAndSetError, Guard, Owned, Shared}; + +pub(crate) struct BucketArrayRef<'a, K, V, S> { + pub(crate) bucket_array: &'a Atomic>, + pub(crate) build_hasher: &'a S, + pub(crate) len: &'a AtomicUsize, +} + +impl<'a, K: Hash + Eq, V, S: BuildHasher> BucketArrayRef<'a, K, V, S> { + pub(crate) fn get_key_value_and T, T>( + &self, + key: &Q, + hash: u64, + with_entry: F, + ) -> Option + where + K: Borrow, + { + let guard = &crossbeam_epoch::pin(); + let current_ref = self.get(guard); + let mut bucket_array_ref = current_ref; + + let result; + + loop { + match bucket_array_ref + .get(guard, hash, key) + .map(|p| unsafe { p.as_ref() }) + { + Ok(Some(Bucket { + key, + maybe_value: value, + })) => { + result = Some(with_entry(key, unsafe { &*value.as_ptr() })); + + break; + } + Ok(None) => { + result = None; + + break; + } + Err(_) => { + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + } + } + + self.swing(guard, current_ref, bucket_array_ref); + + result + } + + pub(crate) fn insert_entry_and T, T>( + &self, + key: K, + hash: u64, + value: V, + with_previous_entry: F, + ) -> Option { + let guard = &crossbeam_epoch::pin(); + let current_ref = self.get(guard); + let mut bucket_array_ref = current_ref; + let mut bucket_ptr = Owned::new(Bucket::new(key, value)); + + let result; + + loop { + while self.len.load(Ordering::Relaxed) > bucket_array_ref.capacity() { + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + + match bucket_array_ref.insert(guard, hash, bucket_ptr) { + Ok(previous_bucket_ptr) => { + if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { + if previous_bucket_ptr.tag() & bucket::TOMBSTONE_TAG != 0 { + self.len.fetch_add(1, Ordering::Relaxed); + result = None; + } else { + let Bucket { + key, + maybe_value: value, + } = previous_bucket_ref; + result = Some(with_previous_entry(key, unsafe { &*value.as_ptr() })); + } + + unsafe { bucket::defer_destroy_bucket(guard, previous_bucket_ptr) }; + } else { + self.len.fetch_add(1, Ordering::Relaxed); + result = None; + } + + break; + } + Err(p) => { + bucket_ptr = p; + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + } + } + + self.swing(guard, current_ref, bucket_array_ref); + + result + } + + pub(crate) fn remove_entry_if_and< + Q: Hash + Eq + ?Sized, + F: FnMut(&K, &V) -> bool, + G: FnOnce(&K, &V) -> T, + T, + >( + &self, + key: &Q, + hash: u64, + mut condition: F, + with_previous_entry: G, + ) -> Option + where + K: Borrow, + { + let guard = &crossbeam_epoch::pin(); + let current_ref = self.get(guard); + let mut bucket_array_ref = current_ref; + + let result; + + loop { + match bucket_array_ref.remove_if(guard, hash, key, condition) { + Ok(previous_bucket_ptr) => { + if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { + let Bucket { + key, + maybe_value: value, + } = previous_bucket_ref; + self.len.fetch_sub(1, Ordering::Relaxed); + result = Some(with_previous_entry(key, unsafe { &*value.as_ptr() })); + + unsafe { bucket::defer_destroy_tombstone(guard, previous_bucket_ptr) }; + } else { + result = None; + } + + break; + } + Err(c) => { + condition = c; + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + } + } + + self.swing(guard, current_ref, bucket_array_ref); + + result + } + + pub(crate) fn insert_with_or_modify_entry_and< + F: FnOnce() -> V, + G: FnMut(&K, &V) -> V, + H: FnOnce(&K, &V) -> T, + T, + >( + &self, + key: K, + hash: u64, + on_insert: F, + mut on_modify: G, + with_old_entry: H, + ) -> Option { + let guard = &crossbeam_epoch::pin(); + let current_ref = self.get(guard); + let mut bucket_array_ref = current_ref; + let mut state = InsertOrModifyState::New(key, on_insert); + + let result; + + loop { + while self.len.load(Ordering::Relaxed) > bucket_array_ref.capacity() { + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + + match bucket_array_ref.insert_or_modify(guard, hash, state, on_modify) { + Ok(previous_bucket_ptr) => { + if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { + if previous_bucket_ptr.tag() & bucket::TOMBSTONE_TAG != 0 { + self.len.fetch_add(1, Ordering::Relaxed); + result = None; + } else { + let Bucket { + key, + maybe_value: value, + } = previous_bucket_ref; + result = Some(with_old_entry(key, unsafe { &*value.as_ptr() })); + } + + unsafe { bucket::defer_destroy_bucket(guard, previous_bucket_ptr) }; + } else { + self.len.fetch_add(1, Ordering::Relaxed); + result = None; + } + + break; + } + Err((s, f)) => { + state = s; + on_modify = f; + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + } + } + + self.swing(guard, current_ref, bucket_array_ref); + + result + } + + pub(crate) fn modify_entry_and V, G: FnOnce(&K, &V) -> T, T>( + &self, + key: K, + hash: u64, + mut on_modify: F, + with_old_entry: G, + ) -> Option { + let guard = &crossbeam_epoch::pin(); + let current_ref = self.get(guard); + let mut bucket_array_ref = current_ref; + let mut key_or_owned_bucket = KeyOrOwnedBucket::Key(key); + + let result; + + loop { + match bucket_array_ref.modify(guard, hash, key_or_owned_bucket, on_modify) { + Ok(previous_bucket_ptr) => { + if let Some(previous_bucket_ref) = unsafe { previous_bucket_ptr.as_ref() } { + let Bucket { + key, + maybe_value: value, + } = previous_bucket_ref; + result = Some(with_old_entry(key, unsafe { &*value.as_ptr() })); + + unsafe { bucket::defer_destroy_bucket(guard, previous_bucket_ptr) }; + } else { + result = None; + } + + break; + } + Err((kb, f)) => { + key_or_owned_bucket = kb; + on_modify = f; + bucket_array_ref = bucket_array_ref.rehash(guard, self.build_hasher); + } + } + } + + self.swing(guard, current_ref, bucket_array_ref); + + result + } +} + +impl<'a, 'g, K, V, S> BucketArrayRef<'a, K, V, S> { + fn get(&self, guard: &'g Guard) -> &'g BucketArray { + const DEFAULT_LENGTH: usize = 128; + + let mut maybe_new_bucket_array = None; + + loop { + let bucket_array_ptr = self.bucket_array.load_consume(guard); + + if let Some(bucket_array_ref) = unsafe { bucket_array_ptr.as_ref() } { + return bucket_array_ref; + } + + let new_bucket_array = maybe_new_bucket_array + .unwrap_or_else(|| Owned::new(BucketArray::with_length(0, DEFAULT_LENGTH))); + + match self.bucket_array.compare_and_set_weak( + Shared::null(), + new_bucket_array, + (Ordering::Release, Ordering::Relaxed), + guard, + ) { + Ok(b) => return unsafe { b.as_ref() }.unwrap(), + Err(CompareAndSetError { new, .. }) => maybe_new_bucket_array = Some(new), + } + } + } + + fn swing( + &self, + guard: &'g Guard, + mut current_ref: &'g BucketArray, + min_ref: &'g BucketArray, + ) { + let min_epoch = min_ref.epoch; + + let mut current_ptr = (current_ref as *const BucketArray).into(); + let min_ptr: Shared<'g, _> = (min_ref as *const BucketArray).into(); + + loop { + if current_ref.epoch >= min_epoch { + return; + } + + match self.bucket_array.compare_and_set_weak( + current_ptr, + min_ptr, + (Ordering::Release, Ordering::Relaxed), + guard, + ) { + Ok(_) => unsafe { bucket::defer_acquire_destroy(guard, current_ptr) }, + Err(_) => { + let new_ptr = self.bucket_array.load_consume(guard); + assert!(!new_ptr.is_null()); + + current_ptr = new_ptr; + current_ref = unsafe { new_ptr.as_ref() }.unwrap(); + } + } + } + } +} diff --git a/src/map/tests.rs b/src/map/tests.rs deleted file mode 100644 index bec0ae9..0000000 --- a/src/map/tests.rs +++ /dev/null @@ -1,903 +0,0 @@ -// MIT License -// -// Copyright (c) 2020 Gregory Meyer -// -// Permission is hereby granted, free of charge, to any person -// obtaining a copy of this software and associated documentation files -// (the "Software"), to deal in the Software without restriction, -// including without limitation the rights to use, copy, modify, merge, -// publish, distribute, sublicense, and/or sell copies of the Software, -// and to permit persons to whom the Software is furnished to do so, -// subject to the following conditions: -// -// The above copyright notice and this permission notice shall be -// included in all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, -// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF -// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND -// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS -// BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN -// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN -// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -// SOFTWARE. - -mod util; - -use util::{DropNotifier, NoisyDropper}; - -use super::*; - -use std::{ - iter, - sync::{Arc, Barrier}, - thread::{self, JoinHandle}, -}; - -#[test] -fn insertion() { - const MAX_VALUE: i32 = 512; - - let map = HashMap::with_capacity(MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.insert(i, i), None); - - assert!(!map.is_empty()); - assert_eq!(map.len(), (i + 1) as usize); - - for j in 0..=i { - assert_eq!(map.get(&j), Some(j)); - assert_eq!(map.insert(j, j), Some(j)); - } - - for k in i + 1..MAX_VALUE { - assert_eq!(map.get(&k), None); - } - } -} - -#[test] -fn growth() { - const MAX_VALUE: i32 = 512; - - let map = HashMap::new(); - - for i in 0..MAX_VALUE { - assert_eq!(map.insert(i, i), None); - - assert!(!map.is_empty()); - assert_eq!(map.len(), (i + 1) as usize); - - for j in 0..=i { - assert_eq!(map.get(&j), Some(j)); - assert_eq!(map.insert(j, j), Some(j)); - } - - for k in i + 1..MAX_VALUE { - assert_eq!(map.get(&k), None); - } - } -} - -#[test] -fn concurrent_insertion() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; - - let map = Arc::new(HashMap::with_capacity(MAX_INSERTED_VALUE as usize)); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.insert(j, j), None); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), MAX_INSERTED_VALUE as usize); - - for i in 0..MAX_INSERTED_VALUE { - assert_eq!(map.get(&i), Some(i)); - } -} - -#[test] -fn concurrent_growth() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; - - let map = Arc::new(HashMap::new()); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.insert(j, j), None); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(|t| t.join()) { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), MAX_INSERTED_VALUE as usize); - - for i in 0..MAX_INSERTED_VALUE { - assert_eq!(map.get(&i), Some(i)); - } -} - -#[test] -fn removal() { - const MAX_VALUE: i32 = 512; - - let map = HashMap::with_capacity(MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.insert(i, i), None); - } - - for i in 0..MAX_VALUE { - assert_eq!(map.remove(&i), Some(i)); - } - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - for i in 0..MAX_VALUE { - assert_eq!(map.get(&i), None); - } -} - -#[test] -fn concurrent_removal() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; - - let map = HashMap::with_capacity(MAX_INSERTED_VALUE as usize); - - for i in 0..MAX_INSERTED_VALUE { - assert_eq!(map.insert(i, i), None); - } - - let map = Arc::new(map); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.remove(&j), Some(j)); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(|t| t.join()) { - assert!(result.is_ok()); - } - - assert_eq!(map.len(), 0); - - for i in 0..MAX_INSERTED_VALUE { - assert_eq!(map.get(&i), None); - } -} - -#[test] -fn concurrent_insertion_and_removal() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE * 2; - const INSERTED_MIDPOINT: i32 = MAX_INSERTED_VALUE / 2; - - let map = HashMap::with_capacity(MAX_INSERTED_VALUE as usize); - - for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { - assert_eq!(map.insert(i, i), None); - } - - let map = Arc::new(map); - let barrier = Arc::new(Barrier::new(NUM_THREADS * 2)); - - let insert_threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.insert(j, j), None); - } - }) - }) - .collect(); - - let remove_threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| INSERTED_MIDPOINT + j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.remove(&j), Some(j)); - } - }) - }) - .collect(); - - for result in insert_threads - .into_iter() - .chain(remove_threads.into_iter()) - .map(|t| t.join()) - { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), INSERTED_MIDPOINT as usize); - - for i in 0..INSERTED_MIDPOINT { - assert_eq!(map.get(&i), Some(i)); - } - - for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { - assert_eq!(map.get(&i), None); - } -} - -#[test] -fn concurrent_growth_and_removal() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE * 2; - const INSERTED_MIDPOINT: i32 = MAX_INSERTED_VALUE / 2; - - let map = HashMap::with_capacity(INSERTED_MIDPOINT as usize); - - for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { - assert_eq!(map.insert(i, i), None); - } - - let map = Arc::new(map); - let barrier = Arc::new(Barrier::new(NUM_THREADS * 2)); - - let insert_threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.insert(j, j), None); - } - }) - }) - .collect(); - - let remove_threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (0..MAX_VALUE).map(|j| INSERTED_MIDPOINT + j + (i as i32 * MAX_VALUE)) { - assert_eq!(map.remove(&j), Some(j)); - } - }) - }) - .collect(); - - for result in insert_threads - .into_iter() - .chain(remove_threads.into_iter()) - .map(JoinHandle::join) - { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), INSERTED_MIDPOINT as usize); - - for i in 0..INSERTED_MIDPOINT { - assert_eq!(map.get(&i), Some(i)); - } - - for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { - assert_eq!(map.get(&i), None); - } -} - -#[test] -fn modify() { - let map = HashMap::new(); - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - assert_eq!(map.modify("foo", |_, x| x * 2), None); - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - map.insert("foo", 1); - assert_eq!(map.modify("foo", |_, x| x * 2), Some(1)); - - assert!(!map.is_empty()); - assert_eq!(map.len(), 1); - - map.remove("foo"); - assert_eq!(map.modify("foo", |_, x| x * 2), None); - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); -} - -#[test] -fn concurrent_modification() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; - - let map = HashMap::with_capacity(MAX_INSERTED_VALUE as usize); - - for i in 0..MAX_INSERTED_VALUE { - map.insert(i, i); - } - - let map = Arc::new(map); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in (i as i32 * MAX_VALUE)..((i as i32 + 1) * MAX_VALUE) { - assert_eq!(map.modify(j, |_, x| x * 2), Some(j)); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), MAX_INSERTED_VALUE as usize); - - for i in 0..MAX_INSERTED_VALUE { - assert_eq!(map.get(&i), Some(i * 2)); - } -} - -#[test] -fn concurrent_overlapped_modification() { - const MAX_VALUE: i32 = 512; - const NUM_THREADS: usize = 64; - - let map = HashMap::with_capacity(MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.insert(i, 0), None); - } - - let map = Arc::new(map); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|_| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for i in 0..MAX_VALUE { - assert!(map.modify(i, |_, x| x + 1).is_some()); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.get(&i), Some(NUM_THREADS as i32)); - } -} - -#[test] -fn insert_or_modify() { - let map = HashMap::new(); - - assert_eq!(map.insert_or_modify("foo", 1, |_, x| x + 1), None); - assert_eq!(map.get("foo"), Some(1)); - - assert_eq!(map.insert_or_modify("foo", 1, |_, x| x + 1), Some(1)); - assert_eq!(map.get("foo"), Some(2)); -} - -#[test] -fn concurrent_insert_or_modify() { - const NUM_THREADS: usize = 64; - const MAX_VALUE: i32 = 512; - - let map = Arc::new(HashMap::new()); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|_| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in 0..MAX_VALUE { - map.insert_or_modify(j, 1, |_, x| x + 1); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert_eq!(map.len(), MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.get(&i), Some(NUM_THREADS as i32)); - } -} - -#[test] -fn concurrent_overlapped_insertion() { - const NUM_THREADS: usize = 64; - const MAX_VALUE: i32 = 512; - - let map = Arc::new(HashMap::with_capacity(MAX_VALUE as usize)); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|_| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in 0..MAX_VALUE { - map.insert(j, j); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert_eq!(map.len(), MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.get(&i), Some(i)); - } -} - -#[test] -fn concurrent_overlapped_growth() { - const NUM_THREADS: usize = 64; - const MAX_VALUE: i32 = 512; - - let map = Arc::new(HashMap::with_capacity(1)); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|_| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in 0..MAX_VALUE { - map.insert(j, j); - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert_eq!(map.len(), MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - assert_eq!(map.get(&i), Some(i)); - } -} - -#[test] -fn concurrent_overlapped_removal() { - const NUM_THREADS: usize = 64; - const MAX_VALUE: i32 = 512; - - let map = HashMap::with_capacity(MAX_VALUE as usize); - - for i in 0..MAX_VALUE { - map.insert(i, i); - } - - let map = Arc::new(map); - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let threads: Vec<_> = (0..NUM_THREADS) - .map(|_| { - let map = map.clone(); - let barrier = barrier.clone(); - - thread::spawn(move || { - barrier.wait(); - - for j in 0..MAX_VALUE { - let prev_value = map.remove(&j); - - if let Some(v) = prev_value { - assert_eq!(v, j); - } - } - }) - }) - .collect(); - - for result in threads.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - for i in 0..MAX_VALUE { - assert_eq!(map.get(&i), None); - } -} - -#[test] -fn drop_value() { - let key_parent = Arc::new(DropNotifier::new()); - let value_parent = Arc::new(DropNotifier::new()); - - { - let map = HashMap::new(); - - assert_eq!( - map.insert_and( - NoisyDropper::new(key_parent.clone(), 0), - NoisyDropper::new(value_parent.clone(), 0), - |_| () - ), - None - ); - assert!(!map.is_empty()); - assert_eq!(map.len(), 1); - map.get_and(&0, |v| assert_eq!(v, &0)); - - map.remove_and(&0, |v| assert_eq!(v, &0)); - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - assert_eq!(map.get_and(&0, |_| ()), None); - - util::run_deferred(); - - assert!(!key_parent.was_dropped()); - assert!(value_parent.was_dropped()); - } - - util::run_deferred(); - - assert!(key_parent.was_dropped()); - assert!(value_parent.was_dropped()); -} - -#[test] -fn drop_many_values() { - const NUM_VALUES: usize = 1 << 16; - - let key_parents: Vec<_> = iter::repeat_with(|| Arc::new(DropNotifier::new())) - .take(NUM_VALUES) - .collect(); - let value_parents: Vec<_> = iter::repeat_with(|| Arc::new(DropNotifier::new())) - .take(NUM_VALUES) - .collect(); - - { - let map = HashMap::new(); - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - for (i, (this_key_parent, this_value_parent)) in - key_parents.iter().zip(value_parents.iter()).enumerate() - { - assert_eq!( - map.insert_and( - NoisyDropper::new(this_key_parent.clone(), i), - NoisyDropper::new(this_value_parent.clone(), i), - |_| () - ), - None - ); - - assert!(!map.is_empty()); - assert_eq!(map.len(), i + 1); - } - - for i in 0..NUM_VALUES { - assert_eq!( - map.get_key_value_and(&i, |k, v| { - assert_eq!(*k, i); - assert_eq!(*v, i); - }), - Some(()) - ); - } - - for i in 0..NUM_VALUES { - assert_eq!( - map.remove_entry_and(&i, |k, v| { - assert_eq!(*k, i); - assert_eq!(*v, i); - }), - Some(()) - ); - } - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - util::run_deferred(); - - for this_key_parent in key_parents.iter() { - assert!(!this_key_parent.was_dropped()); - } - - for this_value_parent in value_parents.iter() { - assert!(this_value_parent.was_dropped()); - } - - for i in 0..NUM_VALUES { - assert_eq!(map.get_and(&i, |_| ()), None); - } - } - - util::run_deferred(); - - for this_key_parent in key_parents.into_iter() { - assert!(this_key_parent.was_dropped()); - } - - for this_value_parent in value_parents.into_iter() { - assert!(this_value_parent.was_dropped()); - } -} - -#[test] -fn drop_many_values_concurrent() { - const NUM_THREADS: usize = 64; - const NUM_VALUES_PER_THREAD: usize = 512; - const NUM_VALUES: usize = NUM_THREADS * NUM_VALUES_PER_THREAD; - - let key_parents: Arc> = Arc::new( - iter::repeat_with(|| Arc::new(DropNotifier::new())) - .take(NUM_VALUES) - .collect(), - ); - let value_parents: Arc> = Arc::new( - iter::repeat_with(|| Arc::new(DropNotifier::new())) - .take(NUM_VALUES) - .collect(), - ); - - { - let map = Arc::new(HashMap::new()); - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - let barrier = Arc::new(Barrier::new(NUM_THREADS)); - - let handles: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = Arc::clone(&map); - let barrier = Arc::clone(&barrier); - let key_parents = Arc::clone(&key_parents); - let value_parents = Arc::clone(&value_parents); - - thread::spawn(move || { - barrier.wait(); - - let these_key_parents = - &key_parents[i * NUM_VALUES_PER_THREAD..(i + 1) * NUM_VALUES_PER_THREAD]; - let these_value_parents = - &value_parents[i * NUM_VALUES_PER_THREAD..(i + 1) * NUM_VALUES_PER_THREAD]; - - for (j, (this_key_parent, this_value_parent)) in these_key_parents - .iter() - .zip(these_value_parents.iter()) - .enumerate() - { - let key_value = i * NUM_VALUES_PER_THREAD + j; - - assert_eq!( - map.insert_and( - NoisyDropper::new(this_key_parent.clone(), key_value), - NoisyDropper::new(this_value_parent.clone(), key_value), - |_| () - ), - None - ); - } - }) - }) - .collect(); - - for result in handles.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert!(!map.is_empty()); - assert_eq!(map.len(), NUM_VALUES); - - util::run_deferred(); - - for this_key_parent in key_parents.iter() { - assert!(!this_key_parent.was_dropped()); - } - - for this_value_parent in value_parents.iter() { - assert!(!this_value_parent.was_dropped()); - } - - for i in 0..NUM_VALUES { - assert_eq!( - map.get_key_value_and(&i, |k, v| { - assert_eq!(*k, i); - assert_eq!(*v, i); - }), - Some(()) - ); - } - - let handles: Vec<_> = (0..NUM_THREADS) - .map(|i| { - let map = Arc::clone(&map); - let barrier = Arc::clone(&barrier); - - thread::spawn(move || { - barrier.wait(); - - for j in 0..NUM_VALUES_PER_THREAD { - let key_value = i * NUM_VALUES_PER_THREAD + j; - - assert_eq!( - map.remove_entry_and(&key_value, |k, v| { - assert_eq!(*k, key_value); - assert_eq!(*v, key_value); - }), - Some(()) - ); - } - }) - }) - .collect(); - - for result in handles.into_iter().map(JoinHandle::join) { - assert!(result.is_ok()); - } - - assert!(map.is_empty()); - assert_eq!(map.len(), 0); - - util::run_deferred(); - - for this_key_parent in key_parents.iter() { - assert!(!this_key_parent.was_dropped()); - } - - for this_value_parent in value_parents.iter() { - assert!(this_value_parent.was_dropped()); - } - - for i in 0..NUM_VALUES { - assert_eq!(map.get_and(&i, |_| ()), None); - } - } - - util::run_deferred(); - - for this_key_parent in key_parents.iter() { - assert!(this_key_parent.was_dropped()); - } - - for this_value_parent in value_parents.iter() { - assert!(this_value_parent.was_dropped()); - } -} - -#[test] -fn remove_if() { - const NUM_VALUES: usize = 512; - - let is_even = |_: &usize, v: &usize| *v % 2 == 0; - - let map = HashMap::new(); - - for i in 0..NUM_VALUES { - assert_eq!(map.insert(i, i), None); - } - - for i in 0..NUM_VALUES { - if is_even(&i, &i) { - assert_eq!(map.remove_if(&i, is_even), Some(i)); - } else { - assert_eq!(map.remove_if(&i, is_even), None); - } - } - - for i in (0..NUM_VALUES).filter(|i| i % 2 == 0) { - assert_eq!(map.get(&i), None); - } - - for i in (0..NUM_VALUES).filter(|i| i % 2 != 0) { - assert_eq!(map.get(&i), Some(i)); - } -} diff --git a/src/segment.rs b/src/segment.rs new file mode 100644 index 0000000..de96124 --- /dev/null +++ b/src/segment.rs @@ -0,0 +1,38 @@ +// MIT License +// +// Copyright (c) 2020 Gregory Meyer +// +// Permission is hereby granted, free of charge, to any person +// obtaining a copy of this software and associated documentation files +// (the "Software"), to deal in the Software without restriction, +// including without limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of the Software, +// and to permit persons to whom the Software is furnished to do so, +// subject to the following conditions: +// +// The above copyright notice and this permission notice shall be +// included in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS +// BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN +// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +//! Segmented hash tables. +//! +//! Segmented hash tables are divided into a user-defined number of smaller hash +//! maps. The most-significant bits of hashed keys are used to select which +//! segment a key will be inserted to. +//! +//! Compared to the unsegmented hash table in this crate, the segmented hash +//! table has better maximum concurrent write throughput for disjoint sets of +//! keys, but slightly worse read and single-threaded write performance due to +//! the extra layer of indirection introduced by segmenting. + +pub mod map; + +pub use map::HashMap; diff --git a/src/segment/map.rs b/src/segment/map.rs new file mode 100644 index 0000000..a5c80cf --- /dev/null +++ b/src/segment/map.rs @@ -0,0 +1,1116 @@ +// MIT License +// +// Copyright (c) 2020 Gregory Meyer +// +// Permission is hereby granted, free of charge, to any person +// obtaining a copy of this software and associated documentation files +// (the "Software"), to deal in the Software without restriction, +// including without limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of the Software, +// and to permit persons to whom the Software is furnished to do so, +// subject to the following conditions: +// +// The above copyright notice and this permission notice shall be +// included in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS +// BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN +// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +//! Lockfree concurrent segmented hash map implemented with open addressing and +//! linear probing. + +use crate::map::{ + bucket::{self, BucketArray}, + bucket_array_ref::BucketArrayRef, + DefaultHashBuilder, +}; + +use std::{ + borrow::Borrow, + hash::{BuildHasher, Hash}, + ptr, + sync::atomic::{self, AtomicUsize, Ordering}, +}; + +use crossbeam_epoch::Atomic; + +/// Lockfree concurrent segmented hash map implemented with open addressing and +/// linear probing. +/// +/// The default hashing algorithm is [aHash], a hashing algorithm that is +/// accelerated by the [AES-NI] instruction set on x86 proessors. aHash provides +/// some resistance to DoS attacks, but will not provide the same level of +/// resistance as something like [`RandomState`]. +/// +/// The hashing algorithm to be used can be chosen on a per-`HashMap` basis +/// using the [`with_hasher`], [`with_capacity_and_hasher`], and +/// [`with_num_segments_capacity_and_hasher`] methods. +/// +/// This map is segmented using the most-significant bits of hashed keys, +/// meaning that entries are spread across several smaller hash maps (segments). +/// Changing the number of segments in a map after construction is not +/// supported. +/// +/// The minimum number of segments can be specified as a parameter to +/// [`with_num_segments`], [`with_num_segments_and_capacity`], +/// [`with_num_segments_and_hasher`], and +/// [`with_num_segments_capacity_and_hasher`]. By default, the `num-cpus` +/// feature is enabled and [`new`] and [`with_capacity`] will create maps with +/// at least twice as many segments as the system has CPUs. +/// +/// Key types must implement [`Hash`] and [`Eq`]. Any operations that return a +/// key or value require the return types to implement [`Clone`], as elements +/// may be in use by other threads and as such cannot be moved from. +/// +/// [`new`]: #method.new +/// [`with_capacity`]: #method.with_capacity +/// [`with_num_segments`]: #method.with_num_segments +/// [`with_num_segments_and_capacity`]: #method.with_num_segments_and_capacity +/// [`with_num_segments_and_hasher`]: #method.with_num_segments_and_hasher +/// [`with_num_segments_capacity_and_hasher`]: #method.with_num_segments_capacity_and_hasher +/// [aHash]: https://docs.rs/ahash +/// [AES-NI]: https://en.wikipedia.org/wiki/AES_instruction_set +/// [`RandomState`]: https://doc.rust-lang.org/std/collections/hash_map/struct.RandomState.html +/// [`with_hasher`]: #method.with_hasher +/// [`with_capacity_and_hasher`]: #method.with_capacity_and_hasher +/// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html +/// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html +/// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html +#[derive(Default)] +pub struct HashMap { + segments: Box<[Segment]>, + build_hasher: S, + len: AtomicUsize, + segment_shift: u32, +} + +#[cfg(feature = "num-cpus")] +impl HashMap { + /// Creates an empty `HashMap`. + /// + /// The hash map is created with a capacity of 0 and no memory for segments + /// will be allocated until the first insertion to each segment. However, + /// memory will always be allocated to store segment pointers and lengths. + /// + /// The `HashMap` will be created with at least twice as many segments as + /// the system has CPUs. + pub fn new() -> Self { + Self::with_num_segments_capacity_and_hasher( + Self::default_num_segments(), + 0, + DefaultHashBuilder::default(), + ) + } + + /// Creates an empty `HashMap` with space for at least `capacity` elements + /// without reallocating. + /// + /// If `capacity == 0`, no memory for segments will be allocated until the + /// first insertion to each segment. However, memory will always be + /// allocated to store segment pointers and lengths. + /// + /// The `HashMap` will be created with at least twice as many segments as + /// the system has CPUs. + pub fn with_capacity(capacity: usize) -> Self { + Self::with_num_segments_capacity_and_hasher( + Self::default_num_segments(), + capacity, + DefaultHashBuilder::default(), + ) + } + + fn default_num_segments() -> usize { + num_cpus::get() * 2 + } +} + +impl HashMap { + /// Creates an empty `HashMap` with at least `num_segments` segments. + /// + /// The hash map is created with a capacity of 0 and no memory for segments + /// will be allocated until the first insertion to each segment. However, + /// memory will always be allocated to store segment pointers and lengths. + /// + /// # Panics + /// + /// Panics if `num_segments == 0`. + pub fn with_num_segments(num_segments: usize) -> Self { + Self::with_num_segments_capacity_and_hasher(num_segments, 0, DefaultHashBuilder::default()) + } + + /// Creates an empty `HashMap` with at least `num_segments` segments and + /// space for at least `capacity` elements in each segment without + /// reallocating. + /// + /// If `capacity == 0`, no memory for segments will be allocated until the + /// first insertion to each segment. However, memory will always be + /// allocated to store segment pointers and lengths. + /// + /// # Panics + /// + /// Panics if `num_segments == 0`. + pub fn with_num_segments_and_capacity(num_segments: usize, capacity: usize) -> Self { + Self::with_num_segments_capacity_and_hasher( + num_segments, + capacity, + DefaultHashBuilder::default(), + ) + } +} + +impl HashMap { + /// Creates an empty `HashMap` that will use `build_hasher` to hash keys + /// with at least `num_segments` segments. + /// + /// The hash map is created with a capacity of 0 and no memory for segments + /// will be allocated until the first insertion to each segment. However, + /// memory will always be allocated to store segment pointers and lengths. + /// + /// # Panics + /// + /// Panics if `num_segments == 0`. + pub fn with_num_segments_and_hasher(num_segments: usize, build_hasher: S) -> Self { + Self::with_num_segments_capacity_and_hasher(num_segments, 0, build_hasher) + } + + /// Creates an empty `HashMap` that will use `build_hasher` to hash keys, + /// hold at least `capacity` elements without reallocating, and have at + /// least `num_segments` segments. + /// + /// If `capacity == 0`, no memory for segments will be allocated until the + /// first insertion to each segment. However, memory will always be + /// allocated to store segment pointers and lengths. + /// + /// # Panics + /// + /// Panics if `num_segments == 0`. + pub fn with_num_segments_capacity_and_hasher( + num_segments: usize, + capacity: usize, + build_hasher: S, + ) -> Self { + assert!(num_segments > 0); + + let actual_num_segments = num_segments.next_power_of_two(); + let segment_shift = 64 - actual_num_segments.trailing_zeros(); + + let mut segments = Vec::with_capacity(actual_num_segments); + + if capacity == 0 { + unsafe { + ptr::write_bytes(segments.as_mut_ptr(), 0, actual_num_segments); + segments.set_len(actual_num_segments); + } + } else { + let actual_capacity = (capacity * 2).next_power_of_two(); + + for _ in 0..actual_num_segments { + segments.push(Segment { + bucket_array: Atomic::new(BucketArray::with_length(0, actual_capacity)), + len: AtomicUsize::new(0), + }); + } + } + + let segments = segments.into_boxed_slice(); + + Self { + segments, + build_hasher, + len: AtomicUsize::new(0), + segment_shift, + } + } + + /// Returns the number of elements in the map. + /// + /// # Safety + /// + /// This method on its own is safe, but other threads can add or remove + /// elements at any time. + pub fn len(&self) -> usize { + self.len.load(Ordering::Relaxed) + } + + /// Returns `true` if the map contains no elements. + /// + /// # Safety + /// + /// This method on its own is safe, but other threads can add or remove + /// elements at any time. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Returns the number of elements the map can hold without reallocating a + /// bucket pointer array. + /// + /// As the map is composed of multiple separately allocated segments, this + /// method returns the minimum capacity of all segments. + /// + /// Note that all mutating operations, with the exception of removing + /// elements, will result in an allocation for a new bucket. + /// + /// # Safety + /// + /// This method on its own is safe, but other threads can increase the + /// capacity of each segment at any time by adding elements. + pub fn capacity(&self) -> usize { + let guard = &crossbeam_epoch::pin(); + + self.segments + .iter() + .map(|s| s.bucket_array.load_consume(guard)) + .map(|p| unsafe { p.as_ref() }) + .map(|a| a.map(BucketArray::capacity).unwrap_or(0)) + .min() + .unwrap() + } + + /// Returns the number of elements the `index`-th segment of the map can + /// hold without reallocating a bucket pointer array. + /// + /// Note that all mutating operations, with the exception of removing + /// elements, will result in an allocation for a new bucket. + /// + /// # Safety + /// + /// This method on its own is safe, but other threads can increase the + /// capacity of a segment at any time by adding elements. + pub fn segment_capacity(&self, index: usize) -> usize { + assert!(index < self.segments.len()); + + let guard = &crossbeam_epoch::pin(); + + unsafe { + self.segments[index] + .bucket_array + .load_consume(guard) + .as_ref() + } + .map(BucketArray::capacity) + .unwrap_or(0) + } + + /// Returns the number of segments in the map. + pub fn num_segments(&self) -> usize { + self.segments.len() + } +} + +impl HashMap { + /// Returns the index of the segment that `key` would belong to if inserted + /// into the map. + pub fn segment_index(&self, key: &Q) -> usize + where + K: Borrow, + { + let hash = bucket::hash(&self.build_hasher, key); + + self.segment_index_from_hash(hash) + } +} + +impl HashMap { + /// Returns a copy of the value associated with `key`. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `V` must implement [`Clone`], as other threads + /// may hold references to the associated value. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn get(&self, key: &Q) -> Option + where + K: Borrow, + V: Clone, + { + self.get_key_value_and(key, |_, v| v.clone()) + } + + /// Returns a copy of the key and value associated with `key`. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `K` and `V` must implement [`Clone`], as other + /// threads may hold references to the entry. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn get_key_value(&self, key: &Q) -> Option<(K, V)> + where + K: Borrow + Clone, + V: Clone, + { + self.get_key_value_and(key, |k, v| (k.clone(), v.clone())) + } + + /// Invokes `with_value` with a reference to the value associated with `key`. + /// + /// If there is no value associated with `key` in the map, `with_value` will + /// not be invoked and [`None`] will be returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] + pub fn get_and T, T>( + &self, + key: &Q, + with_value: F, + ) -> Option + where + K: Borrow, + { + self.get_key_value_and(key, move |_, v| with_value(v)) + } + + /// Invokes `with_entry` with a reference to the key and value associated + /// with `key`. + /// + /// If there is no value associated with `key` in the map, `with_entry` will + /// not be invoked and [`None`] will be returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] + pub fn get_key_value_and T, T>( + &self, + key: &Q, + with_entry: F, + ) -> Option + where + K: Borrow, + { + let hash = bucket::hash(&self.build_hasher, &key); + + self.bucket_array_ref(hash) + .get_key_value_and(key, hash, with_entry) + } + + /// Inserts a key-value pair, then returns a copy of the value previously + /// associated with `key`. + /// + /// If the key was not previously present in this hash map, [`None`] is + /// returned. + /// + /// `V` must implement [`Clone`], as other threads may hold references to + /// the associated value. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn insert(&self, key: K, value: V) -> Option + where + V: Clone, + { + self.insert_entry_and(key, value, |_, v| v.clone()) + } + + /// Inserts a key-value pair, then returns a copy of the previous entry. + /// + /// If the key was not previously present in this hash map, [`None`] is + /// returned. + /// + /// `K` and `V` must implement [`Clone`], as other threads may hold + /// references to the entry. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn insert_entry(&self, key: K, value: V) -> Option<(K, V)> + where + K: Clone, + V: Clone, + { + self.insert_entry_and(key, value, |k, v| (k.clone(), v.clone())) + } + + /// Inserts a key-value pair, then invokes `with_previous_value` with the + /// value previously associated with `key`. + /// + /// If the key was not previously present in this hash map, [`None`] is + /// returned and `with_previous_value` is not invoked. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] + pub fn insert_and T, T>( + &self, + key: K, + value: V, + with_previous_value: F, + ) -> Option { + self.insert_entry_and(key, value, move |_, v| with_previous_value(v)) + } + + /// Inserts a key-value pair, then invokes `with_previous_entry` with the + /// previous entry. + /// + /// If the key was not previously present in this hash map, [`None`] is + /// returned and `with_previous_entry` is not invoked. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] + pub fn insert_entry_and T, T>( + &self, + key: K, + value: V, + with_previous_entry: F, + ) -> Option { + let hash = bucket::hash(&self.build_hasher, &key); + + let result = + self.bucket_array_ref(hash) + .insert_entry_and(key, hash, value, with_previous_entry); + + if result.is_none() { + self.len.fetch_add(1, Ordering::Relaxed); + } + + result + } + + /// If there is a value associated with `key`, remove and return a copy of + /// it. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `V` must implement [`Clone`], as other + /// threads may hold references to the associated value. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn remove(&self, key: &Q) -> Option + where + K: Borrow, + V: Clone, + { + self.remove_entry_if_and(key, |_, _| true, |_, v| v.clone()) + } + + /// If there is a value associated with `key`, remove it and return a copy + /// of the previous entity. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `K` and `V` must implement [`Clone`], as other + /// threads may hold references to the entry. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn remove_entry(&self, key: &Q) -> Option<(K, V)> + where + K: Borrow + Clone, + V: Clone, + { + self.remove_entry_if_and(key, |_, _| true, |k, v| (k.clone(), v.clone())) + } + + /// If there is a value associated with `key`, remove it and return the + /// result of invoking `with_previous_value` with that value. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] + pub fn remove_and T, T>( + &self, + key: &Q, + with_previous_value: F, + ) -> Option + where + K: Borrow, + { + self.remove_entry_if_and(key, |_, _| true, move |_, v| with_previous_value(v)) + } + + /// If there is a value associated with `key`, remove it and return the + /// result of invoking `with_previous_entry` with that entry. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] + pub fn remove_entry_and T, T>( + &self, + key: &Q, + with_previous_entry: F, + ) -> Option + where + K: Borrow, + { + self.remove_entry_if_and(key, |_, _| true, with_previous_entry) + } + + /// If there is a value associated with `key` and `condition` returns true + /// when invoked with the current entry, remove and return a copy of its + /// value. + /// + /// `condition` may be invoked one or more times, even if no entry was + /// removed. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `K` and `V` must implement [`Clone`], as other + /// threads may hold references to the entry. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn remove_if bool>( + &self, + key: &Q, + condition: F, + ) -> Option + where + K: Borrow, + V: Clone, + { + self.remove_entry_if_and(key, condition, move |_, v| v.clone()) + } + + /// If there is a value associated with `key` and `condition` returns true + /// when invoked with the current entry, remove and return a copy of it. + /// + /// `condition` may be invoked one or more times, even if no entry was + /// removed. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `K` and `V` must implement [`Clone`], as other + /// threads may hold references to the entry. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn remove_entry_if bool>( + &self, + key: &Q, + condition: F, + ) -> Option<(K, V)> + where + K: Clone + Borrow, + V: Clone, + { + self.remove_entry_if_and(key, condition, move |k, v| (k.clone(), v.clone())) + } + + /// If there is a value associated with `key` and `condition` returns true + /// when invoked with the current entry, remove it and return the result of + /// invoking `with_previous_value` with its value. + /// + /// `condition` may be invoked one or more times, even if no entry was + /// removed. If `condition` failed or there was no value associated with + /// `key`, `with_previous_entry` is not invoked and [`None`] is returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] + pub fn remove_if_and bool, G: FnOnce(&V) -> T, T>( + &self, + key: &Q, + condition: F, + with_previous_value: G, + ) -> Option + where + K: Borrow, + { + self.remove_entry_if_and(key, condition, move |_, v| with_previous_value(v)) + } + + /// If there is a value associated with `key` and `condition` returns true + /// when invoked with the current entry, remove it and return the result of + /// invoking `with_previous_entry` with it. + /// + /// `condition` may be invoked one or more times, even if no entry was + /// removed. If `condition` failed or there was no value associated with + /// `key`, `with_previous_entry` is not invoked and [`None`] is returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] + pub fn remove_entry_if_and< + Q: Hash + Eq + ?Sized, + F: FnMut(&K, &V) -> bool, + G: FnOnce(&K, &V) -> T, + T, + >( + &self, + key: &Q, + condition: F, + with_previous_entry: G, + ) -> Option + where + K: Borrow, + { + let hash = bucket::hash(&self.build_hasher, &key); + + self.bucket_array_ref(hash) + .remove_entry_if_and(key, hash, condition, move |k, v| { + self.len.fetch_sub(1, Ordering::Relaxed); + + with_previous_entry(k, v) + }) + } + + /// Insert a value if none is associated with `key`. Otherwise, replace the + /// value with the result of `on_modify` with the current entry as + /// arguments. Finally, return a copy of the previously associated value. + /// + /// If there is no value associated with `key`, [`None`] will be returned. + /// `on_modify` may be invoked multiple times, even if [`None`] is returned. + /// + /// `V` must implement [`Clone`], as other threads may hold references to + /// the associated value. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn insert_or_modify V>( + &self, + key: K, + value: V, + on_modify: F, + ) -> Option + where + V: Clone, + { + self.insert_with_or_modify_entry_and(key, move || value, on_modify, |_, v| v.clone()) + } + + /// Insert a value if none is associated with `key`. Otherwise, replace the + /// value with the result of `on_modify` with the current entry as + /// arguments. Finally, return a copy of the previous entry. + /// + /// If there is no value associated with `key`, [`None`] will be returned. + /// `on_modify` may be invoked multiple times, even if [`None`] is returned. + /// + /// `K` and `V` must implement [`Clone`], as other threads may hold + /// references to the entry. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn insert_or_modify_entry V>( + &self, + key: K, + value: V, + on_modify: F, + ) -> Option<(K, V)> + where + K: Clone, + V: Clone, + { + self.insert_with_or_modify_entry_and( + key, + move || value, + on_modify, + |k, v| (k.clone(), v.clone()), + ) + } + + /// Insert the result of `on_insert` if no value is associated with `key`. + /// Otherwise, replace the value with the result of `on_modify` with the + /// current entry as arguments. Finally, return a copy of the previously + /// associated value. + /// + /// If there is no value associated with `key`, `on_insert` will be invoked + /// and [`None`] will be returned. `on_modify` may be invoked multiple + /// times, even if [`None`] is returned. Similarly, `on_insert` may be + /// invoked if [`Some`] is returned. + /// + /// `V` must implement [`Clone`], as other threads may hold references to + /// the associated value. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn insert_with_or_modify V, G: FnMut(&K, &V) -> V>( + &self, + key: K, + on_insert: F, + on_modify: G, + ) -> Option + where + V: Clone, + { + self.insert_with_or_modify_entry_and(key, on_insert, on_modify, |_, v| v.clone()) + } + + /// Insert the result of `on_insert` if no value is associated with `key`. + /// Otherwise, replace the value with the result of `on_modify` with the + /// current entry as arguments. Finally, return a copy of the previous + /// entry. + /// + /// If there is no value associated with `key`, `on_insert` will be invoked + /// and [`None`] will be returned. `on_modify` may be invoked multiple + /// times, even if [`None`] is returned. Similarly, `on_insert` may be + /// invoked if [`Some`] is returned. + /// + /// `K` and `V` must implement [`Clone`], as other threads may hold + /// references to the entry. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn insert_with_or_modify_entry V, G: FnMut(&K, &V) -> V>( + &self, + key: K, + on_insert: F, + on_modify: G, + ) -> Option<(K, V)> + where + K: Clone, + V: Clone, + { + self.insert_with_or_modify_entry_and(key, on_insert, on_modify, |k, v| { + (k.clone(), v.clone()) + }) + } + + /// Insert a value if none is associated with `key`. Otherwise, replace the + /// value with the result of `on_modify` with the current entry as + /// arguments. Finally, return the result of invoking `with_old_value` with + /// the previously associated value. + /// + /// If there is no value associated with `key`, `with_old_value` will not be + /// invoked and [`None`] will be returned. `on_modify` may be invoked + /// multiple times, even if [`None`] is returned. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] + pub fn insert_or_modify_and V, G: FnOnce(&V) -> T, T>( + &self, + key: K, + value: V, + on_modify: F, + with_old_value: G, + ) -> Option { + self.insert_with_or_modify_entry_and( + key, + move || value, + on_modify, + move |_, v| with_old_value(v), + ) + } + + /// Insert a value if none is associated with `key`. Otherwise, replace the + /// value with the result of `on_modify` with the current entry as + /// arguments. Finally, return the result of invoking `with_old_entry` with + /// the previous entry. + /// + /// If there is no value associated with `key`, `with_old_value` will not be + /// invoked and [`None`] will be returned. `on_modify` may be invoked + /// multiple times, even if [`None`] is returned. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + #[inline] + pub fn insert_or_modify_entry_and V, G: FnOnce(&K, &V) -> T, T>( + &self, + key: K, + value: V, + on_modify: F, + with_old_entry: G, + ) -> Option { + self.insert_with_or_modify_entry_and(key, move || value, on_modify, with_old_entry) + } + + /// Insert the result of `on_insert` if no value is associated with `key`. + /// Otherwise, replace the value with the result of `on_modify` with the + /// current entry as arguments. Finally, return the result of invoking + /// `with_old_value` with the previously associated value. + /// + /// If there is no value associated with `key`, `on_insert` will be invoked, + /// `with_old_value` will not be invoked, and [`None`] will be returned. + /// `on_modify` may be invoked multiple times, even if [`None`] is returned. + /// Similarly, `on_insert` may be invoked if [`Some`] is returned. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some + #[inline] + pub fn insert_with_or_modify_and< + F: FnOnce() -> V, + G: FnMut(&K, &V) -> V, + H: FnOnce(&V) -> T, + T, + >( + &self, + key: K, + on_insert: F, + on_modify: G, + with_old_value: H, + ) -> Option { + self.insert_with_or_modify_entry_and(key, on_insert, on_modify, move |_, v| { + with_old_value(v) + }) + } + + /// Insert the result of `on_insert` if no value is associated with `key`. + /// Otherwise, replace the value with the result of `on_modify` with the + /// current entry as arguments. Finally, return the result of invoking + /// `with_old_entry` with the previous entry. + /// + /// If there is no value associated with `key`, `on_insert` will be invoked, + /// `with_old_value` will not be invoked, and [`None`] will be returned. + /// `on_modify` may be invoked multiple times, even if [`None`] is returned. + /// Similarly, `on_insert` may be invoked if [`Some`] is returned. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Some`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.Some + #[inline] + pub fn insert_with_or_modify_entry_and< + F: FnOnce() -> V, + G: FnMut(&K, &V) -> V, + H: FnOnce(&K, &V) -> T, + T, + >( + &self, + key: K, + on_insert: F, + on_modify: G, + with_old_entry: H, + ) -> Option { + let hash = bucket::hash(&self.build_hasher, &key); + + let result = self.bucket_array_ref(hash).insert_with_or_modify_entry_and( + key, + hash, + on_insert, + on_modify, + with_old_entry, + ); + + if result.is_none() { + self.len.fetch_add(1, Ordering::Relaxed); + } + + result + } + + /// If there is a value associated with `key`, replace it with the result of + /// invoking `on_modify` using the current key and value, then return a copy + /// of the previously associated value. + /// + /// If there is no value associated with `key`, [`None`] will be returned. + /// `on_modify` may be invoked multiple times, even if [`None`] is returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `V` must implement [`Clone`], as other + /// threads may hold references to the associated value. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn modify V>(&self, key: K, on_modify: F) -> Option + where + V: Clone, + { + self.modify_entry_and(key, on_modify, |_, v| v.clone()) + } + + /// If there is a value associated with `key`, replace it with the result of + /// invoking `on_modify` using the current key and value, then return a copy + /// of the previously entry. + /// + /// If there is no value associated with `key`, [`None`] will be returned. + /// `on_modify` may be invoked multiple times, even if [`None`] is returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. `K` and `V` must implement [`Clone`], as other + /// threads may hold references to the entry. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html + #[inline] + pub fn modify_entry V>(&self, key: K, on_modify: F) -> Option<(K, V)> + where + K: Clone, + V: Clone, + { + self.modify_entry_and(key, on_modify, |k, v| (k.clone(), v.clone())) + } + + /// If there is a value associated with `key`, replace it with the result of + /// invoking `on_modify` using the current key and value, then return the + /// result of invoking `with_old_value` with the previously associated + /// value. + /// + /// If there is no value associated with `key`, `with_old_value` will not be + /// invoked and [`None`] will be returned. `on_modify` may be invoked + /// multiple times, even if [`None`] is returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] + pub fn modify_and V, G: FnOnce(&V) -> T, T>( + &self, + key: K, + on_modify: F, + with_old_value: G, + ) -> Option { + self.modify_entry_and(key, on_modify, move |_, v| with_old_value(v)) + } + + /// If there is a value associated with `key`, replace it with the result of + /// invoking `on_modify` using the current key and value, then return the + /// result of invoking `with_old_value` with the previous entry. + /// + /// If there is no value associated with `key`, `with_old_value` will not be + /// invoked and [`None`] will be returned. `on_modify` may be invoked + /// multiple times, even if [`None`] is returned. + /// + /// `Q` can be any borrowed form of `K`, but [`Hash`] and [`Eq`] on `Q` + /// *must* match that of `K`. + /// + /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None + /// [`Hash`]: https://doc.rust-lang.org/std/hash/trait.Hash.html + /// [`Eq`]: https://doc.rust-lang.org/std/cmp/trait.Eq.html + #[inline] + pub fn modify_entry_and V, G: FnOnce(&K, &V) -> T, T>( + &self, + key: K, + on_modify: F, + with_old_entry: G, + ) -> Option { + let hash = bucket::hash(&self.build_hasher, &key); + + self.bucket_array_ref(hash) + .modify_entry_and(key, hash, on_modify, with_old_entry) + } +} + +impl Drop for HashMap { + fn drop(&mut self) { + let guard = unsafe { &crossbeam_epoch::unprotected() }; + atomic::fence(Ordering::Acquire); + + for Segment { + bucket_array: this_bucket_array, + .. + } in self.segments.iter() + { + let mut current_ptr = this_bucket_array.load(Ordering::Relaxed, guard); + + while let Some(current_ref) = unsafe { current_ptr.as_ref() } { + let next_ptr = current_ref.next.load(Ordering::Relaxed, guard); + + for this_bucket_ptr in current_ref + .buckets + .iter() + .map(|b| b.load(Ordering::Relaxed, guard)) + .filter(|p| !p.is_null()) + .filter(|p| next_ptr.is_null() || p.tag() & bucket::TOMBSTONE_TAG == 0) + { + // only delete tombstones from the newest bucket array + // the only way this becomes a memory leak is if there was a panic during a rehash, + // in which case i'm going to say that running destructors and freeing memory is + // best-effort, and my best effort is to not do it + unsafe { bucket::defer_acquire_destroy(guard, this_bucket_ptr) }; + } + + unsafe { bucket::defer_acquire_destroy(guard, current_ptr) }; + + current_ptr = next_ptr; + } + } + } +} + +impl HashMap { + #[inline] + fn bucket_array_ref(&'_ self, hash: u64) -> BucketArrayRef<'_, K, V, S> { + let index = self.segment_index_from_hash(hash); + + let Segment { + ref bucket_array, + ref len, + } = self.segments[index]; + + BucketArrayRef { + bucket_array, + build_hasher: &self.build_hasher, + len, + } + } + + #[inline] + fn segment_index_from_hash(&'_ self, hash: u64) -> usize { + if self.segment_shift == 64 { + 0 + } else { + (hash >> self.segment_shift) as usize + } + } +} + +struct Segment { + bucket_array: Atomic>, + len: AtomicUsize, +} + +#[cfg(test)] +mod tests { + use crate::write_test_cases_for_me; + + use super::*; + + write_test_cases_for_me!(HashMap); + + #[test] + fn single_segment() { + let map = HashMap::with_num_segments(1); + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + assert_eq!(map.insert("foo", 5), None); + assert_eq!(map.get("foo"), Some(5)); + + assert!(!map.is_empty()); + assert_eq!(map.len(), 1); + + assert_eq!(map.remove("foo"), Some(5)); + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + } +} diff --git a/src/map/tests/util.rs b/src/test_util.rs similarity index 93% rename from src/map/tests/util.rs rename to src/test_util.rs index ad5af00..f4163a6 100644 --- a/src/map/tests/util.rs +++ b/src/test_util.rs @@ -22,6 +22,9 @@ // CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE // SOFTWARE. +#[macro_use] +pub(crate) mod tests; + use std::{ borrow::{Borrow, BorrowMut}, hash::{Hash, Hasher}, @@ -32,6 +35,8 @@ use std::{ }, }; +use crossbeam_epoch::Owned; + #[derive(Debug)] pub(crate) struct NoisyDropper { parent: Arc, @@ -127,6 +132,10 @@ impl DropNotifier { pub(crate) fn run_deferred() { for _ in 0..65536 { - crossbeam_epoch::pin().flush(); + let guard = crossbeam_epoch::pin(); + + unsafe { guard.defer_destroy(Owned::new(0).into_shared(&guard)) }; + + guard.flush(); } } diff --git a/src/test_util/tests.rs b/src/test_util/tests.rs new file mode 100644 index 0000000..04840d8 --- /dev/null +++ b/src/test_util/tests.rs @@ -0,0 +1,957 @@ +// MIT License +// +// Copyright (c) 2020 Gregory Meyer +// +// Permission is hereby granted, free of charge, to any person +// obtaining a copy of this software and associated documentation files +// (the "Software"), to deal in the Software without restriction, +// including without limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of the Software, +// and to permit persons to whom the Software is furnished to do so, +// subject to the following conditions: +// +// The above copyright notice and this permission notice shall be +// included in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS +// BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN +// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +#[macro_export] +macro_rules! write_test_cases_for_me { + ($m:ident) => { + #[test] + fn insertion() { + const MAX_VALUE: i32 = 512; + + let map = $m::with_capacity(MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.insert(i, i), None); + + assert!(!map.is_empty()); + assert_eq!(map.len(), (i + 1) as usize); + + for j in 0..=i { + assert_eq!(map.get(&j), Some(j)); + assert_eq!(map.insert(j, j), Some(j)); + } + + for k in i + 1..MAX_VALUE { + assert_eq!(map.get(&k), None); + } + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn growth() { + const MAX_VALUE: i32 = 512; + + let map = $m::new(); + + for i in 0..MAX_VALUE { + assert_eq!(map.insert(i, i), None); + + assert!(!map.is_empty()); + assert_eq!(map.len(), (i + 1) as usize); + + for j in 0..=i { + assert_eq!(map.get(&j), Some(j)); + assert_eq!(map.insert(j, j), Some(j)); + } + + for k in i + 1..MAX_VALUE { + assert_eq!(map.get(&k), None); + } + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_insertion() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; + + let map = std::sync::Arc::new($m::with_capacity(MAX_INSERTED_VALUE as usize)); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { + assert_eq!(map.insert(j, j), None); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), MAX_INSERTED_VALUE as usize); + + for i in 0..MAX_INSERTED_VALUE { + assert_eq!(map.get(&i), Some(i)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_growth() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; + + let map = std::sync::Arc::new($m::new()); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { + assert_eq!(map.insert(j, j), None); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(|t| t.join()) { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), MAX_INSERTED_VALUE as usize); + + for i in 0..MAX_INSERTED_VALUE { + assert_eq!(map.get(&i), Some(i)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn removal() { + const MAX_VALUE: i32 = 512; + + let map = $m::with_capacity(MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.insert(i, i), None); + } + + for i in 0..MAX_VALUE { + assert_eq!(map.remove(&i), Some(i)); + } + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + for i in 0..MAX_VALUE { + assert_eq!(map.get(&i), None); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_removal() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; + + let map = $m::with_capacity(MAX_INSERTED_VALUE as usize); + + for i in 0..MAX_INSERTED_VALUE { + assert_eq!(map.insert(i, i), None); + } + + let map = std::sync::Arc::new(map); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { + assert_eq!(map.remove(&j), Some(j)); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(|t| t.join()) { + assert!(result.is_ok()); + } + + assert_eq!(map.len(), 0); + + for i in 0..MAX_INSERTED_VALUE { + assert_eq!(map.get(&i), None); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_insertion_and_removal() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE * 2; + const INSERTED_MIDPOINT: i32 = MAX_INSERTED_VALUE / 2; + + let map = $m::with_capacity(MAX_INSERTED_VALUE as usize); + + for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { + assert_eq!(map.insert(i, i), None); + } + + let map = std::sync::Arc::new(map); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS * 2)); + + let insert_threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { + assert_eq!(map.insert(j, j), None); + } + }) + }) + .collect(); + + let remove_threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in + (0..MAX_VALUE).map(|j| INSERTED_MIDPOINT + j + (i as i32 * MAX_VALUE)) + { + assert_eq!(map.remove(&j), Some(j)); + } + }) + }) + .collect(); + + for result in insert_threads + .into_iter() + .chain(remove_threads.into_iter()) + .map(|t| t.join()) + { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), INSERTED_MIDPOINT as usize); + + for i in 0..INSERTED_MIDPOINT { + assert_eq!(map.get(&i), Some(i)); + } + + for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { + assert_eq!(map.get(&i), None); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_growth_and_removal() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE * 2; + const INSERTED_MIDPOINT: i32 = MAX_INSERTED_VALUE / 2; + + let map = $m::with_capacity(INSERTED_MIDPOINT as usize); + + for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { + assert_eq!(map.insert(i, i), None); + } + + let map = std::sync::Arc::new(map); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS * 2)); + + let insert_threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in (0..MAX_VALUE).map(|j| j + (i as i32 * MAX_VALUE)) { + assert_eq!(map.insert(j, j), None); + } + }) + }) + .collect(); + + let remove_threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in + (0..MAX_VALUE).map(|j| INSERTED_MIDPOINT + j + (i as i32 * MAX_VALUE)) + { + assert_eq!(map.remove(&j), Some(j)); + } + }) + }) + .collect(); + + for result in insert_threads + .into_iter() + .chain(remove_threads.into_iter()) + .map(std::thread::JoinHandle::join) + { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), INSERTED_MIDPOINT as usize); + + for i in 0..INSERTED_MIDPOINT { + assert_eq!(map.get(&i), Some(i)); + } + + for i in INSERTED_MIDPOINT..MAX_INSERTED_VALUE { + assert_eq!(map.get(&i), None); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn modify() { + let map = $m::new(); + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + assert_eq!(map.modify("foo", |_, x| x * 2), None); + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + map.insert("foo", 1); + assert_eq!(map.modify("foo", |_, x| x * 2), Some(1)); + + assert!(!map.is_empty()); + assert_eq!(map.len(), 1); + + map.remove("foo"); + assert_eq!(map.modify("foo", |_, x| x * 2), None); + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_modification() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + const MAX_INSERTED_VALUE: i32 = (NUM_THREADS as i32) * MAX_VALUE; + + let map = $m::with_capacity(MAX_INSERTED_VALUE as usize); + + for i in 0..MAX_INSERTED_VALUE { + map.insert(i, i); + } + + let map = std::sync::Arc::new(map); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in (i as i32 * MAX_VALUE)..((i as i32 + 1) * MAX_VALUE) { + assert_eq!(map.modify(j, |_, x| x * 2), Some(j)); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), MAX_INSERTED_VALUE as usize); + + for i in 0..MAX_INSERTED_VALUE { + assert_eq!(map.get(&i), Some(i * 2)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_overlapped_modification() { + const MAX_VALUE: i32 = 512; + const NUM_THREADS: usize = 64; + + let map = $m::with_capacity(MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.insert(i, 0), None); + } + + let map = std::sync::Arc::new(map); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|_| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for i in 0..MAX_VALUE { + assert!(map.modify(i, |_, x| x + 1).is_some()); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.get(&i), Some(NUM_THREADS as i32)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn insert_or_modify() { + let map = $m::new(); + + assert_eq!(map.insert_or_modify("foo", 1, |_, x| x + 1), None); + assert_eq!(map.get("foo"), Some(1)); + + assert_eq!(map.insert_or_modify("foo", 1, |_, x| x + 1), Some(1)); + assert_eq!(map.get("foo"), Some(2)); + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_insert_or_modify() { + const NUM_THREADS: usize = 64; + const MAX_VALUE: i32 = 512; + + let map = std::sync::Arc::new($m::new()); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|_| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in 0..MAX_VALUE { + map.insert_or_modify(j, 1, |_, x| x + 1); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert_eq!(map.len(), MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.get(&i), Some(NUM_THREADS as i32)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_overlapped_insertion() { + const NUM_THREADS: usize = 64; + const MAX_VALUE: i32 = 512; + + let map = std::sync::Arc::new($m::with_capacity(MAX_VALUE as usize)); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|_| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in 0..MAX_VALUE { + map.insert(j, j); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert_eq!(map.len(), MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.get(&i), Some(i)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_overlapped_growth() { + const NUM_THREADS: usize = 64; + const MAX_VALUE: i32 = 512; + + let map = std::sync::Arc::new($m::with_capacity(1)); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|_| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in 0..MAX_VALUE { + map.insert(j, j); + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert_eq!(map.len(), MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + assert_eq!(map.get(&i), Some(i)); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn concurrent_overlapped_removal() { + const NUM_THREADS: usize = 64; + const MAX_VALUE: i32 = 512; + + let map = $m::with_capacity(MAX_VALUE as usize); + + for i in 0..MAX_VALUE { + map.insert(i, i); + } + + let map = std::sync::Arc::new(map); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let threads: Vec<_> = (0..NUM_THREADS) + .map(|_| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in 0..MAX_VALUE { + let prev_value = map.remove(&j); + + if let Some(v) = prev_value { + assert_eq!(v, j); + } + } + }) + }) + .collect(); + + for result in threads.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + for i in 0..MAX_VALUE { + assert_eq!(map.get(&i), None); + } + + $crate::test_util::run_deferred(); + } + + #[test] + fn drop_value() { + let key_parent = std::sync::Arc::new($crate::test_util::DropNotifier::new()); + let value_parent = std::sync::Arc::new($crate::test_util::DropNotifier::new()); + + { + let map = $m::new(); + + assert_eq!( + map.insert_and( + $crate::test_util::NoisyDropper::new(std::sync::Arc::clone(&key_parent), 0), + $crate::test_util::NoisyDropper::new( + std::sync::Arc::clone(&value_parent), + 0 + ), + |_| () + ), + None + ); + assert!(!map.is_empty()); + assert_eq!(map.len(), 1); + map.get_and(&0, |v| assert_eq!(v, &0)); + + map.remove_and(&0, |v| assert_eq!(v, &0)); + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + assert_eq!(map.get_and(&0, |_| ()), None); + + $crate::test_util::run_deferred(); + + assert!(!key_parent.was_dropped()); + assert!(value_parent.was_dropped()); + } + + $crate::test_util::run_deferred(); + + assert!(key_parent.was_dropped()); + assert!(value_parent.was_dropped()); + } + + #[test] + fn drop_many_values() { + const NUM_VALUES: usize = 1 << 16; + + let key_parents: Vec<_> = std::iter::repeat_with(|| { + std::sync::Arc::new($crate::test_util::DropNotifier::new()) + }) + .take(NUM_VALUES) + .collect(); + let value_parents: Vec<_> = std::iter::repeat_with(|| { + std::sync::Arc::new($crate::test_util::DropNotifier::new()) + }) + .take(NUM_VALUES) + .collect(); + + { + let map = $m::new(); + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + for (i, (this_key_parent, this_value_parent)) in + key_parents.iter().zip(value_parents.iter()).enumerate() + { + assert_eq!( + map.insert_and( + $crate::test_util::NoisyDropper::new( + std::sync::Arc::clone(&this_key_parent), + i + ), + $crate::test_util::NoisyDropper::new( + std::sync::Arc::clone(&this_value_parent), + i + ), + |_| () + ), + None + ); + + assert!(!map.is_empty()); + assert_eq!(map.len(), i + 1); + } + + for i in 0..NUM_VALUES { + assert_eq!( + map.get_key_value_and(&i, |k, v| { + assert_eq!(*k, i); + assert_eq!(*v, i); + }), + Some(()) + ); + } + + for i in 0..NUM_VALUES { + assert_eq!( + map.remove_entry_and(&i, |k, v| { + assert_eq!(*k, i); + assert_eq!(*v, i); + }), + Some(()) + ); + } + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + $crate::test_util::run_deferred(); + + for this_key_parent in key_parents.iter() { + assert!(!this_key_parent.was_dropped()); + } + + for this_value_parent in value_parents.iter() { + assert!(this_value_parent.was_dropped()); + } + + for i in 0..NUM_VALUES { + assert_eq!(map.get_and(&i, |_| ()), None); + } + } + + $crate::test_util::run_deferred(); + + for this_key_parent in key_parents.into_iter() { + assert!(this_key_parent.was_dropped()); + } + + for this_value_parent in value_parents.into_iter() { + assert!(this_value_parent.was_dropped()); + } + } + + #[test] + fn drop_many_values_concurrent() { + const NUM_THREADS: usize = 64; + const NUM_VALUES_PER_THREAD: usize = 512; + const NUM_VALUES: usize = NUM_THREADS * NUM_VALUES_PER_THREAD; + + let key_parents: std::sync::Arc> = std::sync::Arc::new( + std::iter::repeat_with(|| { + std::sync::Arc::new($crate::test_util::DropNotifier::new()) + }) + .take(NUM_VALUES) + .collect(), + ); + let value_parents: std::sync::Arc> = std::sync::Arc::new( + std::iter::repeat_with(|| { + std::sync::Arc::new($crate::test_util::DropNotifier::new()) + }) + .take(NUM_VALUES) + .collect(), + ); + + { + let map = std::sync::Arc::new($m::new()); + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + let barrier = std::sync::Arc::new(std::sync::Barrier::new(NUM_THREADS)); + + let handles: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + let key_parents = std::sync::Arc::clone(&key_parents); + let value_parents = std::sync::Arc::clone(&value_parents); + + std::thread::spawn(move || { + barrier.wait(); + + let these_key_parents = &key_parents + [i * NUM_VALUES_PER_THREAD..(i + 1) * NUM_VALUES_PER_THREAD]; + let these_value_parents = &value_parents + [i * NUM_VALUES_PER_THREAD..(i + 1) * NUM_VALUES_PER_THREAD]; + + for (j, (this_key_parent, this_value_parent)) in these_key_parents + .iter() + .zip(these_value_parents.iter()) + .enumerate() + { + let key_value = i * NUM_VALUES_PER_THREAD + j; + + assert_eq!( + map.insert_and( + $crate::test_util::NoisyDropper::new( + std::sync::Arc::clone(&this_key_parent), + key_value as i32 + ), + $crate::test_util::NoisyDropper::new( + std::sync::Arc::clone(&this_value_parent), + key_value as i32 + ), + |_| () + ), + None + ); + } + }) + }) + .collect(); + + for result in handles.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert!(!map.is_empty()); + assert_eq!(map.len(), NUM_VALUES); + + $crate::test_util::run_deferred(); + + for this_key_parent in key_parents.iter() { + assert!(!this_key_parent.was_dropped()); + } + + for this_value_parent in value_parents.iter() { + assert!(!this_value_parent.was_dropped()); + } + + for i in (0..NUM_VALUES).map(|i| i as i32) { + assert_eq!( + map.get_key_value_and(&i, |k, v| { + assert_eq!(*k, i); + assert_eq!(*v, i); + }), + Some(()) + ); + } + + let handles: Vec<_> = (0..NUM_THREADS) + .map(|i| { + let map = std::sync::Arc::clone(&map); + let barrier = std::sync::Arc::clone(&barrier); + + std::thread::spawn(move || { + barrier.wait(); + + for j in 0..NUM_VALUES_PER_THREAD { + let key_value = (i * NUM_VALUES_PER_THREAD + j) as i32; + + assert_eq!( + map.remove_entry_and(&key_value, |k, v| { + assert_eq!(*k, key_value); + assert_eq!(*v, key_value); + }), + Some(()) + ); + } + }) + }) + .collect(); + + for result in handles.into_iter().map(std::thread::JoinHandle::join) { + assert!(result.is_ok()); + } + + assert!(map.is_empty()); + assert_eq!(map.len(), 0); + + $crate::test_util::run_deferred(); + + for this_key_parent in key_parents.iter() { + assert!(!this_key_parent.was_dropped()); + } + + for this_value_parent in value_parents.iter() { + assert!(this_value_parent.was_dropped()); + } + + for i in (0..NUM_VALUES).map(|i| i as i32) { + assert_eq!(map.get_and(&i, |_| ()), None); + } + } + + $crate::test_util::run_deferred(); + + for this_key_parent in key_parents.iter() { + assert!(this_key_parent.was_dropped()); + } + + for this_value_parent in value_parents.iter() { + assert!(this_value_parent.was_dropped()); + } + } + + #[test] + fn remove_if() { + const NUM_VALUES: i32 = 512; + + let is_even = |_: &i32, v: &i32| *v % 2 == 0; + + let map = $m::new(); + + for i in 0..NUM_VALUES { + assert_eq!(map.insert(i, i), None); + } + + for i in 0..NUM_VALUES { + if is_even(&i, &i) { + assert_eq!(map.remove_if(&i, is_even), Some(i)); + } else { + assert_eq!(map.remove_if(&i, is_even), None); + } + } + + for i in (0..NUM_VALUES).filter(|i| i % 2 == 0) { + assert_eq!(map.get(&i), None); + } + + for i in (0..NUM_VALUES).filter(|i| i % 2 != 0) { + assert_eq!(map.get(&i), Some(i)); + } + + $crate::test_util::run_deferred(); + } + }; +}