diff --git a/CHANGELOG.md b/CHANGELOG.md index 9530a68..42f1dd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,11 @@ ## Version 3 +3.4.3 + +* Maximum theoretically capacity is adjusted to `2^(usize::BITS - 2)`. +* Minor `Future` size improvement. + 3.4.2 * Minor optimization. diff --git a/Cargo.toml b/Cargo.toml index 0fdfa99..a310422 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,7 @@ name = "scc" description = "A collection of high-performance containers providing both asynchronous and synchronous interfaces" documentation = "https://docs.rs/scc" -version = "3.4.2" +version = "3.4.3" authors = ["wvwwvwwv "] edition = "2024" rust-version = "1.85.0" @@ -26,7 +26,7 @@ serde = { version = "1.0", optional = true } loom = ["dep:loom", "saa/loom", "sdd/loom"] [dev-dependencies] -criterion = { version = "0.7", features = ["async_futures"] } +criterion = { version = "0.8", features = ["async_futures"] } fnv = "1.0" futures = "0.3" proptest = "1.9" diff --git a/src/async_helper.rs b/src/async_helper.rs index bf490bc..fca8a60 100644 --- a/src/async_helper.rs +++ b/src/async_helper.rs @@ -30,6 +30,12 @@ pub(crate) trait TryWait { fn try_wait(&mut self, lock: &Lock); } +/// Returns a fake [`Guard`] reference for methods that require a [`Guard`] to check the lifetime. +#[inline] +pub(super) const fn fake_guard() -> &'static Guard { + unsafe { &*ptr::from_ref(&FAKE_GUARD_GLOBAL).cast::() } +} + impl AsyncGuard { /// Returns `true` if the [`AsyncGuard`] contains a valid [`Guard`]. #[inline] @@ -56,18 +62,21 @@ impl AsyncGuard { } } - /// Loads the content of the [`AtomicShared`] without exposing the [`Guard`]. + /// Loads the content of the [`AtomicShared`] without exposing the [`Guard`] or checking tag + /// bits. #[inline] - pub(crate) fn load(&self, atomic_ptr: &AtomicShared, mo: Ordering) -> Option<&T> { - atomic_ptr.load(mo, self.guard()).as_ref() + pub(crate) fn load_unchecked( + &self, + atomic_ptr: &AtomicShared, + mo: Ordering, + ) -> Option<&T> { + unsafe { atomic_ptr.load(mo, self.guard()).as_ref_unchecked() } } /// Checks if the reference is valid. #[inline] pub(crate) fn check_ref(&self, atomic_ptr: &AtomicShared, r: &T, mo: Ordering) -> bool { - atomic_ptr - .load(mo, self.guard()) - .as_ref() + self.load_unchecked(atomic_ptr, mo) .is_some_and(|s| ptr::eq(s, r)) } } @@ -120,3 +129,5 @@ impl TryWait for () { let _: Result<_, _> = pinned_pager.poll_sync(); } } + +static FAKE_GUARD_GLOBAL: usize = 0; diff --git a/src/hash_cache.rs b/src/hash_cache.rs index bc53735..5957095 100644 --- a/src/hash_cache.rs +++ b/src/hash_cache.rs @@ -5,17 +5,20 @@ use std::fmt::{self, Debug}; use std::hash::{BuildHasher, Hash}; use std::mem::replace; use std::ops::{Deref, DerefMut, RangeInclusive}; -use std::pin::pin; +#[cfg(not(feature = "loom"))] use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::Relaxed; +#[cfg(feature = "loom")] +use loom::sync::atomic::AtomicUsize; use sdd::{AtomicShared, Guard, Shared, Tag}; use super::Equivalent; +use super::async_helper::fake_guard; +use super::hash_table::MAXIMUM_CAPACITY_LIMIT; use super::hash_table::bucket::{CACHE, DoublyLinkedList, EntryPtr}; use super::hash_table::bucket_array::BucketArray; use super::hash_table::{HashTable, LockedBucket}; -use crate::async_helper::AsyncGuard; /// Scalable concurrent 32-way associative cache backed by [`HashMap`](super::HashMap). /// @@ -189,7 +192,7 @@ where let maximum_capacity = maximum_capacity .max(minimum_capacity.load(Relaxed)) .max(BucketArray::::minimum_capacity()) - .min(1_usize << (usize::BITS - 1)) + .min(MAXIMUM_CAPACITY_LIMIT) .next_power_of_two(); HashCache { bucket_array: array, @@ -219,10 +222,9 @@ where #[inline] pub async fn entry_async(&self, key: K) -> Entry<'_, K, V, H> { let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let prolonged_guard = self.prolonged_guard_ref(async_guard.guard()); - let entry_ptr = locked_bucket.search(&key, hash, prolonged_guard); + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + let entry_ptr = locked_bucket.search(&key, hash, fake_guard); if entry_ptr.is_valid() { Entry::Occupied(OccupiedEntry { hashcache: self, @@ -341,14 +343,13 @@ where #[inline] pub async fn put_async(&self, key: K, val: V) -> Result, (K, V)> { let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let guard = async_guard.guard(); - if locked_bucket.search(&key, hash, guard).is_valid() { + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + if locked_bucket.search(&key, hash, fake_guard).is_valid() { Err((key, val)) } else { let evicted = locked_bucket.evict_lru_head(locked_bucket.data_block); - let entry_ptr = locked_bucket.insert(hash, (key, val), guard); + let entry_ptr = locked_bucket.insert(hash, (key, val), fake_guard); locked_bucket.update_lru_tail(&entry_ptr); Ok(evicted) } @@ -441,10 +442,9 @@ where #[inline] pub async fn replace_async(&self, key: K) -> ReplaceResult<'_, K, V, H> { let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let prolonged_guard = self.prolonged_guard_ref(async_guard.guard()); - let mut entry_ptr = locked_bucket.search(&key, hash, prolonged_guard); + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(&key, hash, fake_guard); if entry_ptr.is_valid() { let prev_key = replace( &mut entry_ptr @@ -610,10 +610,9 @@ where Q: Equivalent + Hash + ?Sized, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.optional_writer_async(hash, &async_guard).await?; - let prolonged_guard = self.prolonged_guard_ref(async_guard.guard()); - let entry_ptr = locked_bucket.search(key, hash, prolonged_guard); + let locked_bucket = self.optional_writer_async(hash).await?; + let fake_guard = fake_guard(); + let entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() { locked_bucket.writer.update_lru_tail(&entry_ptr); return Some(OccupiedEntry { @@ -686,8 +685,7 @@ where Q: Equivalent + Hash + ?Sized, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - self.reader_async(key, hash, reader, &async_guard).await + self.reader_async(key, hash, reader).await } /// Reads a key-value pair. @@ -731,7 +729,8 @@ where where Q: Equivalent + Hash + ?Sized, { - self.read_async(key, |_, _| ()).await.is_some() + let hash = self.hash(key); + self.reader_async(key, hash, |_, _| ()).await.is_some() } /// Returns `true` if the [`HashCache`] contains a value for the specified key. @@ -778,11 +777,11 @@ where Q: Equivalent + Hash + ?Sized, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - let mut locked_bucket = self.optional_writer_async(hash, &async_guard).await?; - let mut entry_ptr = locked_bucket.search(key, hash, async_guard.guard()); + let mut locked_bucket = self.optional_writer_async(hash).await?; + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() && condition(&mut locked_bucket.entry_mut(&mut entry_ptr).1) { - Some(locked_bucket.remove(self, &mut entry_ptr, async_guard.guard())) + Some(locked_bucket.remove(self, &mut entry_ptr, &Guard::new())) } else { None } @@ -845,12 +844,11 @@ where /// ``` #[inline] pub async fn iter_async bool>(&self, mut f: F) -> bool { - let async_guard = pin!(AsyncGuard::default()); let mut result = true; - self.for_each_reader_async(&async_guard, |reader, data_block| { - let guard = async_guard.guard(); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&reader, guard) { + self.for_each_reader_async(|reader, data_block| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&reader, fake_guard) { let (k, v) = entry_ptr.get(data_block); if !f(k, v) { result = false; @@ -937,17 +935,16 @@ where &self, mut f: F, ) -> bool { - let async_guard = pin!(AsyncGuard::default()); let mut result = true; - self.for_each_writer_async(0, 0, &async_guard, |mut locked_bucket, removed| { - let guard = async_guard.guard(); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&locked_bucket.writer, guard) { + self.for_each_writer_async(0, 0, |mut locked_bucket, removed| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&locked_bucket.writer, fake_guard) { let consumable_entry = ConsumableEntry { locked_bucket: &mut locked_bucket, entry_ptr: &mut entry_ptr, remove_probe: removed, - guard, + guard: fake_guard, }; if !f(consumable_entry) { result = false; @@ -1329,12 +1326,12 @@ where } #[inline] - fn bucket_array(&self) -> &AtomicShared> { + fn bucket_array_var(&self) -> &AtomicShared> { &self.bucket_array } #[inline] - fn minimum_capacity(&self) -> &AtomicUsize { + fn minimum_capacity_var(&self) -> &AtomicUsize { &self.minimum_capacity } diff --git a/src/hash_index.rs b/src/hash_index.rs index 2232315..e845dd5 100644 --- a/src/hash_index.rs +++ b/src/hash_index.rs @@ -8,17 +8,20 @@ use std::ops::{Deref, RangeInclusive}; use std::panic::UnwindSafe; use std::pin::pin; use std::ptr; +use std::sync::atomic::AtomicU8; +#[cfg(not(feature = "loom"))] +use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release}; -use std::sync::atomic::{AtomicU8, AtomicUsize}; -use sdd::{AtomicShared, Epoch, Guard, Ptr, Shared, Tag}; +#[cfg(feature = "loom")] +use loom::sync::atomic::AtomicUsize; +use sdd::{AtomicShared, Epoch, Guard, Shared, Tag}; use super::Equivalent; -use super::hash_table::HashTable; +use super::async_helper::{AsyncGuard, fake_guard}; use super::hash_table::bucket::{Bucket, EntryPtr, INDEX}; use super::hash_table::bucket_array::BucketArray; -use crate::async_helper::AsyncGuard; -use crate::hash_table::LockedBucket; +use super::hash_table::{HashTable, LockedBucket}; /// Scalable concurrent hash index. /// @@ -264,12 +267,11 @@ where /// ``` #[inline] pub async fn entry_async(&self, key: K) -> Entry<'_, K, V, H> { + self.reclaim_memory(); let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let prolonged_guard = self.prolonged_guard_ref(async_guard.guard()); - let entry_ptr = locked_bucket.search(&key, hash, prolonged_guard); + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + let entry_ptr = locked_bucket.search(&key, hash, fake_guard); if entry_ptr.is_valid() { Entry::Occupied(OccupiedEntry { hashindex: self, @@ -346,9 +348,9 @@ where /// ``` #[inline] pub fn try_entry(&self, key: K) -> Option> { + self.reclaim_memory(); let hash = self.hash(&key); let guard = Guard::new(); - self.reclaim_memory(&guard); let prolonged_guard = self.prolonged_guard_ref(&guard); let locked_bucket = self.try_reserve_bucket(hash, prolonged_guard)?; let entry_ptr = locked_bucket.search(&key, hash, prolonged_guard); @@ -430,13 +432,12 @@ where &self, mut pred: P, ) -> Option> { - let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); + self.reclaim_memory(); let mut entry = None; - self.for_each_writer_async(0, 0, &async_guard, |locked_bucket, _| { - let guard = self.prolonged_guard_ref(async_guard.guard()); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&locked_bucket.writer, guard) { + self.for_each_writer_async(0, 0, |locked_bucket, _| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&locked_bucket.writer, fake_guard) { let (k, v) = locked_bucket.entry(&entry_ptr); if pred(k, v) { entry = Some(OccupiedEntry { @@ -473,9 +474,9 @@ where &self, mut pred: P, ) -> Option> { + self.reclaim_memory(); let mut entry = None; let guard = Guard::new(); - self.reclaim_memory(&guard); let prolonged_guard = self.prolonged_guard_ref(&guard); self.for_each_writer_sync(0, 0, prolonged_guard, |locked_bucket, _| { let mut entry_ptr = EntryPtr::new(prolonged_guard); @@ -515,10 +516,10 @@ where /// ``` #[inline] pub async fn insert_async(&self, key: K, val: V) -> Result<(), (K, V)> { + self.reclaim_memory(); let hash = self.hash(&key); let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); - let locked_bucket = self.writer_async(hash, &async_guard).await; + let locked_bucket = self.writer_async(hash).await; let guard = async_guard.guard(); if locked_bucket.search(&key, hash, guard).is_valid() { Err((key, val)) @@ -550,9 +551,9 @@ where /// ``` #[inline] pub fn insert_sync(&self, key: K, val: V) -> Result<(), (K, V)> { + self.reclaim_memory(); let hash = self.hash(&key); let guard = Guard::new(); - self.reclaim_memory(&guard); let locked_bucket = self.writer_sync(hash, &guard); if locked_bucket.search(&key, hash, &guard).is_valid() { Err((key, val)) @@ -633,15 +634,15 @@ where where Q: Equivalent + Hash + ?Sized, { + self.reclaim_memory(); let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); - let Some(mut locked_bucket) = self.optional_writer_async(hash, &async_guard).await else { + let Some(mut locked_bucket) = self.optional_writer_async(hash).await else { return false; }; - let mut entry_ptr = locked_bucket.search(key, hash, async_guard.guard()); + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() && condition(&mut locked_bucket.entry_mut(&mut entry_ptr).1) { - locked_bucket.mark_removed(self, &mut entry_ptr, async_guard.guard()); + locked_bucket.mark_removed(self, &mut entry_ptr, &Guard::new()); true } else { false @@ -671,9 +672,9 @@ where where Q: Equivalent + Hash + ?Sized, { + self.reclaim_memory(); let hash = self.hash(key); let guard = Guard::default(); - self.reclaim_memory(&guard); let Some(mut locked_bucket) = self.optional_writer_sync(hash, &guard) else { return false; }; @@ -707,12 +708,11 @@ where where Q: Equivalent + Hash + ?Sized, { + self.reclaim_memory(); let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); - let locked_bucket = self.optional_writer_async(hash, &async_guard).await?; - let guard = self.prolonged_guard_ref(async_guard.guard()); - let entry_ptr = locked_bucket.search(key, hash, guard); + let locked_bucket = self.optional_writer_async(hash).await?; + let fake_guard = fake_guard(); + let entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() { return Some(OccupiedEntry { hashindex: self, @@ -747,9 +747,9 @@ where where Q: Equivalent + Hash + ?Sized, { + self.reclaim_memory(); let hash = self.hash(key); let guard = Guard::new(); - self.reclaim_memory(&guard); let prolonged_guard = self.prolonged_guard_ref(&guard); let locked_bucket = self.optional_writer_sync(hash, prolonged_guard)?; let entry_ptr = locked_bucket.search(key, hash, prolonged_guard); @@ -797,7 +797,7 @@ where where Q: Equivalent + Hash + ?Sized, { - self.reclaim_memory(guard); + self.reclaim_memory(); self.peek_entry(key, self.hash(key), guard).map(|(_, v)| v) } @@ -833,8 +833,8 @@ where where Q: Equivalent + Hash + ?Sized, { + self.reclaim_memory(); let guard = Guard::new(); - self.reclaim_memory(&guard); self.peek_entry(key, self.hash(key), &guard) .map(|(k, v)| reader(k, v)) } @@ -882,13 +882,12 @@ where /// ``` #[inline] pub async fn iter_async bool>(&self, mut f: F) -> bool { - let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); + self.reclaim_memory(); let mut result = true; - self.for_each_reader_async(&async_guard, |reader, data_block| { - let guard = async_guard.guard(); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&reader, guard) { + self.for_each_reader_async(|reader, data_block| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&reader, fake_guard) { let (k, v) = entry_ptr.get(data_block); if !f(k, v) { result = false; @@ -927,9 +926,9 @@ where /// ``` #[inline] pub fn iter_sync bool>(&self, mut f: F) -> bool { + self.reclaim_memory(); let mut result = true; let guard = Guard::new(); - self.reclaim_memory(&guard); self.for_each_reader_sync(&guard, |reader, data_block| { let mut entry_ptr = EntryPtr::new(&guard); while entry_ptr.move_to_next(&reader, &guard) { @@ -962,15 +961,16 @@ where /// ``` #[inline] pub async fn retain_async bool>(&self, mut pred: F) { - let async_guard = pin!(AsyncGuard::default()); - self.reclaim_memory(async_guard.guard()); - self.for_each_writer_async(0, 0, &async_guard, |mut locked_bucket, removed| { - let guard = async_guard.guard(); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&locked_bucket.writer, guard) { + self.reclaim_memory(); + self.for_each_writer_async(0, 0, |mut locked_bucket, removed| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&locked_bucket.writer, fake_guard) { let (k, v) = locked_bucket.entry_mut(&mut entry_ptr); if !pred(k, v) { - locked_bucket.writer.mark_removed(&mut entry_ptr, guard); + locked_bucket + .writer + .mark_removed(&mut entry_ptr, fake_guard); *removed = true; } } @@ -1004,8 +1004,8 @@ where /// ``` #[inline] pub fn retain_sync bool>(&self, mut pred: F) { + self.reclaim_memory(); let guard = Guard::new(); - self.reclaim_memory(&guard); self.for_each_writer_sync(0, 0, &guard, |mut locked_bucket, removed| { let mut entry_ptr = EntryPtr::new(&guard); while entry_ptr.move_to_next(&locked_bucket.writer, &guard) { @@ -1122,10 +1122,10 @@ where /// /// let hashindex: HashIndex = HashIndex::default(); /// - /// assert_eq!(hashindex.capacity_range(), 0..=(1_usize << (usize::BITS - 1))); + /// assert_eq!(hashindex.capacity_range(), 0..=(1_usize << (usize::BITS - 2))); /// /// let reserved = hashindex.reserve(1000); - /// assert_eq!(hashindex.capacity_range(), 1000..=(1_usize << (usize::BITS - 1))); + /// assert_eq!(hashindex.capacity_range(), 1000..=(1_usize << (usize::BITS - 2))); /// ``` #[inline] pub fn capacity_range(&self) -> RangeInclusive { @@ -1196,24 +1196,32 @@ where /// Reclaims memory by dropping all garbage bucket arrays if they are unreachable. #[inline] - fn reclaim_memory(&self, guard: &Guard) { - let head_ptr = self.garbage_chain.load(Acquire, guard); + fn reclaim_memory(&self) { + let head_ptr = self.garbage_chain.load(Acquire, fake_guard()); if !head_ptr.is_null() { - self.dealloc_garbage(head_ptr, guard); + self.dealloc_garbage(); } } /// Deallocates the supplied bucket array. - fn dealloc_garbage(&self, ptr: Ptr>, guard: &Guard) { + fn dealloc_garbage(&self) { + let guard = Guard::new(); + let head_ptr = self.garbage_chain.load(Acquire, &guard); + if head_ptr.is_null() { + return; + } let garbage_epoch = self.garbage_epoch.load(Acquire); if Epoch::try_from(garbage_epoch).is_ok_and(|e| !e.in_same_generation(guard.epoch())) { - if let Ok((mut garbage_head, _)) = - self.garbage_chain - .compare_exchange(ptr, (None, Tag::None), Acquire, Relaxed, guard) - { + if let Ok((mut garbage_head, _)) = self.garbage_chain.compare_exchange( + head_ptr, + (None, Tag::None), + Acquire, + Relaxed, + &guard, + ) { while let Some(garbage_bucket_array) = garbage_head { garbage_head = garbage_bucket_array - .bucket_link() + .linked_array_var() .swap((None, Tag::None), Acquire) .0; let dropped = unsafe { garbage_bucket_array.drop_in_place() }; @@ -1250,8 +1258,8 @@ where { #[inline] fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.reclaim_memory(); let guard = Guard::new(); - self.reclaim_memory(&guard); f.debug_map().entries(self.iter(&guard)).finish() } } @@ -1341,7 +1349,7 @@ where let mut garbage_head = self.garbage_chain.swap((None, Tag::None), Acquire).0; while let Some(garbage_bucket_array) = garbage_head { garbage_head = garbage_bucket_array - .bucket_link() + .linked_array_var() .swap((None, Tag::None), Acquire) .0; let dropped = unsafe { garbage_bucket_array.drop_in_place() }; @@ -1381,7 +1389,7 @@ where #[inline] fn defer_reclaim(&self, bucket_array: Shared>, guard: &Guard) { - self.reclaim_memory(guard); + self.reclaim_memory(); self.garbage_epoch.swap(u8::from(guard.epoch()), Release); let (Some(prev_head), _) = self .garbage_chain @@ -1391,17 +1399,17 @@ where }; // The bucket array will be dropped when the epoch enters the next generation. bucket_array - .bucket_link() + .linked_array_var() .swap((Some(prev_head), Tag::None), Release); } #[inline] - fn bucket_array(&self) -> &AtomicShared> { + fn bucket_array_var(&self) -> &AtomicShared> { &self.bucket_array } #[inline] - fn minimum_capacity(&self) -> &AtomicUsize { + fn minimum_capacity_var(&self) -> &AtomicUsize { &self.minimum_capacity } } @@ -1414,8 +1422,8 @@ where { #[inline] fn eq(&self, other: &Self) -> bool { + self.reclaim_memory(); let guard = Guard::new(); - self.reclaim_memory(&guard); if !self .iter(&guard) .any(|(k, v)| other.peek_with(k, |_, ov| v == ov) != Some(true)) @@ -1723,14 +1731,12 @@ where pub async fn remove_and_async(self) -> Option> { let hashindex = self.hashindex; let mut entry_ptr = self.entry_ptr.clone(); - let mut async_guard = pin!(AsyncGuard::default()); - self.locked_bucket.writer.mark_removed( - &mut entry_ptr, - hashindex.prolonged_guard_ref(async_guard.guard()), - ); + self.locked_bucket + .writer + .mark_removed(&mut entry_ptr, hashindex.prolonged_guard_ref(&Guard::new())); if let Some(locked_bucket) = self .locked_bucket - .next_async(hashindex, &mut entry_ptr, &mut async_guard) + .next_async(hashindex, &mut entry_ptr) .await { return Some(OccupiedEntry { @@ -1816,10 +1822,9 @@ where pub async fn next_async(self) -> Option> { let hashindex = self.hashindex; let mut entry_ptr = self.entry_ptr.clone(); - let mut async_guard = pin!(AsyncGuard::default()); if let Some(locked_bucket) = self .locked_bucket - .next_async(hashindex, &mut entry_ptr, &mut async_guard) + .next_async(hashindex, &mut entry_ptr) .await { return Some(OccupiedEntry { @@ -2090,7 +2095,7 @@ where debug_assert!(result >= self.additional); let guard = Guard::new(); - if let Some(current_array) = self.hashindex.bucket_array.load(Acquire, &guard).as_ref() { + if let Some(current_array) = self.hashindex.bucket_array(&guard) { self.try_shrink_or_rebuild(current_array, 0, &guard); } } @@ -2123,13 +2128,8 @@ where array } else { // Start scanning. - let current_array = self - .hashindex - .bucket_array() - .load(Acquire, self.guard) - .as_ref()?; - let old_array_ptr = current_array.old_array(self.guard); - let array = if let Some(old_array) = old_array_ptr.as_ref() { + let current_array = self.hashindex.bucket_array(self.guard)?; + let array = if let Some(old_array) = current_array.linked_array(self.guard) { old_array } else { current_array @@ -2154,11 +2154,7 @@ where if self.index + 1 == array.len() { // Move to a newer bucket array. self.index = 0; - let current_array = self - .hashindex - .bucket_array() - .load(Acquire, self.guard) - .as_ref()?; + let current_array = self.hashindex.bucket_array(self.guard)?; if self .bucket_array .as_ref() @@ -2168,7 +2164,7 @@ where break; } - array = if let Some(old_array) = current_array.old_array(self.guard).as_ref() { + array = if let Some(old_array) = current_array.linked_array(self.guard) { if self .bucket_array .as_ref() diff --git a/src/hash_map.rs b/src/hash_map.rs index 502aa05..e733c38 100644 --- a/src/hash_map.rs +++ b/src/hash_map.rs @@ -5,18 +5,19 @@ use std::fmt::{self, Debug}; use std::hash::{BuildHasher, Hash}; use std::mem::replace; use std::ops::{Deref, DerefMut, RangeInclusive}; -use std::pin::pin; +#[cfg(not(feature = "loom"))] use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering::{Acquire, Relaxed}; +use std::sync::atomic::Ordering::Relaxed; +#[cfg(feature = "loom")] +use loom::sync::atomic::AtomicUsize; use sdd::{AtomicShared, Guard, Shared, Tag}; use super::Equivalent; -use super::hash_table::HashTable; +use super::async_helper::fake_guard; use super::hash_table::bucket::{EntryPtr, MAP}; use super::hash_table::bucket_array::BucketArray; -use crate::async_helper::AsyncGuard; -use crate::hash_table::LockedBucket; +use super::hash_table::{HashTable, LockedBucket}; /// Scalable concurrent hash map. /// @@ -296,10 +297,9 @@ where #[inline] pub async fn entry_async(&self, key: K) -> Entry<'_, K, V, H> { let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let prolonged_guard = self.prolonged_guard_ref(async_guard.guard()); - let entry_ptr = locked_bucket.search(&key, hash, prolonged_guard); + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + let entry_ptr = locked_bucket.search(&key, hash, fake_guard); if entry_ptr.is_valid() { Entry::Occupied(OccupiedEntry { hashmap: self, @@ -455,12 +455,11 @@ where &self, mut pred: P, ) -> Option> { - let async_guard = pin!(AsyncGuard::default()); let mut entry = None; - self.for_each_writer_async(0, 0, &async_guard, |locked_bucket, _| { - let guard = self.prolonged_guard_ref(async_guard.guard()); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&locked_bucket.writer, guard) { + self.for_each_writer_async(0, 0, |locked_bucket, _| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&locked_bucket.writer, fake_guard) { let (k, v) = locked_bucket.entry(&entry_ptr); if pred(k, v) { entry = Some(OccupiedEntry { @@ -540,13 +539,12 @@ where #[inline] pub async fn insert_async(&self, key: K, val: V) -> Result<(), (K, V)> { let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let guard = async_guard.guard(); - if locked_bucket.search(&key, hash, guard).is_valid() { + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + if locked_bucket.search(&key, hash, fake_guard).is_valid() { Err((key, val)) } else { - locked_bucket.insert(hash, (key, val), guard); + locked_bucket.insert(hash, (key, val), fake_guard); Ok(()) } } @@ -655,10 +653,9 @@ where U: FnOnce(&K, &mut V) -> R, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - let mut locked_bucket = self.optional_writer_async(hash, &async_guard).await?; - let guard = async_guard.guard(); - let mut entry_ptr = locked_bucket.search(key, hash, guard); + let mut locked_bucket = self.optional_writer_async(hash).await?; + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() { let (k, v) = locked_bucket.entry_mut(&mut entry_ptr); Some(updater(k, v)) @@ -748,10 +745,9 @@ where #[inline] pub async fn replace_async(&self, key: K) -> ReplaceResult<'_, K, V, H> { let hash = self.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.writer_async(hash, &async_guard).await; - let prolonged_guard = self.prolonged_guard_ref(async_guard.guard()); - let mut entry_ptr = locked_bucket.search(&key, hash, prolonged_guard); + let locked_bucket = self.writer_async(hash).await; + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(&key, hash, fake_guard); if entry_ptr.is_valid() { let prev_key = replace( &mut entry_ptr @@ -918,11 +914,11 @@ where Q: Equivalent + Hash + ?Sized, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - let mut locked_bucket = self.optional_writer_async(hash, &async_guard).await?; - let mut entry_ptr = locked_bucket.search(key, hash, async_guard.guard()); + let mut locked_bucket = self.optional_writer_async(hash).await?; + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() && condition(&mut locked_bucket.entry_mut(&mut entry_ptr).1) { - Some(locked_bucket.remove(self, &mut entry_ptr, async_guard.guard())) + Some(locked_bucket.remove(self, &mut entry_ptr, &Guard::new())) } else { None } @@ -985,10 +981,9 @@ where Q: Equivalent + Hash + ?Sized, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - let locked_bucket = self.optional_writer_async(hash, &async_guard).await?; - let guard = self.prolonged_guard_ref(async_guard.guard()); - let entry_ptr = locked_bucket.search(key, hash, guard); + let locked_bucket = self.optional_writer_async(hash).await?; + let fake_guard = fake_guard(); + let entry_ptr = locked_bucket.search(key, hash, fake_guard); if entry_ptr.is_valid() { return Some(OccupiedEntry { hashmap: self, @@ -1059,8 +1054,7 @@ where Q: Equivalent + Hash + ?Sized, { let hash = self.hash(key); - let async_guard = pin!(AsyncGuard::default()); - self.reader_async(key, hash, reader, &async_guard).await + self.reader_async(key, hash, reader).await } /// Reads a key-value pair. @@ -1104,7 +1098,8 @@ where where Q: Equivalent + Hash + ?Sized, { - self.read_async(key, |_, _| ()).await.is_some() + let hash = self.hash(key); + self.reader_async(key, hash, |_, _| ()).await.is_some() } /// Returns `true` if the [`HashMap`] contains a value for the specified key. @@ -1150,12 +1145,11 @@ where /// ``` #[inline] pub async fn iter_async bool>(&self, mut f: F) -> bool { - let async_guard = pin!(AsyncGuard::default()); let mut result = true; - self.for_each_reader_async(&async_guard, |reader, data_block| { - let guard = async_guard.guard(); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&reader, guard) { + self.for_each_reader_async(|reader, data_block| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&reader, fake_guard) { let (k, v) = entry_ptr.get(data_block); if !f(k, v) { result = false; @@ -1242,17 +1236,16 @@ where &self, mut f: F, ) -> bool { - let async_guard = pin!(AsyncGuard::default()); let mut result = true; - self.for_each_writer_async(0, 0, &async_guard, |mut locked_bucket, removed| { - let guard = async_guard.guard(); - let mut entry_ptr = EntryPtr::new(guard); - while entry_ptr.move_to_next(&locked_bucket.writer, guard) { + self.for_each_writer_async(0, 0, |mut locked_bucket, removed| { + let fake_guard = fake_guard(); + let mut entry_ptr = EntryPtr::new(fake_guard); + while entry_ptr.move_to_next(&locked_bucket.writer, fake_guard) { let consumable_entry = ConsumableEntry { locked_bucket: &mut locked_bucket, entry_ptr: &mut entry_ptr, remove_probe: removed, - guard, + guard: fake_guard, }; if !f(consumable_entry) { result = false; @@ -1486,10 +1479,10 @@ where /// /// let hashmap: HashMap = HashMap::default(); /// - /// assert_eq!(hashmap.capacity_range(), 0..=(1_usize << (usize::BITS - 1))); + /// assert_eq!(hashmap.capacity_range(), 0..=(1_usize << (usize::BITS - 2))); /// /// let reserved = hashmap.reserve(1000); - /// assert_eq!(hashmap.capacity_range(), 1000..=(1_usize << (usize::BITS - 1))); + /// assert_eq!(hashmap.capacity_range(), 1000..=(1_usize << (usize::BITS - 2))); /// ``` #[inline] pub fn capacity_range(&self) -> RangeInclusive { @@ -1670,12 +1663,12 @@ where } #[inline] - fn bucket_array(&self) -> &AtomicShared> { + fn bucket_array_var(&self) -> &AtomicShared> { &self.bucket_array } #[inline] - fn minimum_capacity(&self) -> &AtomicUsize { + fn minimum_capacity_var(&self) -> &AtomicUsize { &self.minimum_capacity } } @@ -2042,17 +2035,12 @@ where pub async fn remove_and_async(self) -> ((K, V), Option>) { let hashmap = self.hashmap; let mut entry_ptr = self.entry_ptr.clone(); - let mut async_guard = pin!(AsyncGuard::default()); let entry = self.locked_bucket.writer.remove( self.locked_bucket.data_block, &mut entry_ptr, - hashmap.prolonged_guard_ref(async_guard.guard()), + hashmap.prolonged_guard_ref(&Guard::new()), ); - if let Some(locked_bucket) = self - .locked_bucket - .next_async(hashmap, &mut entry_ptr, &mut async_guard) - .await - { + if let Some(locked_bucket) = self.locked_bucket.next_async(hashmap, &mut entry_ptr).await { return ( entry, Some(OccupiedEntry { @@ -2145,12 +2133,7 @@ where pub async fn next_async(self) -> Option> { let hashmap = self.hashmap; let mut entry_ptr = self.entry_ptr.clone(); - let mut async_guard = pin!(AsyncGuard::default()); - if let Some(locked_bucket) = self - .locked_bucket - .next_async(hashmap, &mut entry_ptr, &mut async_guard) - .await - { + if let Some(locked_bucket) = self.locked_bucket.next_async(hashmap, &mut entry_ptr).await { return Some(OccupiedEntry { hashmap, locked_bucket, @@ -2441,7 +2424,7 @@ where debug_assert!(result >= self.additional); let guard = Guard::new(); - if let Some(current_array) = self.hashmap.bucket_array.load(Acquire, &guard).as_ref() { + if let Some(current_array) = self.hashmap.bucket_array(&guard) { self.try_shrink_or_rebuild(current_array, 0, &guard); } } diff --git a/src/hash_set.rs b/src/hash_set.rs index adbb2bb..3877a05 100644 --- a/src/hash_set.rs +++ b/src/hash_set.rs @@ -7,14 +7,13 @@ use std::fmt::{self, Debug}; use std::hash::{BuildHasher, Hash}; use std::mem::swap; use std::ops::{Deref, RangeInclusive}; -use std::pin::pin; use sdd::Guard; +use super::async_helper::fake_guard; +use super::hash_map; use super::hash_table::HashTable; use super::{Equivalent, HashMap}; -use crate::async_helper::AsyncGuard; -use crate::hash_map; /// Scalable concurrent hash set. /// @@ -220,16 +219,15 @@ where #[inline] pub async fn replace_async(&self, mut key: K) -> Option { let hash = self.map.hash(&key); - let async_guard = pin!(AsyncGuard::default()); - let mut locked_bucket = self.map.writer_async(hash, &async_guard).await; - let guard = async_guard.guard(); - let mut entry_ptr = locked_bucket.search(&key, hash, guard); + let mut locked_bucket = self.map.writer_async(hash).await; + let fake_guard = fake_guard(); + let mut entry_ptr = locked_bucket.search(&key, hash, fake_guard); if entry_ptr.is_valid() { let k = &mut locked_bucket.entry_mut(&mut entry_ptr).0; swap(k, &mut key); Some(key) } else { - locked_bucket.insert(hash, (key, ()), guard); + locked_bucket.insert(hash, (key, ()), fake_guard); None } } @@ -744,10 +742,10 @@ where /// /// let hashset: HashSet = HashSet::default(); /// - /// assert_eq!(hashset.capacity_range(), 0..=(1_usize << (usize::BITS - 1))); + /// assert_eq!(hashset.capacity_range(), 0..=(1_usize << (usize::BITS - 2))); /// /// let reserved = hashset.reserve(1000); - /// assert_eq!(hashset.capacity_range(), 1000..=(1_usize << (usize::BITS - 1))); + /// assert_eq!(hashset.capacity_range(), 1000..=(1_usize << (usize::BITS - 2))); /// ``` #[inline] pub fn capacity_range(&self) -> RangeInclusive { diff --git a/src/hash_table.rs b/src/hash_table.rs index 6218187..ae57062 100644 --- a/src/hash_table.rs +++ b/src/hash_table.rs @@ -4,19 +4,23 @@ pub mod bucket_array; use std::hash::{BuildHasher, Hash}; use std::mem::forget; use std::ops::Deref; -use std::pin::Pin; +use std::pin::pin; use std::ptr::{self, NonNull, from_ref}; + +#[cfg(not(feature = "loom"))] use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release}; use bucket::{BUCKET_LEN, CACHE, DataBlock, EntryPtr, INDEX, LruList, Reader, Writer}; use bucket_array::BucketArray; +#[cfg(feature = "loom")] +use loom::sync::atomic::AtomicUsize; use sdd::{AtomicShared, Guard, Ptr, Shared, Tag}; use super::Equivalent; +use super::async_helper::{AsyncGuard, fake_guard}; use super::exit_guard::ExitGuard; -use crate::async_helper::AsyncGuard; -use crate::hash_table::bucket::Bucket; +use super::hash_table::bucket::Bucket; /// `HashTable` defines common functions for hash table implementations. pub(super) trait HashTable @@ -33,11 +37,20 @@ where self.hasher().hash_one(key) } - /// Returns a reference to its [`BuildHasher`]. + /// Returns its [`BuildHasher`]. fn hasher(&self) -> &H; /// Returns a reference to the [`BucketArray`] pointer. - fn bucket_array(&self) -> &AtomicShared>; + fn bucket_array_var(&self) -> &AtomicShared>; + + /// Returns a reference to the current [`BucketArray`]. + fn bucket_array<'g>(&self, guard: &'g Guard) -> Option<&'g BucketArray> { + unsafe { + self.bucket_array_var() + .load(Acquire, guard) + .as_ref_unchecked() + } + } /// Passes the bucket array to the garbage collector associated with the hash table type. #[inline] @@ -51,20 +64,27 @@ where where Q: Equivalent + Hash + ?Sized, { - self.bucket_array() - .load(Acquire, &Guard::new()) - .as_ref() - .map_or(0, |a| a.calculate_bucket_index(self.hash(key))) + unsafe { + self.bucket_array_var() + .load(Acquire, &Guard::new()) + .as_ref_unchecked() + .map_or(0, |a| a.bucket_index(self.hash(key))) + } } - /// Returns the minimum allowed capacity. - fn minimum_capacity(&self) -> &AtomicUsize; + /// Returns a reference to a variable containing the minimum allowed capacity. + fn minimum_capacity_var(&self) -> &AtomicUsize; + + /// Returns the current minimum allowed capacity. + fn minimum_capacity(&self) -> usize { + self.minimum_capacity_var().load(Relaxed) & (!RESIZING) + } /// Returns the maximum capacity. /// /// The maximum capacity must be a power of `2`. fn maximum_capacity(&self) -> usize { - 1_usize << (usize::BITS - 1) + MAXIMUM_CAPACITY_LIMIT } /// Reserves the specified capacity. @@ -72,12 +92,14 @@ where /// Returns the actually allocated capacity. Return `0` if the sum of the current minimum /// capacity and the additional capacity exceeds [`Self::maximum_capacity`]. fn reserve_capacity(&self, additional_capacity: usize) -> usize { - let mut current_minimum_capacity = self.minimum_capacity().load(Relaxed); + let mut current_minimum_capacity = self.minimum_capacity_var().load(Relaxed); loop { - if additional_capacity > self.maximum_capacity() - current_minimum_capacity { + if additional_capacity + > self.maximum_capacity() - (current_minimum_capacity & (!RESIZING)) + { return 0; } - match self.minimum_capacity().compare_exchange_weak( + match self.minimum_capacity_var().compare_exchange_weak( current_minimum_capacity, additional_capacity + current_minimum_capacity, Relaxed, @@ -85,9 +107,8 @@ where ) { Ok(_) => { let guard = Guard::new(); - if let Some(current_array) = self.bucket_array().load(Acquire, &guard).as_ref() - { - if !current_array.has_old_array() { + if let Some(current_array) = self.bucket_array(&guard) { + if !current_array.has_linked_array() { self.try_resize(current_array, 0, &guard); } } @@ -103,7 +124,7 @@ where /// Allocates a new one if no bucket array has been allocated. #[inline] fn get_or_create_bucket_array<'g>(&self, guard: &'g Guard) -> &'g BucketArray { - if let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { + if let Some(current_array) = self.bucket_array(guard) { current_array } else { self.allocate_bucket_array(guard) @@ -113,16 +134,16 @@ where /// Allocates a new bucket array. fn allocate_bucket_array<'g>(&self, guard: &'g Guard) -> &'g BucketArray { unsafe { - let capacity = self.minimum_capacity().load(Relaxed); + let capacity = self.minimum_capacity(); let allocated = Shared::new_unchecked(BucketArray::new(capacity, AtomicShared::null())); - match self.bucket_array().compare_exchange( + match self.bucket_array_var().compare_exchange( Ptr::null(), (Some(allocated), Tag::None), AcqRel, Acquire, guard, ) { - Ok((_, ptr)) | Err((_, ptr)) => ptr.as_ref().unwrap_unchecked(), + Ok((_, ptr)) | Err((_, ptr)) => ptr.as_ref_unchecked().unwrap_unchecked(), } } } @@ -130,7 +151,7 @@ where /// Returns the number of entry slots. #[inline] fn num_slots(&self, guard: &Guard) -> usize { - if let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { + if let Some(current_array) = self.bucket_array(guard) { current_array.num_slots() } else { 0 @@ -142,9 +163,8 @@ where /// In case there are more than `usize::MAX` entries, it returns `usize::MAX`. fn num_entries(&self, guard: &Guard) -> usize { let mut num_entries: usize = 0; - if let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { - let old_array_ptr = current_array.old_array(guard); - if let Some(old_array) = old_array_ptr.as_ref() { + if let Some(current_array) = self.bucket_array(guard) { + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); for i in 0..old_array.len() { num_entries = num_entries.saturating_add(old_array.bucket(i).len()); @@ -153,10 +173,7 @@ where for i in 0..current_array.len() { num_entries = num_entries.saturating_add(current_array.bucket(i).len()); } - if old_array_ptr.is_null() - && num_entries == 0 - && self.minimum_capacity().load(Relaxed) == 0 - { + if num_entries == 0 && self.minimum_capacity() == 0 { self.try_resize(current_array, 0, guard); } } @@ -165,9 +182,8 @@ where /// Returns `true` if a valid entry is found. fn has_entry(&self, guard: &Guard) -> bool { - if let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { - let old_array_ptr = current_array.old_array(guard); - if let Some(old_array) = old_array_ptr.as_ref() { + if let Some(current_array) = self.bucket_array(guard) { + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); for i in 0..old_array.len() { if old_array.bucket(i).len() != 0 { @@ -180,7 +196,7 @@ where return true; } } - if old_array_ptr.is_null() && self.minimum_capacity().load(Relaxed) == 0 { + if self.minimum_capacity() == 0 { self.try_resize(current_array, 0, guard); } } @@ -234,11 +250,11 @@ where { debug_assert_eq!(TYPE, INDEX); - let mut current_array_ptr = self.bucket_array().load(Acquire, guard); - while let Some(current_array) = current_array_ptr.as_ref() { - if let Some(old_array) = current_array.old_array(guard).as_ref() { + let mut current_array_ptr = self.bucket_array_var().load(Acquire, guard); + while let Some(current_array) = unsafe { current_array_ptr.as_ref_unchecked() } { + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); - let index = old_array.calculate_bucket_index(hash); + let index = old_array.bucket_index(hash); if let Some(entry) = old_array.bucket(index).search_entry( old_array.data_block(index), key, @@ -249,7 +265,7 @@ where } } - let index = current_array.calculate_bucket_index(hash); + let index = current_array.bucket_index(hash); if let Some(entry) = current_array.bucket(index).search_entry( current_array.data_block(index), key, @@ -259,7 +275,7 @@ where return Some(entry); } - let new_current_array_ptr = self.bucket_array().load(Acquire, guard); + let new_current_array_ptr = self.bucket_array_var().load(Acquire, guard); if current_array_ptr == new_current_array_ptr { break; } @@ -276,28 +292,33 @@ where key: &Q, hash: u64, f: F, - async_guard: &AsyncGuard, ) -> Option where Q: Equivalent + Hash + ?Sized, { - while let Some(current_array) = async_guard.load(self.bucket_array(), Acquire) { - let index = current_array.calculate_bucket_index(hash); - if current_array.has_old_array() { - self.incremental_rehash_async(current_array, async_guard) + let async_guard = pin!(AsyncGuard::default()); + while let Some(current_array) = async_guard.load_unchecked(self.bucket_array_var(), Acquire) + { + if current_array.has_linked_array() { + self.incremental_rehash_async(current_array, &async_guard) .await; if !self - .dedup_bucket_async(current_array, index, async_guard) + .dedup_bucket_async( + current_array, + current_array.bucket_index(hash), + &async_guard, + ) .await { continue; } } - let bucket = current_array.bucket(index); + let bucket_index = current_array.bucket_index(hash); + let bucket = current_array.bucket(bucket_index); if let Some(reader) = Reader::try_lock(bucket) { if let Some(entry) = reader.search_entry( - current_array.data_block(index), + current_array.data_block(bucket_index), key, hash, async_guard.guard(), @@ -305,9 +326,9 @@ where return Some(f(&entry.0, &entry.1)); } break; - } else if let Some(reader) = Reader::lock_async(bucket, async_guard).await { + } else if let Some(reader) = Reader::lock_async(bucket, &async_guard).await { if let Some(entry) = reader.search_entry( - current_array.data_block(index), + current_array.data_block(bucket_index), key, hash, async_guard.guard(), @@ -333,9 +354,9 @@ where where Q: Equivalent + Hash + ?Sized, { - while let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { - let index = current_array.calculate_bucket_index(hash); - if let Some(old_array) = current_array.old_array(guard).as_ref() { + while let Some(current_array) = self.bucket_array(guard) { + let index = current_array.bucket_index(hash); + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); self.dedup_bucket_sync::(current_array, old_array, index, guard); } @@ -357,29 +378,29 @@ where /// /// If the container is empty, a new bucket array is allocated. #[inline] - async fn writer_async( - &self, - hash: u64, - async_guard: &AsyncGuard, - ) -> LockedBucket { + async fn writer_async(&self, hash: u64) -> LockedBucket { + let async_guard = pin!(AsyncGuard::default()); if let Some(locked_bucket) = self.try_optional_writer::(hash, async_guard.guard()) { return locked_bucket; } - loop { let current_array = self.get_or_create_bucket_array(async_guard.guard()); - let bucket_index = current_array.calculate_bucket_index(hash); - if current_array.has_old_array() { - self.incremental_rehash_async(current_array, async_guard) + if current_array.has_linked_array() { + self.incremental_rehash_async(current_array, &async_guard) .await; if !self - .dedup_bucket_async(current_array, bucket_index, async_guard) + .dedup_bucket_async( + current_array, + current_array.bucket_index(hash), + &async_guard, + ) .await { continue; } } + let bucket_index = current_array.bucket_index(hash); let bucket = current_array.bucket(bucket_index); if (TYPE != CACHE || current_array.num_slots() < self.maximum_capacity()) && bucket.len() >= BUCKET_LEN - 1 @@ -392,8 +413,7 @@ where async_guard.guard(), ); } - - if let Some(writer) = Writer::lock_async(bucket, async_guard).await { + if let Some(writer) = Writer::lock_async(bucket, &async_guard).await { return LockedBucket { writer, data_block: current_array.data_block(bucket_index), @@ -415,8 +435,8 @@ where loop { let current_array = self.get_or_create_bucket_array(guard); - let bucket_index = current_array.calculate_bucket_index(hash); - if let Some(old_array) = current_array.old_array(guard).as_ref() { + let bucket_index = current_array.bucket_index(hash); + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); self.dedup_bucket_sync::(current_array, old_array, bucket_index, guard); } @@ -444,30 +464,31 @@ where /// /// If the container is empty, `None` is returned. #[inline] - async fn optional_writer_async( - &self, - hash: u64, - async_guard: &AsyncGuard, - ) -> Option> { + async fn optional_writer_async(&self, hash: u64) -> Option> { + let async_guard = pin!(AsyncGuard::default()); if let Some(locked_bucket) = self.try_optional_writer::(hash, async_guard.guard()) { return Some(locked_bucket); } - - while let Some(current_array) = async_guard.load(self.bucket_array(), Acquire) { - let bucket_index = current_array.calculate_bucket_index(hash); - if current_array.has_old_array() { - self.incremental_rehash_async(current_array, async_guard) + while let Some(current_array) = async_guard.load_unchecked(self.bucket_array_var(), Acquire) + { + if current_array.has_linked_array() { + self.incremental_rehash_async(current_array, &async_guard) .await; if !self - .dedup_bucket_async(current_array, bucket_index, async_guard) + .dedup_bucket_async( + current_array, + current_array.bucket_index(hash), + &async_guard, + ) .await { continue; } } + let bucket_index = current_array.bucket_index(hash); let bucket = current_array.bucket(bucket_index); - if let Some(writer) = Writer::lock_async(bucket, async_guard).await { + if let Some(writer) = Writer::lock_async(bucket, &async_guard).await { return Some(LockedBucket { writer, data_block: current_array.data_block(bucket_index), @@ -492,9 +513,9 @@ where return Some(locked_bucket); } - while let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { - let bucket_index = current_array.calculate_bucket_index(hash); - if let Some(old_array) = current_array.old_array(guard).as_ref() { + while let Some(current_array) = self.bucket_array(guard) { + let bucket_index = current_array.bucket_index(hash); + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); self.dedup_bucket_sync::(current_array, old_array, bucket_index, guard); } @@ -519,11 +540,11 @@ where hash: u64, guard: &Guard, ) -> Option> { - if let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { - if current_array.has_old_array() { + if let Some(current_array) = self.bucket_array(guard) { + if current_array.has_linked_array() { return None; } - let bucket_index = current_array.calculate_bucket_index(hash); + let bucket_index = current_array.bucket_index(hash); let bucket = current_array.bucket(bucket_index); if CHECK_SIZE && bucket.len() >= BUCKET_LEN { return None; @@ -544,13 +565,15 @@ where /// /// This method stops iterating when the closure returns `false`. #[inline] - async fn for_each_reader_async(&self, async_guard: &AsyncGuard, mut f: F) + async fn for_each_reader_async(&self, mut f: F) where F: FnMut(Reader, NonNull>) -> bool, { + let async_guard = pin!(AsyncGuard::default()); let mut start_index = 0; let mut prev_len = 0; - while let Some(current_array) = async_guard.load(self.bucket_array(), Acquire) { + while let Some(current_array) = async_guard.load_unchecked(self.bucket_array_var(), Acquire) + { // In case the method is repeating the routine, iterate over entries from the middle of // the array. start_index = if prev_len == 0 || prev_len == current_array.len() { @@ -561,12 +584,11 @@ where prev_len = current_array.len(); while start_index < current_array.len() { - let index = start_index; - if current_array.has_old_array() { - self.incremental_rehash_async(current_array, async_guard) + if current_array.has_linked_array() { + self.incremental_rehash_async(current_array, &async_guard) .await; if !self - .dedup_bucket_async(current_array, index, async_guard) + .dedup_bucket_async(current_array, start_index, &async_guard) .await { // Retry the operation since there is a possibility that the current bucket @@ -575,13 +597,13 @@ where } } - let bucket = current_array.bucket(index); - if let Some(reader) = Reader::lock_async(bucket, async_guard).await { - if !async_guard.check_ref(self.bucket_array(), current_array, Acquire) { + let bucket = current_array.bucket(start_index); + if let Some(reader) = Reader::lock_async(bucket, &async_guard).await { + if !async_guard.check_ref(self.bucket_array_var(), current_array, Acquire) { // `current_array` is no longer the current one. break; } - let data_block = current_array.data_block(index); + let data_block = current_array.data_block(start_index); if !f(reader, data_block) { return; } @@ -609,7 +631,7 @@ where { let mut start_index = 0; let mut prev_len = 0; - while let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { + while let Some(current_array) = self.bucket_array(guard) { // In case the method is repeating the routine, iterate over entries from the middle of // the array. start_index = if prev_len == 0 || prev_len == current_array.len() { @@ -621,7 +643,7 @@ where while start_index < current_array.len() { let index = start_index; - if let Some(old_array) = current_array.old_array(guard).as_ref() { + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); self.dedup_bucket_sync::(current_array, old_array, index, guard); } @@ -653,14 +675,15 @@ where &self, mut start_index: usize, expected_array_len: usize, - async_guard: &AsyncGuard, mut f: F, ) where F: FnMut(LockedBucket, &mut bool) -> bool, { + let async_guard = pin!(AsyncGuard::default()); let mut prev_len = expected_array_len; let mut removed = false; - while let Some(current_array) = async_guard.load(self.bucket_array(), Acquire) { + while let Some(current_array) = async_guard.load_unchecked(self.bucket_array_var(), Acquire) + { // In case the method is repeating the routine, iterate over entries from the middle of // the array. let current_array_len = current_array.len(); @@ -673,11 +696,11 @@ where while start_index < current_array_len { let bucket_index = start_index; - if current_array.has_old_array() { - self.incremental_rehash_async(current_array, async_guard) + if current_array.has_linked_array() { + self.incremental_rehash_async(current_array, &async_guard) .await; if !self - .dedup_bucket_async(current_array, bucket_index, async_guard) + .dedup_bucket_async(current_array, bucket_index, &async_guard) .await { // Retry the operation since there is a possibility that the current bucket @@ -687,8 +710,8 @@ where } let bucket = current_array.bucket(bucket_index); - if let Some(writer) = Writer::lock_async(bucket, async_guard).await { - if !async_guard.check_ref(self.bucket_array(), current_array, Acquire) { + if let Some(writer) = Writer::lock_async(bucket, &async_guard).await { + if !async_guard.check_ref(self.bucket_array_var(), current_array, Acquire) { // `current_array` is no longer the current one. break; } @@ -717,11 +740,7 @@ where } if removed { - if let Some(current_array) = self - .bucket_array() - .load(Acquire, async_guard.guard()) - .as_ref() - { + if let Some(current_array) = self.bucket_array(async_guard.guard()) { self.try_shrink_or_rebuild(current_array, 0, async_guard.guard()); } } @@ -742,7 +761,7 @@ where { let mut prev_len = expected_array_len; let mut removed = false; - while let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { + while let Some(current_array) = self.bucket_array(guard) { // In case the method is repeating the routine, iterate over entries from the middle of // the array. let current_array_len = current_array.len(); @@ -755,7 +774,7 @@ where while start_index < current_array_len { let bucket_index = start_index; - if let Some(old_array) = current_array.old_array(guard).as_ref() { + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); self.dedup_bucket_sync::(current_array, old_array, bucket_index, guard); } @@ -787,7 +806,7 @@ where } if removed { - if let Some(current_array) = self.bucket_array().load(Acquire, guard).as_ref() { + if let Some(current_array) = self.bucket_array(guard) { self.try_shrink_or_rebuild(current_array, 0, guard); } } @@ -798,8 +817,8 @@ where fn try_reserve_bucket(&self, hash: u64, guard: &Guard) -> Option> { loop { let current_array = self.get_or_create_bucket_array(guard); - let bucket_index = current_array.calculate_bucket_index(hash); - if let Some(old_array) = current_array.old_array(guard).as_ref() { + let bucket_index = current_array.bucket_index(hash); + if let Some(old_array) = current_array.linked_array(guard) { self.incremental_rehash_sync::(current_array, guard); if !self.dedup_bucket_sync::(current_array, old_array, bucket_index, guard) { return None; @@ -846,12 +865,14 @@ where index: usize, async_guard: &'g AsyncGuard, ) -> bool { - if !async_guard.check_ref(self.bucket_array(), current_array, Acquire) { + if !async_guard.check_ref(self.bucket_array_var(), current_array, Acquire) { // A new bucket array was created in the meantime. return false; } - if let Some(old_array) = async_guard.load(current_array.bucket_link(), Acquire) { + if let Some(old_array) = + async_guard.load_unchecked(current_array.linked_array_var(), Acquire) + { let range = from_index_to_range(current_array.len(), old_array.len(), index); for old_index in range.0..range.1 { let bucket = old_array.bucket(old_index); @@ -873,7 +894,7 @@ where } // The old bucket array was removed, no point in trying to move entries from it. - if !current_array.has_old_array() { + if !current_array.has_linked_array() { break; } } @@ -949,7 +970,7 @@ where // It may seem inefficient to reevaluate the same values, but it is beneficial for reducing // the `Future` size. - let Some(old_array) = current_array.old_array(async_guard.guard()).as_ref() else { + let Some(old_array) = current_array.linked_array(async_guard.guard()) else { return; }; let (target_index, end_target_index) = @@ -1066,7 +1087,7 @@ where (0, u64::from(entry_ptr.partial_hash(&**old_writer))) } else { let hash = self.hash(&entry_ptr.get(old_data_block).0); - let new_index = current_array.calculate_bucket_index(hash); + let new_index = current_array.bucket_index(hash); debug_assert!(new_index - target_index < (current_array.len() / old_array.len())); (new_index - target_index, hash) }; @@ -1123,7 +1144,7 @@ where let index = if old_array.len() >= current_array.len() { target_index } else { - current_array.calculate_bucket_index(hash) + current_array.bucket_index(hash) }; current_array.bucket(index).extract_from( current_array.data_block(index), @@ -1206,7 +1227,9 @@ where current_array: &'g BucketArray, async_guard: &'g AsyncGuard, ) { - if let Some(old_array) = async_guard.load(current_array.bucket_link(), Acquire) { + if let Some(old_array) = + async_guard.load_unchecked(current_array.linked_array_var(), Acquire) + { if let Some(current) = Self::start_incremental_rehash(old_array) { let rehashing_guard = ExitGuard::new((old_array, current), |(old_array, prev)| { Self::end_incremental_rehash(old_array, prev, false); @@ -1227,12 +1250,12 @@ where ) .await; } - debug_assert!(current_array.has_old_array()); + debug_assert!(current_array.has_linked_array()); } if Self::end_incremental_rehash(rehashing_guard.0, rehashing_guard.1, true) { if let Some(bucket_array) = current_array - .bucket_link() + .linked_array_var() .swap((None, Tag::None), Release) .0 { @@ -1252,7 +1275,7 @@ where current_array: &'g BucketArray, guard: &'g Guard, ) { - if let Some(old_array) = current_array.old_array(guard).as_ref() { + if let Some(old_array) = current_array.linked_array(guard) { if let Some(current) = Self::start_incremental_rehash(old_array) { let rehashing_guard = ExitGuard::new((old_array, current), |(old_array, prev)| { Self::end_incremental_rehash(old_array, prev, false); @@ -1285,7 +1308,7 @@ where if Self::end_incremental_rehash(rehashing_guard.0, rehashing_guard.1, true) { if let Some(bucket_array) = current_array - .bucket_link() + .linked_array_var() .swap((None, Tag::None), Release) .0 { @@ -1305,7 +1328,7 @@ where mut num_entries: usize, guard: &Guard, ) { - if !current_array.has_old_array() { + if !current_array.has_linked_array() { // Try to grow if the estimated load factor is greater than `25/32`. let threshold = current_array.sample_size() * (BUCKET_LEN / 32) * 25; if num_entries > threshold @@ -1329,8 +1352,8 @@ where index: usize, guard: &Guard, ) { - if !current_array.has_old_array() { - let minimum_capacity = self.minimum_capacity().load(Relaxed); + if !current_array.has_linked_array() { + let minimum_capacity = self.minimum_capacity(); if TYPE == INDEX || current_array.num_slots() > minimum_capacity { // Try to shrink if the estimated load factor is less than `1/8`. let shrink_threshold = current_array.sample_size() * BUCKET_LEN / 8; @@ -1370,21 +1393,20 @@ where sampling_index: usize, guard: &Guard, ) { - let current_array_ptr = self.bucket_array().load(Acquire, guard); - if current_array_ptr.tag() != Tag::None { - // Another thread is currently allocating a new bucket array. - return; - } - let Some(current_array) = current_array_ptr.as_ref() else { + let current_array_ptr = self.bucket_array_var().load(Acquire, guard); + let Some(current_array) = (unsafe { current_array_ptr.as_ref_unchecked() }) else { // The hash table is empty. return; }; if !ptr::eq(current_array, sampled_array) { // The preliminary sampling result cannot be trusted anymore. return; + } else if current_array.has_linked_array() { + // Cannot resize with a bucket array linked to the current bucket array. + return; } - let minimum_capacity = self.minimum_capacity().load(Relaxed); + let minimum_capacity = self.minimum_capacity(); let capacity = current_array.num_slots(); let estimated_num_entries = Self::sample(current_array, sampling_index); @@ -1426,18 +1448,27 @@ where return; } - // Mark that the thread may allocate a new array to prevent multiple threads from - // allocating bucket arrays at the same time. - if !self.bucket_array().update_tag_if( - Tag::First, - |ptr| ptr == current_array_ptr, - AcqRel, - Relaxed, - ) { + if self + .minimum_capacity_var() + .fetch_update(AcqRel, Acquire, |lock_state| { + if lock_state >= RESIZING { + None + } else { + Some(lock_state + RESIZING) + } + }) + .is_err() + { // The bucket array is being replaced with a new one. return; } - debug_assert!(!current_array.has_old_array()); + let _lock_guard = ExitGuard::new((), |()| { + self.minimum_capacity_var().fetch_sub(RESIZING, Release); + }); + + if self.bucket_array_var().load(Acquire, guard) != current_array_ptr { + return; + } if try_drop_table { // Try to drop the hash table with all the buckets locked. @@ -1464,32 +1495,21 @@ where }) { // All the buckets are empty and locked. writer_guard.1 = true; - if let Some(bucket_array) = self.bucket_array().swap((None, Tag::None), Release).0 { + if let Some(bucket_array) = + self.bucket_array_var().swap((None, Tag::None), Release).0 + { self.defer_reclaim(bucket_array, guard); } - return; } - } - - let allocated_array: Option>> = None; - let mut mutex_guard = ExitGuard::new(allocated_array, |allocated_array| { - if let Some(allocated_array) = allocated_array { - // A new array was allocated. - self.bucket_array() - .swap((Some(allocated_array), Tag::None), Release); - } else { - // Release the lock. - self.bucket_array() - .update_tag_if(Tag::None, |_| true, Release, Relaxed); - } - }); - if try_resize || try_rebuild { - mutex_guard.replace(unsafe { + } else if try_resize || try_rebuild { + let new_bucket_array = unsafe { Shared::new_unchecked(BucketArray::::new( new_capacity, - (*self.bucket_array()).clone(Relaxed, guard), + (*self.bucket_array_var()).clone(Relaxed, guard), )) - }); + }; + self.bucket_array_var() + .swap((Some(new_bucket_array), Tag::None), Release); } } @@ -1513,6 +1533,12 @@ where } } +/// Hard limit of the maximum capacity of each container type. +pub(super) const MAXIMUM_CAPACITY_LIMIT: usize = 1_usize << (usize::BITS - 2); + +/// Denotes a state where a thread is resizing the container. +pub(super) const RESIZING: usize = 1_usize << (usize::BITS - 1); + /// [`LockedBucket`] has exclusive access to a [`Bucket`]. #[derive(Debug)] pub(crate) struct LockedBucket { @@ -1628,7 +1654,7 @@ impl LockedBucket H: BuildHasher, { if (TYPE == INDEX && self.writer.need_rebuild()) || self.writer.len() == 0 { - if let Some(current_array) = hash_table.bucket_array().load(Acquire, guard).as_ref() { + if let Some(current_array) = hash_table.bucket_array(guard) { if ptr::eq(current_array, self.bucket_array()) { let bucket_index = self.bucket_index; drop(self); @@ -1646,13 +1672,11 @@ impl LockedBucket self, hash_table: &'h T, entry_ptr: &mut EntryPtr<'h, K, V, TYPE>, - async_guard: &mut Pin<&mut AsyncGuard>, ) -> Option> where H: BuildHasher, { - let prolonged_guard = hash_table.prolonged_guard_ref(async_guard.guard()); - if entry_ptr.move_to_next(&self.writer, prolonged_guard) { + if entry_ptr.move_to_next(&self.writer, fake_guard()) { return Some(self); } @@ -1660,7 +1684,7 @@ impl LockedBucket let len = self.bucket_array().len(); if self.writer.len() == 0 { - self.try_shrink_or_rebuild(hash_table, prolonged_guard); + self.try_shrink_or_rebuild(hash_table, &Guard::new()); } else { drop(self); } @@ -1671,10 +1695,10 @@ impl LockedBucket let mut next_entry = None; hash_table - .for_each_writer_async(next_index, len, async_guard, |locked_bucket, _| { - let guard = hash_table.prolonged_guard_ref(async_guard.guard()); - *entry_ptr = EntryPtr::new(guard); - if entry_ptr.move_to_next(&locked_bucket.writer, guard) { + .for_each_writer_async(next_index, len, |locked_bucket, _| { + let fake_guard = fake_guard(); + *entry_ptr = EntryPtr::new(fake_guard); + if entry_ptr.move_to_next(&locked_bucket.writer, fake_guard) { next_entry = Some(locked_bucket); return true; } diff --git a/src/hash_table/bucket.rs b/src/hash_table/bucket.rs index afe3d6c..e5770b7 100644 --- a/src/hash_table/bucket.rs +++ b/src/hash_table/bucket.rs @@ -201,7 +201,7 @@ impl Bucket { self.len.store(self.len.load(Relaxed) - 1, Relaxed); - if let Some(link) = entry_ptr.current_link_ptr.as_ref() { + if let Some(link) = link_ref(entry_ptr.current_link_ptr) { let mut occupied_bitmap = link.metadata.occupied_bitmap.load(Relaxed); debug_assert_ne!(occupied_bitmap & (1_u32 << entry_ptr.current_index), 0); @@ -243,7 +243,7 @@ impl Bucket { self.len.store(self.len.load(Relaxed) - 1, Relaxed); - if let Some(link) = entry_ptr.current_link_ptr.as_ref() { + if let Some(link) = link_ref(entry_ptr.current_link_ptr) { let mut removed_bitmap = link.metadata.removed_bitmap_or_lru_tail.load(Relaxed); debug_assert_eq!(removed_bitmap & (1_u32 << entry_ptr.current_index), 0); @@ -327,7 +327,7 @@ impl Bucket { BUCKET_LEN - self.metadata.occupied_bitmap.load(Relaxed).count_ones() as usize; if capacity < additional { let mut link_ptr = self.metadata.link.load(Acquire, guard); - while let Some(link) = link_ptr.as_ref() { + while let Some(link) = link_ref(link_ptr) { capacity += LINKED_BUCKET_LEN - link.metadata.occupied_bitmap.load(Relaxed).count_ones() as usize; if capacity >= additional { @@ -362,7 +362,7 @@ impl Bucket { ) { debug_assert!(self.rw_lock.is_locked(Relaxed)); - let entry = if let Some(link) = from_entry_ptr.current_link_ptr.as_ref() { + let entry = if let Some(link) = link_ref(from_entry_ptr.current_link_ptr) { Self::read_data_block(&link.data_block, from_entry_ptr.current_index) } else { Self::read_data_block( @@ -373,7 +373,7 @@ impl Bucket { self.insert(data_block, hash, entry, guard); let mo = if TYPE == INDEX { Release } else { Relaxed }; - if let Some(link) = from_entry_ptr.current_link_ptr.as_ref() { + if let Some(link) = link_ref(from_entry_ptr.current_link_ptr) { let occupied_bitmap = link.metadata.occupied_bitmap.load(Relaxed); debug_assert_ne!(occupied_bitmap & (1_u32 << from_entry_ptr.current_index), 0); @@ -444,7 +444,7 @@ impl Bucket { guard: &'g Guard, ) -> EntryPtr<'g, K, V, TYPE> { let mut link_ptr = self.metadata.link.load(Acquire, guard); - while let Some(link) = link_ptr.as_ref() { + while let Some(link) = link_ref(link_ptr) { let occupied_bitmap = link.metadata.occupied_bitmap.load(Relaxed); let free_index = occupied_bitmap.trailing_ones() as usize; if free_index != LINKED_BUCKET_LEN { @@ -476,7 +476,7 @@ impl Bucket { *h = Self::partial_hash(hash); }); link.metadata.occupied_bitmap.store(1, Relaxed); - if let Some(head) = link.metadata.link.load(Relaxed, guard).as_ref() { + if let Some(head) = link_ref(link.metadata.link.load(Relaxed, guard)) { head.prev_link.store(link.as_ptr().cast_mut(), Relaxed); } let link_ptr = link.get_guarded_ptr(guard); @@ -499,7 +499,7 @@ impl Bucket { let mut empty = Self::cleanup_removed_entries(&self.metadata, unsafe { data_block.as_ref() }); let mut link_ptr = self.metadata.link.load(Acquire, guard); - while let Some(link) = link_ptr.as_ref() { + while let Some(link) = link_ref(link_ptr) { empty &= Self::cleanup_removed_entries(&link.metadata, &link.data_block); let next_link_ptr = link.metadata.link.load(Acquire, guard); if next_link_ptr.is_null() { @@ -705,7 +705,7 @@ impl Bucket { } let mut link_ptr = self.metadata.link.load(Acquire, guard); - while let Some(link) = link_ptr.as_ref() { + while let Some(link) = link_ref(link_ptr) { if let Some((entry, _)) = Self::search_data_block(&link.metadata, &link.data_block, key, hash) { @@ -743,7 +743,7 @@ impl Bucket { } let mut current_link_ptr = self.metadata.link.load(Acquire, guard); - while let Some(link) = current_link_ptr.as_ref() { + while let Some(link) = link_ref(current_link_ptr) { if let Some((_, index)) = Self::search_data_block(&link.metadata, &link.data_block, key, hash) { @@ -1015,7 +1015,7 @@ impl<'g, K, V, const TYPE: char> EntryPtr<'g, K, V, TYPE> { { return true; } - while let Some(link) = self.current_link_ptr.as_ref() { + while let Some(link) = link_ref(self.current_link_ptr) { if self.next_entry::(&link.metadata, guard) { return true; } @@ -1035,7 +1035,7 @@ impl<'g, K, V, const TYPE: char> EntryPtr<'g, K, V, TYPE> { pub(crate) fn get(&self, data_block: NonNull>) -> &'g (K, V) { debug_assert_ne!(self.current_index, usize::MAX); - let entry_ptr = if let Some(link) = self.current_link_ptr.as_ref() { + let entry_ptr = if let Some(link) = link_ref(self.current_link_ptr) { Bucket::::entry_ptr(&link.data_block, self.current_index) } else { Bucket::::entry_ptr(unsafe { data_block.as_ref() }, self.current_index) @@ -1054,7 +1054,7 @@ impl<'g, K, V, const TYPE: char> EntryPtr<'g, K, V, TYPE> { ) -> &mut (K, V) { debug_assert_ne!(self.current_index, usize::MAX); - let entry_ptr = if let Some(link) = self.current_link_ptr.as_ref() { + let entry_ptr = if let Some(link) = link_ref(self.current_link_ptr) { Bucket::::entry_mut_ptr(&link.data_block, self.current_index) } else { Bucket::::entry_mut_ptr( @@ -1072,7 +1072,7 @@ impl<'g, K, V, const TYPE: char> EntryPtr<'g, K, V, TYPE> { pub(crate) fn partial_hash(&self, bucket: &Bucket) -> u8 { debug_assert_ne!(self.current_index, usize::MAX); - if let Some(link) = self.current_link_ptr.as_ref() { + if let Some(link) = link_ref(self.current_link_ptr) { *Bucket::::read_cell( &link.metadata.partial_hash_array[self.current_index], ) @@ -1379,6 +1379,11 @@ impl Drop for LinkedBucket { } } +/// Returns a reference to the linked bucket that the pointer might point to. +fn link_ref(ptr: Ptr<'_, LinkedBucket>) -> Option<&LinkedBucket> { + unsafe { ptr.as_ref_unchecked() } +} + #[cfg(not(feature = "loom"))] #[cfg(test)] mod test { diff --git a/src/hash_table/bucket_array.rs b/src/hash_table/bucket_array.rs index ba5edae..e428487 100644 --- a/src/hash_table/bucket_array.rs +++ b/src/hash_table/bucket_array.rs @@ -5,7 +5,7 @@ use std::ptr::NonNull; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{Acquire, Relaxed}; -use sdd::{AtomicShared, Guard, Ptr, Tag}; +use sdd::{AtomicShared, Guard, Tag}; use super::bucket::{BUCKET_LEN, Bucket, DataBlock, INDEX, LruList}; use crate::exit_guard::ExitGuard; @@ -18,7 +18,7 @@ pub struct BucketArray { hash_offset: u8, sample_size: u8, bucket_ptr_offset: u16, - old_array: AtomicShared>, + linked_array: AtomicShared>, num_cleared_buckets: AtomicUsize, } @@ -28,7 +28,7 @@ impl BucketArray { /// `capacity` is the desired number of entries, not the length of the bucket array. pub(crate) fn new( capacity: usize, - old_array: AtomicShared>, + linked_array: AtomicShared>, ) -> Self { let adjusted_capacity = capacity .min(1_usize << (usize::BITS - 1)) @@ -90,7 +90,7 @@ impl BucketArray { hash_offset: u8::try_from(u64::BITS).unwrap_or(64) - log2_array_len, sample_size, bucket_ptr_offset: bucket_array_ptr_offset, - old_array, + linked_array, num_cleared_buckets: AtomicUsize::new(0), } } @@ -111,7 +111,7 @@ impl BucketArray { /// Calculates the [`Bucket`] index for the hash value. #[allow(clippy::cast_possible_truncation)] // Intended truncation. #[inline] - pub(crate) const fn calculate_bucket_index(&self, hash: u64) -> usize { + pub(crate) const fn bucket_index(&self, hash: u64) -> usize { // Take the upper n-bits to make sure that a single bucket is spread across a few adjacent // buckets when the hash table is resized. (hash >> self.hash_offset) as usize @@ -157,22 +157,25 @@ impl BucketArray { unsafe { self.data_blocks.add(index) } } - /// Returns a reference to the old array pointer. + /// Returns `true` if an linked bucket array exists. #[inline] - pub(crate) const fn bucket_link(&self) -> &AtomicShared> { - &self.old_array + pub(crate) fn has_linked_array(&self) -> bool { + !self.linked_array.is_null(Acquire) } - /// Returns `true` if the old array exists. + /// Returns a reference to the linked bucket array pointer. #[inline] - pub(crate) fn has_old_array(&self) -> bool { - !self.old_array.is_null(Acquire) + pub(crate) const fn linked_array_var(&self) -> &AtomicShared> { + &self.linked_array } - /// Returns a [`Ptr`] to the old array. + /// Returns a reference to the linked bucket array. #[inline] - pub(crate) fn old_array<'g>(&self, guard: &'g Guard) -> Ptr<'g, BucketArray> { - self.old_array.load(Acquire, guard) + pub(crate) fn linked_array<'g>( + &self, + guard: &'g Guard, + ) -> Option<&'g BucketArray> { + unsafe { self.linked_array.load(Acquire, guard).as_ref_unchecked() } } /// Calculates the layout of the memory block for an array of `T`. @@ -187,9 +190,9 @@ impl BucketArray { impl Drop for BucketArray { fn drop(&mut self) { - if !self.old_array.is_null(Relaxed) { + if !self.linked_array.is_null(Relaxed) { unsafe { - self.old_array + self.linked_array .swap((None, Tag::None), Relaxed) .0 .map(|a| a.drop_in_place()); diff --git a/src/tests/models.rs b/src/tests/models.rs index 89fac82..751aadc 100644 --- a/src/tests/models.rs +++ b/src/tests/models.rs @@ -127,7 +127,7 @@ fn hashmap_key_uniqueness() { assert!(thread_insert.join().is_ok()); for k in 0..max_key { - assert!(hashmap.read_sync(&k, |_, _| ()).is_some()); + assert!(hashmap.read_sync(&k, |_, _| ()).is_some(), "{k} {max_key}"); } assert!(hashmap.read_sync(&usize::MAX, |_, _| ()).is_some()); assert_eq!(hashmap.len(), max_key + 1); diff --git a/src/tests/unit_tests.rs b/src/tests/unit_tests.rs index ea1359d..1846b06 100644 --- a/src/tests/unit_tests.rs +++ b/src/tests/unit_tests.rs @@ -143,18 +143,68 @@ mod hashmap { #[test] fn future_size() { - let limit = 584; - let hashmap: HashMap = HashMap::default(); - let insert_size = size_of_val(&hashmap.insert_async(0, 0)); - assert!(insert_size < limit, "{insert_size}"); - let entry_size = size_of_val(&hashmap.entry_async(0)); - assert!(entry_size < limit, "{entry_size}"); - let read_size = size_of_val(&hashmap.read_async(&0, |_, _| {})); - assert!(read_size < limit, "{read_size}"); - let remove_size = size_of_val(&hashmap.remove_async(&0)); - assert!(remove_size < limit, "{remove_size}"); - let iter_size = size_of_val(&hashmap.iter_async(|_, _| true)); - assert!(iter_size < limit, "{iter_size}"); + // TODO: writer_async = 480B. + // TODO: incremental_rehash_async/dedup_bucket_async = 416B. + // TODO: relocate_bucket_async + // TODO: lock_async = 200B. + let base_size = 504; // In v2, 104. + // Small type. + { + let limit = base_size; // In v2, 104. + let hashmap: HashMap<(), ()> = HashMap::default(); + let get_size = size_of_val(&hashmap.get_async(&())); + assert!(get_size <= limit + 24, "{get_size}"); + let contains_size = size_of_val(&hashmap.contains_async(&())); + assert!(contains_size <= limit + 16, "{contains_size}"); + let insert_size = size_of_val(&hashmap.insert_async((), ())); + assert!(insert_size <= limit, "{insert_size}"); + let entry_size = size_of_val(&hashmap.entry_async(())); + assert!(entry_size <= limit + 8, "{entry_size}"); + let read_size = size_of_val(&hashmap.read_async(&(), |(), ()| {})); + assert!(read_size <= limit + 16, "{read_size}"); + let remove_size = size_of_val(&hashmap.remove_async(&())); + assert!(remove_size <= limit + 48, "{remove_size}"); + let iter_size = size_of_val(&hashmap.iter_async(|(), ()| true)); + assert!(iter_size <= limit + 40, "{iter_size}"); + } + // Medium type. + { + let limit = base_size + 2 * size_of::<(u64, u64)>(); // In v2, 104 + 2 * size_of::<(u64, u64)>. + let hashmap: HashMap = HashMap::default(); + let get_size = size_of_val(&hashmap.get_async(&0)); + assert!(get_size <= limit, "{get_size}"); + let contains_size = size_of_val(&hashmap.contains_async(&0)); + assert!(contains_size <= limit, "{contains_size}"); + let insert_size = size_of_val(&hashmap.insert_async(0, 0)); + assert!(insert_size <= limit, "{insert_size}"); + let entry_size = size_of_val(&hashmap.entry_async(0)); + assert!(entry_size <= limit, "{entry_size}"); + let read_size = size_of_val(&hashmap.read_async(&0, |_, _| {})); + assert!(read_size <= limit, "{read_size}"); + let remove_size = size_of_val(&hashmap.remove_async(&0)); + assert!(remove_size <= limit + 16, "{remove_size}"); + let iter_size = size_of_val(&hashmap.iter_async(|_, _| true)); + assert!(iter_size <= limit + 8, "{iter_size}"); + } + { + type Large = [u64; 32]; + let limit = base_size + 2 * size_of::<(Vec, Large)>(); // In v2, 104 + 2 * size_of::<(Vec, Large)>. + let hashmap: HashMap, Large> = HashMap::default(); + let get_size = size_of_val(&hashmap.get_async(&vec![])); + assert!(get_size <= limit, "{get_size}"); + let contains_size = size_of_val(&hashmap.contains_async(&vec![])); + assert!(contains_size <= limit + 16, "{contains_size}"); + let insert_size = size_of_val(&hashmap.insert_async(vec![], [0; 32])); + assert!(insert_size <= limit, "{insert_size}"); + let entry_size = size_of_val(&hashmap.entry_async(vec![])); + assert!(entry_size <= limit, "{entry_size}"); + let read_size = size_of_val(&hashmap.read_async(&vec![], |_, _| {})); + assert!(read_size <= limit, "{read_size}"); + let remove_size = size_of_val(&hashmap.remove_async(&vec![])); + assert!(remove_size <= limit, "{remove_size}"); + let iter_size = size_of_val(&hashmap.iter_async(|_, _| true)); + assert!(iter_size <= limit, "{iter_size}"); + } } #[cfg_attr(miri, ignore)]