1
//! A sharded concurrent hash table exposing a raw, hash-and-equality-closure
2
//! API.
3
//!
4
//! Unlike a `HashMap<K, V>`, this table does not own or hash a key type: every
5
//! operation is parameterised by a precomputed hash and an equality closure, so
6
//! the stored element can be a compact handle (for example an index) whose hash
7
//! and identity are derived from data living elsewhere. This makes it suitable
8
//! as an interning index that does not duplicate the interned payload.
9
//!
10
//! Concurrency comes from striping the elements across a fixed, power-of-two
11
//! number of shards, each a [`parking_lot::RwLock`] around a
12
//! [`hashbrown::HashTable`]. The shard is selected from the hash bits just below
13
//! the top seven that `hashbrown` reserves for its control byte, so shard
14
//! selection does not degrade in-shard probing.
15
//!
16
//! Two API layers are provided:
17
//!
18
//! * the raw [`ShardedHashMap::find`], [`ShardedHashMap::find_or_insert_with`]
19
//!   and [`ShardedHashMap::remove`], where the caller supplies the hash and the
20
//!   equality (and, for insertion, the rehashing) closures; and
21
//! * `DashSet`-shaped convenience methods ([`ShardedHashMap::insert`],
22
//!   [`ShardedHashMap::get`], [`ShardedHashMap::contains`],
23
//!   [`ShardedHashMap::remove_equiv`]) that hash through the table's own
24
//!   [`BuildHasher`] for callers that store self-describing elements.
25

            
26
use std::hash::BuildHasher;
27
use std::hash::Hash;
28
use std::hash::RandomState;
29
use std::thread::available_parallelism;
30

            
31
use crossbeam_utils::CachePadded;
32
use equivalent::Equivalent;
33
use hashbrown::HashTable;
34
use hashbrown::hash_table::Entry;
35
use parking_lot::RwLock;
36

            
37
/// Caps the default shard count so machines with many cores do not allocate an
38
/// unreasonable number of locks.
39
const MAX_DEFAULT_SHARDS: usize = 256;
40

            
41
/// Number of high hash bits `hashbrown` reserves for its control byte; the shard
42
/// is selected from the bits immediately below them.
43
const CONTROL_BITS: u32 = 7;
44

            
45
/// A concurrent hash table that stores bare elements of type `T`, looked up by a
46
/// caller-supplied hash and equality closure. See the module documentation.
47
pub struct ShardedHashMap<T, S = RandomState> {
48
    /// Element shards, selected by a slice of the hash. Always a power of two in
49
    /// length so the shard index is a cheap bit extraction.
50
    shards: Box<[CachePadded<RwLock<HashTable<T>>>]>,
51
    /// Builds the hashes the convenience methods use; exposed via
52
    /// [`ShardedHashMap::hash`] so raw callers hash consistently with the table.
53
    hasher: S,
54
    /// `64 - log2(shards.len())`; shifts the hash so the shard index ends up in
55
    /// the low bits. Unused when there is a single shard.
56
    shard_shift: u32,
57
}
58

            
59
/// Returns a sensible default shard count for the running machine.
60
4731
fn default_shard_count() -> usize {
61
4731
    let parallelism = available_parallelism().map(|n| n.get()).unwrap_or(1);
62
4731
    (parallelism * 4).clamp(1, MAX_DEFAULT_SHARDS).next_power_of_two()
63
4731
}
64

            
65
impl<T, S: Default> ShardedHashMap<T, S> {
66
    /// Creates a table with a machine-dependent number of shards.
67
    pub fn new() -> ShardedHashMap<T, S> {
68
        ShardedHashMap::with_hasher(S::default())
69
    }
70

            
71
    /// Creates a table with at least `shards` shards (rounded up to a power of
72
    /// two). Fewer shards reduce memory; more reduce write contention.
73
191
    pub fn with_shards(shards: usize) -> ShardedHashMap<T, S> {
74
191
        ShardedHashMap::with_shards_and_hasher(shards, S::default())
75
191
    }
76
}
77

            
78
impl<T, S: Default> Default for ShardedHashMap<T, S> {
79
    fn default() -> ShardedHashMap<T, S> {
80
        ShardedHashMap::with_hasher(S::default())
81
    }
82
}
83

            
84
impl<T, S> ShardedHashMap<T, S> {
85
    /// Creates a table with a machine-dependent number of shards and the given
86
    /// hash builder.
87
246
    pub fn with_hasher(hasher: S) -> ShardedHashMap<T, S> {
88
246
        ShardedHashMap::build(default_shard_count(), 0, hasher)
89
246
    }
90

            
91
    /// Creates a table with at least `shards` shards (rounded up to a power of
92
    /// two) and the given hash builder.
93
20287
    pub fn with_shards_and_hasher(shards: usize, hasher: S) -> ShardedHashMap<T, S> {
94
20287
        ShardedHashMap::build(shards, 0, hasher)
95
20287
    }
96

            
97
    /// Creates a table sized to hold at least `capacity` elements before
98
    /// reallocating, with the given hash builder.
99
    pub fn with_capacity_and_hasher(capacity: usize, hasher: S) -> ShardedHashMap<T, S> {
100
        let shards = default_shard_count();
101
        ShardedHashMap::build(shards, capacity.div_ceil(shards), hasher)
102
    }
103

            
104
    /// Builds a table with `shards` (rounded up to a power of two) shards, each
105
    /// pre-sized for `per_shard_capacity` elements.
106
20533
    fn build(shards: usize, per_shard_capacity: usize, hasher: S) -> ShardedHashMap<T, S> {
107
20533
        let shards = shards.max(1).next_power_of_two();
108
20533
        let shard_shift = u64::BITS - shards.trailing_zeros();
109
20533
        let shards = (0..shards)
110
327474
            .map(|_| CachePadded::new(RwLock::new(HashTable::with_capacity(per_shard_capacity))))
111
20533
            .collect();
112
20533
        ShardedHashMap {
113
20533
            shards,
114
20533
            hasher,
115
20533
            shard_shift,
116
20533
        }
117
20533
    }
118

            
119
    /// Returns the shard a hash maps to.
120
1134000
    fn shard(&self, hash: u64) -> &RwLock<HashTable<T>> {
121
        // With a single shard the shift would be 64 (undefined for `>>`), so
122
        // short-circuit. Otherwise drop the top `CONTROL_BITS` (used by
123
        // `hashbrown`'s control byte) and take the next `log2(shards)` bits.
124
1134000
        let index = if self.shards.len() == 1 {
125
12724
            0
126
        } else {
127
1121276
            ((hash << CONTROL_BITS) >> self.shard_shift) as usize
128
        };
129
1134000
        &self.shards[index]
130
1134000
    }
131

            
132
    /// Returns the number of elements across all shards.
133
100051
    pub fn len(&self) -> usize {
134
1069293
        self.shards.iter().map(|shard| shard.read().len()).sum()
135
100051
    }
136

            
137
    /// Returns true if every shard is empty.
138
    pub fn is_empty(&self) -> bool {
139
        self.shards.iter().all(|shard| shard.read().is_empty())
140
    }
141

            
142
    /// Returns the total capacity across all shards.
143
    pub fn capacity(&self) -> usize {
144
        self.shards.iter().map(|shard| shard.read().capacity()).sum()
145
    }
146

            
147
    /// Removes all elements, keeping the allocated shard capacity.
148
    pub fn clear(&self) {
149
        for shard in &self.shards {
150
            shard.write().clear();
151
        }
152
    }
153

            
154
    /// Retains only the elements for which `keep` returns true.
155
    pub fn retain(&self, mut keep: impl FnMut(&T) -> bool) {
156
        for shard in &self.shards {
157
            shard.write().retain(|element| keep(element));
158
        }
159
    }
160

            
161
    /// Calls `f` on a reference to every element. The shard holding each element
162
    /// is read-locked while `f` runs against it, so `f` must not call back into
163
    /// the same table.
164
1
    pub fn for_each(&self, mut f: impl FnMut(&T)) {
165
8
        for shard in &self.shards {
166
8
            let table = shard.read();
167
256
            for element in table.iter() {
168
256
                f(element);
169
256
            }
170
        }
171
1
    }
172

            
173
    /// Returns a clone of the element matching `hash`/`eq`, if present.
174
769396
    pub fn find(&self, hash: u64, mut eq: impl FnMut(&T) -> bool) -> Option<T>
175
769396
    where
176
769396
        T: Clone,
177
    {
178
769396
        self.shard(hash).read().find(hash, |element| eq(element)).cloned()
179
769396
    }
180

            
181
    /// Looks up the element matching `hash`/`eq` and, if present, calls `f` on a
182
    /// shared reference to it, returning its result; returns `None` otherwise.
183
    ///
184
    /// Unlike [`ShardedHashMap::find`], the stored element is not required to be
185
    /// [`Clone`] and is never copied: `f` runs while the shard read lock is
186
    /// held. Consequently `f` must not call back into the same table (it would
187
    /// deadlock) and should be kept short to limit the lock hold time.
188
818
    pub fn find_with<R>(&self, hash: u64, mut eq: impl FnMut(&T) -> bool, f: impl FnOnce(&T) -> R) -> Option<R> {
189
818
        self.shard(hash).read().find(hash, |element| eq(element)).map(f)
190
818
    }
191

            
192
    /// Returns a clone of the element matching `hash`/`eq`, inserting the value
193
    /// produced by `make` if none is present. `make` runs only on insertion and
194
    /// while the shard is write-locked; `hash_of` recomputes an element's hash
195
    /// when the shard grows. Returns the resident element and whether it was
196
    /// freshly inserted.
197
247698
    pub fn find_or_insert_with(
198
247698
        &self,
199
247698
        hash: u64,
200
247698
        mut eq: impl FnMut(&T) -> bool,
201
247698
        hash_of: impl Fn(&T) -> u64,
202
247698
        make: impl FnOnce() -> T,
203
247698
    ) -> (T, bool)
204
247698
    where
205
247698
        T: Clone,
206
    {
207
247698
        let mut table = self.shard(hash).write();
208
247698
        match table.entry(hash, |element| eq(element), &hash_of) {
209
1
            Entry::Occupied(entry) => (entry.get().clone(), false),
210
247697
            Entry::Vacant(entry) => (entry.insert(make()).get().clone(), true),
211
        }
212
247698
    }
213

            
214
    /// Inserts the element produced by `make` unless an element matching
215
    /// `hash`/`eq` is already present, returning whether a new element was
216
    /// inserted. `make` runs only on insertion and while the shard is
217
    /// write-locked; `hash_of` recomputes an element's hash when the shard
218
    /// grows.
219
    ///
220
    /// Unlike [`ShardedHashMap::find_or_insert_with`], the resident element is
221
    /// neither returned nor cloned, so `T` need not be [`Clone`]. Use this when
222
    /// only publication matters and the stored element is expensive to clone.
223
88
    pub fn insert_with(
224
88
        &self,
225
88
        hash: u64,
226
88
        mut eq: impl FnMut(&T) -> bool,
227
88
        hash_of: impl Fn(&T) -> u64,
228
88
        make: impl FnOnce() -> T,
229
88
    ) -> bool {
230
88
        let mut table = self.shard(hash).write();
231
88
        match table.entry(hash, |element| eq(element), &hash_of) {
232
            Entry::Occupied(_) => false,
233
88
            Entry::Vacant(entry) => {
234
88
                entry.insert(make());
235
88
                true
236
            }
237
        }
238
88
    }
239

            
240
    /// Removes and returns the element matching `hash`/`eq`, if present.
241
    pub fn remove(&self, hash: u64, mut eq: impl FnMut(&T) -> bool) -> Option<T> {
242
        match self.shard(hash).write().find_entry(hash, |element| eq(element)) {
243
            Ok(occupied) => Some(occupied.remove().0),
244
            Err(_) => None,
245
        }
246
    }
247
}
248

            
249
impl<T, S> ShardedHashMap<T, S>
250
where
251
    S: BuildHasher,
252
{
253
    /// Hashes `value` with the table's hash builder. Raw callers use this so the
254
    /// hashes they pass to [`ShardedHashMap::find`] and friends are consistent
255
    /// with the table's convenience methods.
256
1171808
    pub fn hash<Q>(&self, value: &Q) -> u64
257
1171808
    where
258
1171808
        Q: ?Sized + Hash,
259
    {
260
1171808
        self.hasher.hash_one(value)
261
1171808
    }
262

            
263
    /// Inserts `value` unless an equal element is already present. Returns the
264
    /// resident element and whether it was freshly inserted.
265
40976
    pub fn insert(&self, value: T) -> (T, bool)
266
40976
    where
267
40976
        T: Hash + Eq + Clone,
268
    {
269
40976
        let hash = self.hash(&value);
270
40976
        let mut table = self.shard(hash).write();
271
40976
        match table.entry(
272
40976
            hash,
273
27085
            |element| element == &value,
274
4266
            |element| self.hasher.hash_one(element),
275
        ) {
276
26685
            Entry::Occupied(entry) => (entry.get().clone(), false),
277
14291
            Entry::Vacant(entry) => (entry.insert(value).get().clone(), true),
278
        }
279
40976
    }
280

            
281
    /// Returns a clone of the element equivalent to `value`, if present.
282
24917
    pub fn get<Q>(&self, value: &Q) -> Option<T>
283
24917
    where
284
24917
        T: Clone,
285
24917
        Q: ?Sized + Hash + Equivalent<T>,
286
    {
287
24917
        let hash = self.hash(value);
288
24917
        self.shard(hash)
289
24917
            .read()
290
24917
            .find(hash, |element| value.equivalent(element))
291
24917
            .cloned()
292
24917
    }
293

            
294
    /// Returns true if an element equivalent to `value` is present.
295
25096
    pub fn contains<Q>(&self, value: &Q) -> bool
296
25096
    where
297
25096
        Q: ?Sized + Hash + Equivalent<T>,
298
    {
299
25096
        let hash = self.hash(value);
300
25096
        self.shard(hash)
301
25096
            .read()
302
25096
            .find(hash, |element| value.equivalent(element))
303
25096
            .is_some()
304
25096
    }
305

            
306
    /// Removes and returns the element equivalent to `value`, if present.
307
25011
    pub fn remove_equiv<Q>(&self, value: &Q) -> Option<T>
308
25011
    where
309
25011
        Q: ?Sized + Hash + Equivalent<T>,
310
    {
311
25011
        let hash = self.hash(value);
312
25011
        match self
313
25011
            .shard(hash)
314
25011
            .write()
315
25011
            .find_entry(hash, |element| value.equivalent(element))
316
        {
317
10856
            Ok(occupied) => Some(occupied.remove().0),
318
14155
            Err(_) => None,
319
        }
320
25011
    }
321
}
322

            
323
#[cfg(test)]
324
mod tests {
325
    use std::collections::HashMap;
326
    use std::collections::HashSet;
327
    use std::sync::Arc;
328
    use std::sync::Mutex;
329
    use std::sync::atomic::AtomicUsize;
330
    use std::sync::atomic::Ordering;
331

            
332
    use rand::RngExt;
333

            
334
    use merc_utilities::random_test;
335
    use merc_utilities::random_test_threads;
336

            
337
    use super::ShardedHashMap;
338

            
339
    /// Drives the convenience API with random operations against a `HashSet`
340
    /// oracle, checking that the table agrees with the model after each step.
341
    #[test]
342
    #[cfg_attr(miri, ignore)]
343
1
    fn random_insert_get_remove() {
344
100
        random_test(100, |rng| {
345
100
            let map: ShardedHashMap<u64> = ShardedHashMap::with_shards(rng.random_range(1..=16));
346
100
            let mut model: HashSet<u64> = HashSet::new();
347

            
348
100
            for _ in 0..1000 {
349
100000
                let value = rng.random_range(0..64u64);
350
100000
                match rng.random_range(0..4) {
351
24976
                    0 => assert_eq!(map.insert(value).1, model.insert(value), "fresh-insert flags agree"),
352
25096
                    1 => assert_eq!(map.contains(&value), model.contains(&value)),
353
                    2 => {
354
24917
                        let expected = model.contains(&value).then_some(value);
355
24917
                        assert_eq!(map.get(&value), expected);
356
                    }
357
25011
                    _ => assert_eq!(map.remove_equiv(&value).is_some(), model.remove(&value)),
358
                }
359
100000
                assert_eq!(map.len(), model.len());
360
            }
361
100
        });
362
1
    }
363

            
364
    /// Exercises the raw API the way the forest does: the table stores indices
365
    /// into a side vector and the equality/hash closures dereference it. Random
366
    /// payloads are interned and checked against a `HashMap` oracle mapping each
367
    /// payload to the index it should resolve to.
368
    #[test]
369
    #[cfg_attr(miri, ignore)]
370
1
    fn random_raw_interning_dedup() {
371
50
        random_test(50, |rng| {
372
50
            let map: ShardedHashMap<usize> = ShardedHashMap::with_shards(rng.random_range(1..=8));
373
50
            let store: Mutex<Vec<u64>> = Mutex::new(Vec::new());
374

            
375
25000
            let intern = |value: u64| -> usize {
376
25000
                let hash = map.hash(&value);
377
25000
                let eq = |&index: &usize| store.lock().unwrap()[index] == value;
378
25000
                if let Some(index) = map.find(hash, eq) {
379
23400
                    return index;
380
1600
                }
381
1600
                map.find_or_insert_with(
382
1600
                    hash,
383
1600
                    eq,
384
1258
                    |&index: &usize| map.hash(&store.lock().unwrap()[index]),
385
1600
                    || {
386
1600
                        let mut store = store.lock().unwrap();
387
1600
                        store.push(value);
388
1600
                        store.len() - 1
389
1600
                    },
390
                )
391
                .0
392
25000
            };
393

            
394
50
            let mut model: HashMap<u64, usize> = HashMap::new();
395
50
            for _ in 0..500 {
396
25000
                let value = rng.random_range(0..32u64);
397
25000
                let index = intern(value);
398
25000
                let next = model.len();
399
25000
                assert_eq!(
400
                    index,
401
25000
                    *model.entry(value).or_insert(next),
402
                    "equal payloads share an index"
403
                );
404
            }
405

            
406
50
            assert_eq!(map.len(), model.len());
407
50
            assert_eq!(
408
50
                store.lock().unwrap().len(),
409
50
                model.len(),
410
                "no payload pushed for a duplicate"
411
            );
412
50
        });
413
1
    }
414

            
415
    /// Concurrently inserts random values from a fixed value space. Across all
416
    /// threads each value yields exactly one fresh insertion, so the count of
417
    /// fresh insertions must equal the table size.
418
    #[test]
419
    #[cfg_attr(miri, ignore)]
420
1
    fn random_concurrent_insert_counts_each_once() {
421
1
        let map: Arc<ShardedHashMap<u64>> = Arc::new(ShardedHashMap::with_shards(8));
422
1
        let inserted = Arc::new(AtomicUsize::new(0));
423
1
        let values = 256u64;
424

            
425
1
        random_test_threads(
426
            2000,
427
            8,
428
8
            || (Arc::clone(&map), Arc::clone(&inserted)),
429
16000
            move |rng, (map, inserted)| {
430
16000
                let value = rng.random_range(0..values);
431
16000
                if map.insert(value).1 {
432
256
                    inserted.fetch_add(1, Ordering::Relaxed);
433
15744
                }
434
16000
            },
435
        );
436

            
437
1
        assert_eq!(
438
1
            inserted.load(Ordering::Relaxed),
439
1
            map.len(),
440
            "every fresh insertion is a distinct resident value"
441
        );
442
256
        map.for_each(|&value| assert!(value < values, "only inserted values are resident"));
443
1
    }
444
}