1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
//! Resizing and clearing a table. Both take every home guard of the table first, so no writer
//! is inside a home while its slots are copied or cleared, and a resize keeps the guards of the
//! table it replaced, so that table never admits a writer again.
extern crate alloc;
use super::table::{Entry, GUARD, Table, Walk, Word, hop_bit};
use super::{HopscotchMap, MIN_CAPACITY, NEIGHBORHOOD_SIZE};
use crate::sync::spin_hint;
use core::hash::{BuildHasher, Hash};
use core::sync::atomic::Ordering;
use kovan::{Shared, alloc_node, drop_node, pin, retire_node};
#[cfg(test)]
use super::pause;
use crate::MaybeSend;
impl<K, V, S> HopscotchMap<K, V, S>
where
K: Hash + Eq + Clone + MaybeSend + 'static,
V: Clone + MaybeSend + 'static,
S: BuildHasher,
{
/// Take every home bucket's writer guard of `table`, waiting out each
/// writer that holds one (bounded: a guard holder runs its write to its
/// end and waits for nothing). Once this returns no writer is inside a
/// home anywhere in `table` (no insert between its existence scan and its
/// slot claim, no remove, no move with an entry in two slots), and none
/// can start there: a writer that fails its guard reloads the table. Only
/// the thread that set `resizing` calls it (a resize or a clear), so two
/// such callers never wait on each other.
pub(super) fn hold_writers(table: &Table<K, V>) {
for idx in 0..table.capacity {
let control = &table.get_bucket(idx).control;
// Acquire pairs with a holder's release of the guard: every slot
// and hop bit it wrote is visible to the copy or clear after this.
while control.fetch_or(GUARD, Ordering::Acquire) & GUARD != 0 {
#[cfg(test)]
pause::at(pause::Point::ResizerMetHeldGuard);
spin_hint();
}
}
}
/// Give back what [`Self::hold_writers`] took, for a table that stays
/// live (a clear, or a resize that did not publish its new table).
pub(super) fn release_writers(table: &Table<K, V>) {
for idx in 0..table.capacity {
// Release: pairs with the next holder's acquire.
table
.get_bucket(idx)
.control
.fetch_and(!GUARD, Ordering::Release);
}
}
fn insert_into_new_table(
&self,
table: &Table<K, V>,
hash: u64,
key: K,
value: V,
guard: &kovan::Guard,
) -> bool {
let bucket_idx = table.bucket_index(hash);
for probe_offset in 0..(table.capacity + NEIGHBORHOOD_SIZE) {
let probe_idx = bucket_idx + probe_offset;
if probe_idx >= table.capacity + NEIGHBORHOOD_SIZE {
break;
}
let probe_bucket = table.get_bucket(probe_idx);
if probe_bucket.load(Ordering::Relaxed, guard).is_free() {
let offset_from_home = probe_idx - bucket_idx;
if offset_from_home < NEIGHBORHOOD_SIZE {
probe_bucket.store(Word::of(Entry::boxed(hash, key, value)), Ordering::Release);
// A read and a store, not a read-modify-write: the new table is this
// resize's alone until it publishes it (the table pointer's release store
// carries this word to every thread that acquires the table), so no other
// thread writes the word meanwhile.
let control = &table.get_bucket(bucket_idx).control;
let word = control.load(Ordering::Relaxed);
control.store(word | hop_bit(offset_from_home), Ordering::Relaxed);
return true;
} else {
return false;
}
}
}
false
}
pub(super) fn try_resize(&self, new_capacity: usize) {
if self
.resizing
.compare_exchange(false, true, Ordering::SeqCst, Ordering::Relaxed)
.is_err()
{
return;
}
let mut new_capacity = new_capacity.next_power_of_two().max(MIN_CAPACITY);
let guard = pin();
let old_table_ptr = self.table.load(Ordering::Acquire, &guard);
let old_table = unsafe { &*old_table_ptr.as_raw() };
if old_table.capacity == new_capacity {
self.resizing.store(false, Ordering::Release);
return;
}
// Every insert into the old table lands before the copy below reads
// its slot, or waits and lands in the new table: none is lost, and
// none has to retry into the new table and meet its own entry there.
// No move has an entry in two slots for the copy to meet twice either
// (a move holds its entry's home guard).
Self::hold_writers(old_table);
// The copy puts each entry in the first free slot of its neighborhood
// and displaces nothing, so it can find a neighborhood full in a table
// that could hold every entry. A grow then doubles again and copies
// again: each doubling splits the entries of a crowded neighborhood by
// one more bit of their hashes (no neighborhood of the old table held
// more than `NEIGHBORHOOD_SIZE` entries of one home, so the copy fits
// once the hashes that crowd it differ). Before, the table stayed as it
// was and the insert that needed the room asked for the same resize
// again, forever. A shrink that does not fit is given up.
// A `K::clone` or `V::clone` that panics in a copy unwinds through this, which undoes
// the resize (see `Undo`): the old table stays current and its writers get their guards
// back, where the guards and the latch held for good would have stopped every writer.
let mut undo = Undo {
old: old_table,
new: core::ptr::null_mut(),
latch: &self.resizing,
};
let published = loop {
let new_table = alloc_node(Table::new(new_capacity));
undo.new = new_table;
let new_table_ref = unsafe { &*new_table };
if self.copy_into(old_table, new_table_ref, &guard) {
break Some(new_table);
}
undo.new = core::ptr::null_mut();
// The unpublished new table's destructor frees its cloned entries.
unsafe { drop_node(new_table) };
if new_capacity < old_table.capacity {
break None;
}
new_capacity *= 2;
};
core::mem::forget(undo);
match published {
// The new table starts with no failed shrink (`Table::new` sets its `shrink_at`).
Some(new_table) => {
self.table
.store(unsafe { Shared::from_raw(new_table) }, Ordering::Release);
// The replaced table keeps its writer guards held: a writer that
// still holds its pointer fails its guard, reloads and writes to
// the new table.
unsafe { retire_node(old_table_ptr.as_raw()) };
}
// The old table stays live, so its writers get their guards back. Its removes ask for
// the shrink again only once half the entries it failed at have gone (at least one
// remains), and never while the table is a quarter full or more: written while every
// home guard is still held, before they are given back.
None => {
let failed_at = (self.count.load(Ordering::Relaxed) / 2).max(1);
old_table.shrink_at.store(
super::shrink_at(old_table.capacity).min(failed_at),
Ordering::Relaxed,
);
Self::release_writers(old_table);
}
}
self.resizing.store(false, Ordering::Release);
}
/// Copy every entry of `old` (whose writers are held) into `new`, unpublished: `false` when
/// an entry found its neighborhood in `new` full.
fn copy_into(&self, old: &Table<K, V>, new: &Table<K, V>, guard: &kovan::Guard) -> bool {
let mut walk = Walk::new(old, guard);
while let Some((_, entry)) = walk.next(old, guard) {
// SAFETY: loaded under `guard`; the held writers keep it in its slot.
if let Some(entry) = unsafe { entry.as_ref() }
&& !self.insert_into_new_table(
new,
entry.hash,
entry.key.clone(),
entry.value.clone(),
guard,
)
{
return false;
}
}
true
}
}
/// Undoes a resize whose copy unwinds (a clone of a key or value panicked in it): the new table
/// (no other thread ever saw it) dropped with the copies made so far, the old table's writer
/// guards given back, and the latch let go, so the map goes on with its old table.
struct Undo<'a, K: 'static, V: 'static> {
old: &'a Table<K, V>,
new: *mut Table<K, V>,
latch: &'a crate::sync::AtomicBool,
}
impl<K: 'static, V: 'static> Drop for Undo<'_, K, V> {
fn drop(&mut self) {
if !self.new.is_null() {
// SAFETY: never published, so this thread alone reaches it and its copies.
unsafe { drop_node(self.new) };
}
for idx in 0..self.old.capacity {
// Release: pairs with the next holder's acquire, as `release_writers` does.
self.old
.get_bucket(idx)
.control
.fetch_and(!GUARD, Ordering::Release);
}
self.latch.store(false, Ordering::Release);
}
}