forked from jonhoo/flurry
-
Notifications
You must be signed in to change notification settings - Fork 0
/
map.rs
2652 lines (2410 loc) · 107 KB
/
map.rs
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
use crate::iter::*;
use crate::node::*;
use crate::raw::*;
use crossbeam_epoch::{self as epoch, Atomic, Guard, Owned, Shared};
use std::borrow::Borrow;
use std::fmt::{self, Debug, Formatter};
use std::hash::{BuildHasher, Hash, Hasher};
use std::iter::FromIterator;
use std::sync::{
atomic::{AtomicIsize, AtomicUsize, Ordering},
Once,
};
const ISIZE_BITS: usize = core::mem::size_of::<isize>() * 8;
/// The largest possible table capacity. This value must be
/// exactly 1<<30 to stay within Java array allocation and indexing
/// bounds for power of two table sizes, and is further required
/// because the top two bits of 32bit hash fields are used for
/// control purposes.
const MAXIMUM_CAPACITY: usize = 1 << 30; // TODO: use ISIZE_BITS
/// The default initial table capacity. Must be a power of 2
/// (i.e., at least 1) and at most `MAXIMUM_CAPACITY`.
const DEFAULT_CAPACITY: usize = 16;
/// The bin count threshold for using a tree rather than list for a bin. Bins are
/// converted to trees when adding an element to a bin with at least this many
/// nodes. The value must be greater than 2, and should be at least 8 to mesh
/// with assumptions in tree removal about conversion back to plain bins upon
/// shrinkage.
const TREEIFY_THRESHOLD: usize = 8;
/// The bin count threshold for untreeifying a (split) bin during a resize
/// operation. Should be less than TREEIFY_THRESHOLD, and at most 6 to mesh with
/// shrinkage detection under removal.
const UNTREEIFY_THRESHOLD: usize = 6;
/// The smallest table capacity for which bins may be treeified. (Otherwise the
/// table is resized if too many nodes in a bin.) The value should be at least 4
/// * TREEIFY_THRESHOLD to avoid conflicts between resizing and treeification
/// thresholds.
const MIN_TREEIFY_CAPACITY: usize = 64;
/// Minimum number of rebinnings per transfer step. Ranges are
/// subdivided to allow multiple resizer threads. This value
/// serves as a lower bound to avoid resizers encountering
/// excessive memory contention. The value should be at least
/// `DEFAULT_CAPACITY`.
const MIN_TRANSFER_STRIDE: isize = 16;
/// The number of bits used for generation stamp in `size_ctl`.
/// Must be at least 6 for 32bit arrays.
const RESIZE_STAMP_BITS: usize = ISIZE_BITS / 2;
/// The maximum number of threads that can help resize.
/// Must fit in `32 - RESIZE_STAMP_BITS` bits for 32 bit architectures
/// and `64 - RESIZE_STAMP_BITS` bits for 64 bit architectures
const MAX_RESIZERS: isize = (1 << (ISIZE_BITS - RESIZE_STAMP_BITS)) - 1;
/// The bit shift for recording size stamp in `size_ctl`.
const RESIZE_STAMP_SHIFT: usize = ISIZE_BITS - RESIZE_STAMP_BITS;
static NCPU_INITIALIZER: Once = Once::new();
static NCPU: AtomicUsize = AtomicUsize::new(0);
macro_rules! load_factor {
($n: expr) => {
// ¾ n = n - n/4 = n - (n >> 2)
$n - ($n >> 2)
};
}
/// A concurrent hash table.
///
/// Flurry uses an [`Guards`] to control the lifetime of the resources that get stored and
/// extracted from the map. [`Guards`] are acquired through the [`epoch::pin`], [`HashMap::pin`]
/// and [`HashMap::guard`] functions. For more information, see the [notes in the crate-level
/// documentation].
///
/// [notes in the crate-level documentation]: index.html#a-note-on-guard-and-memory-use
/// [`Guards`]: index.html#a-note-on-guard-and-memory-use
pub struct HashMap<K, V, S = crate::DefaultHashBuilder> {
/// The array of bins. Lazily initialized upon first insertion.
/// Size is always a power of two. Accessed directly by iterators.
table: Atomic<Table<K, V>>,
/// The next table to use; non-null only while resizing.
next_table: Atomic<Table<K, V>>,
/// The next table index (plus one) to split while resizing.
transfer_index: AtomicIsize,
count: AtomicUsize,
/// Table initialization and resizing control. When negative, the
/// table is being initialized or resized: -1 for initialization,
/// else -(1 + the number of active resizing threads). Otherwise,
/// when table is null, holds the initial table size to use upon
/// creation, or 0 for default. After initialization, holds the
/// next element count value upon which to resize the table.
size_ctl: AtomicIsize,
/// Collector that all `Guard` references used for operations on this map must be tied to. It
/// is important that they all assocate with the _same_ `Collector`, otherwise you end up with
/// unsoundness as described in https://github.com/jonhoo/flurry/issues/46. Specifically, a
/// user can do:
///
// this should be:
// ```rust,should_panic
// but that won't work with coverage at the moment:
// https://github.com/xd009642/tarpaulin/issues/344
// so:
/// ```rust,no_run
/// # use flurry::HashMap;
/// # use crossbeam_epoch;
/// let map: HashMap<_, _> = HashMap::default();
/// map.insert(42, String::from("hello"), &crossbeam_epoch::pin());
///
/// let evil = crossbeam_epoch::Collector::new();
/// let evil = evil.register();
/// let guard = evil.pin();
/// let oops = map.get(&42, &guard);
///
/// map.remove(&42, &crossbeam_epoch::pin());
/// // at this point, the default collector is allowed to free `"hello"`
/// // since no-one has the global epoch pinned as far as it is aware.
/// // `oops` is tied to the lifetime of a Guard that is not a part of
/// // the same epoch group, and so can now be dangling.
/// // but we can still access it!
/// assert_eq!(oops.unwrap(), "hello");
/// ```
///
/// We avoid that by checking that every external guard that is passed in is associated with
/// the `Collector` that was specified when the map was created (which may be the global
/// collector).
///
/// Note also that the fact that this can be a global collector is what necessitates the
/// `'static` bounds on `K` and `V`. Since deallocation can be deferred arbitrarily, it is not
/// okay for us to take a `K` or `V` with a limited lifetime, since we may drop it far after
/// that lifetime has passed.
///
/// One possibility is to never use the global allocator, and instead _always_ create and use
/// our own `Collector`. If we did that, then we could accept non-`'static` keys and values since
/// the destruction of the collector would ensure that that all deferred destructors are run.
/// It would, sadly, mean that we don't get to share a collector with other things that use
/// `crossbeam-epoch` though. For more on this (and a cool optimization), see:
/// https://github.com/crossbeam-rs/crossbeam/blob/ebecb82c740a1b3d9d10f235387848f7e3fa9c68/crossbeam-skiplist/src/base.rs#L308-L319
collector: epoch::Collector,
build_hasher: S,
}
#[cfg(test)]
#[test]
#[should_panic]
fn disallow_evil() {
let map: HashMap<_, _> = HashMap::default();
map.insert(42, String::from("hello"), &crossbeam_epoch::pin());
let evil = crossbeam_epoch::Collector::new();
let evil = evil.register();
let guard = evil.pin();
let oops = map.get(&42, &guard);
map.remove(&42, &crossbeam_epoch::pin());
// at this point, the default collector is allowed to free `"hello"`
// since no-one has the global epoch pinned as far as it is aware.
// `oops` is tied to the lifetime of a Guard that is not a part of
// the same epoch group, and so can now be dangling.
// but we can still access it!
assert_eq!(oops.unwrap(), "hello");
}
// ===
// the following methods only see Ks and Vs if there have been inserts.
// modifications to the map are all guarded by thread-safety bounds (Send + Sync + 'static).
// but _these_ methods do not need to be, since they will never introduce keys or values, only give
// out ones that have already been inserted (which implies they must be thread-safe).
// ===
impl<K, V> HashMap<K, V, crate::DefaultHashBuilder> {
/// Creates an empty `HashMap`.
///
/// The hash map is initially created with a capacity of 0, so it will not allocate until it
/// is first inserted into.
///
/// # Examples
///
/// ```
/// use flurry::HashMap;
/// let map: HashMap<&str, i32> = HashMap::new();
/// ```
pub fn new() -> Self {
Self::default()
}
/// Creates an empty `HashMap` with the specified capacity.
///
/// The hash map will be able to hold at least `capacity` elements without
/// reallocating. If `capacity` is 0, the hash map will not allocate.
///
/// # Examples
///
/// ```
/// use flurry::HashMap;
/// let map: HashMap<&str, i32> = HashMap::with_capacity(10);
/// ```
///
/// # Notes
///
/// There is no guarantee that the HashMap will not resize if `capacity`
/// elements are inserted. The map will resize based on key collision, so
/// bad key distribution may cause a resize before `capacity` is reached.
/// For more information see the [`resizing behavior`]
///
/// [`resizing behavior`]: index.html#resizing-behavior
pub fn with_capacity(capacity: usize) -> Self {
Self::with_capacity_and_hasher(capacity, crate::DefaultHashBuilder::default())
}
}
impl<K, V, S> Default for HashMap<K, V, S>
where
S: Default,
{
fn default() -> Self {
Self::with_hasher(S::default())
}
}
impl<K, V, S> HashMap<K, V, S> {
/// Creates an empty map which will use `hash_builder` to hash keys.
///
/// The created map has the default initial capacity.
///
/// Warning: `hash_builder` is normally randomly generated, and is designed to
/// allow the map to be resistant to attacks that cause many collisions and
/// very poor performance. Setting it manually using this
/// function can expose a DoS attack vector.
///
/// # Examples
///
/// ```
/// use flurry::{HashMap, DefaultHashBuilder};
///
/// let map = HashMap::with_hasher(DefaultHashBuilder::default());
/// map.pin().insert(1, 2);
/// ```
pub fn with_hasher(hash_builder: S) -> Self {
Self {
table: Atomic::null(),
next_table: Atomic::null(),
transfer_index: AtomicIsize::new(0),
count: AtomicUsize::new(0),
size_ctl: AtomicIsize::new(0),
build_hasher: hash_builder,
collector: epoch::default_collector().clone(),
}
}
/// Creates an empty map with the specified `capacity`, using `hash_builder` to hash the keys.
///
/// The map will be sized to accommodate `capacity` elements with a low chance of reallocating
/// (assuming uniformly distributed hashes). If `capacity` is 0, the call will not allocate,
/// and is equivalent to [`HashMap::new`].
///
/// Warning: `hash_builder` is normally randomly generated, and is designed to allow the map
/// to be resistant to attacks that cause many collisions and very poor performance.
/// Setting it manually using this function can expose a DoS attack vector.
///
/// # Examples
///
/// ```
/// use flurry::HashMap;
/// use std::collections::hash_map::RandomState;
///
/// let s = RandomState::new();
/// let map = HashMap::with_capacity_and_hasher(10, s);
/// map.pin().insert(1, 2);
/// ```
pub fn with_capacity_and_hasher(capacity: usize, hash_builder: S) -> Self {
if capacity == 0 {
return Self::with_hasher(hash_builder);
}
let mut map = Self::with_hasher(hash_builder);
map.presize(capacity);
map
}
/*
NOTE: This method is intentionally left out atm as it is a potentially large foot-gun.
See https://github.com/jonhoo/flurry/pull/49#issuecomment-580514518.
*/
/*
/// Associate a custom [`epoch::Collector`] with this map.
///
/// By default, the global collector is used. With this method you can use a different
/// collector instead. This may be desireable if you want more control over when and how memory
/// reclamation happens.
///
/// Note that _all_ `Guard` references provided to access the returned map _must_ be
/// constructed using guards produced by `collector`. You can use [`HashMap::register`] to get
/// a thread-local handle to the collector that then lets you construct an [`epoch::Guard`].
pub fn with_collector(mut self, collector: epoch::Collector) -> Self {
self.collector = collector;
self
}
/// Allocate a thread-local handle to the [`epoch::Collector`] associated with this map.
///
/// You can use the returned handle to produce [`epoch::Guard`] references.
pub fn register(&self) -> epoch::LocalHandle {
self.collector.register()
}
*/
/// Pin a `Guard` for use with this map.
///
/// Keep in mind that for as long as you hold onto this `Guard`, you are preventing the
/// collection of garbage generated by the map.
pub fn guard(&self) -> epoch::Guard {
self.collector.register().pin()
}
#[inline]
fn check_guard(&self, guard: &Guard) {
// guard.collector() may be `None` if it is unprotected
if let Some(c) = guard.collector() {
assert_eq!(c, &self.collector);
}
}
/// Returns the number of entries in the map.
///
/// # Examples
///
/// ```
/// use flurry::HashMap;
///
/// let map = HashMap::new();
///
/// map.pin().insert(1, "a");
/// map.pin().insert(2, "b");
/// assert!(map.pin().len() == 2);
/// ```
pub fn len(&self) -> usize {
self.count.load(Ordering::Relaxed)
}
/// Returns `true` if the map is empty. Otherwise returns `false`.
///
/// # Examples
///
/// ```
/// use flurry::HashMap;
///
/// let map = HashMap::new();
/// assert!(map.pin().is_empty());
/// map.pin().insert("a", 1);
/// assert!(!map.pin().is_empty());
/// ```
pub fn is_empty(&self) -> bool {
self.len() == 0
}
#[cfg(test)]
/// Returns the capacity of the map.
fn capacity(&self, guard: &Guard) -> usize {
self.check_guard(guard);
let table = self.table.load(Ordering::Relaxed, &guard);
if table.is_null() {
0
} else {
// Safety: we loaded `table` under the `guard`,
// so it must still be valid here
unsafe { table.deref() }.len()
}
}
/// Returns the stamp bits for resizing a table of size n.
/// Must be negative when shifted left by `RESIZE_STAMP_SHIFT`.
fn resize_stamp(n: usize) -> isize {
n.leading_zeros() as isize | (1_isize << (RESIZE_STAMP_BITS - 1))
}
/// An iterator visiting all key-value pairs in arbitrary order.
///
/// The iterator element type is `(&'g K, &'g V)`.
pub fn iter<'g>(&'g self, guard: &'g Guard) -> Iter<'g, K, V> {
self.check_guard(guard);
let table = self.table.load(Ordering::SeqCst, guard);
let node_iter = NodeIter::new(table, guard);
Iter { node_iter, guard }
}
/// An iterator visiting all keys in arbitrary order.
///
/// The iterator element type is `&'g K`.
pub fn keys<'g>(&'g self, guard: &'g Guard) -> Keys<'g, K, V> {
self.check_guard(guard);
let table = self.table.load(Ordering::SeqCst, guard);
let node_iter = NodeIter::new(table, guard);
Keys { node_iter }
}
/// An iterator visiting all values in arbitrary order.
///
/// The iterator element type is `&'g V`.
pub fn values<'g>(&'g self, guard: &'g Guard) -> Values<'g, K, V> {
self.check_guard(guard);
let table = self.table.load(Ordering::SeqCst, guard);
let node_iter = NodeIter::new(table, guard);
Values { node_iter, guard }
}
fn init_table<'g>(&'g self, guard: &'g Guard) -> Shared<'g, Table<K, V>> {
loop {
let table = self.table.load(Ordering::SeqCst, guard);
// safety: we loaded the table while epoch was pinned. table won't be deallocated until
// next epoch at the earliest.
if !table.is_null() && !unsafe { table.deref() }.is_empty() {
break table;
}
// try to allocate the table
let mut sc = self.size_ctl.load(Ordering::SeqCst);
if sc < 0 {
// we lost the initialization race; just spin
std::thread::yield_now();
continue;
}
if self.size_ctl.compare_and_swap(sc, -1, Ordering::SeqCst) == sc {
// we get to do it!
let mut table = self.table.load(Ordering::SeqCst, guard);
// safety: we loaded the table while epoch was pinned. table won't be deallocated
// until next epoch at the earliest.
if table.is_null() || unsafe { table.deref() }.is_empty() {
let n = if sc > 0 {
sc as usize
} else {
DEFAULT_CAPACITY
};
let new_table = Owned::new(Table::new(n));
table = new_table.into_shared(guard);
self.table.store(table, Ordering::SeqCst);
sc = load_factor!(n as isize)
}
self.size_ctl.store(sc, Ordering::SeqCst);
break table;
}
}
}
/// Presize the table to accommodate the given number of elements.
fn presize(&mut self, size: usize) {
// NOTE: this is a stripped-down version of try_presize for use only when we _know_ that
// the table is new, and that therefore we won't have to help out with transfers or deal
// with contending initializations.
// safety: we are creating this map, so no other thread can access it,
// while we are initializing it.
let guard = unsafe { epoch::unprotected() };
let requested_capacity = if size >= MAXIMUM_CAPACITY / 2 {
MAXIMUM_CAPACITY
} else {
// round the requested_capacity to the next power of to from 1.5 * size + 1
// TODO: find out if this is neccessary
let size = size + (size >> 1) + 1;
std::cmp::min(MAXIMUM_CAPACITY, size.next_power_of_two())
} as usize;
// sanity check that the map has indeed not been set up already
assert_eq!(self.size_ctl.load(Ordering::SeqCst), 0);
assert!(self.table.load(Ordering::SeqCst, &guard).is_null());
// the table has not yet been initialized, so we can just create it
// with as many bins as were requested
// create a table with `new_capacity` empty bins
let new_table = Owned::new(Table::new(requested_capacity)).into_shared(guard);
// store the new table to `self.table`
self.table.store(new_table, Ordering::SeqCst);
// resize the table once it is 75% full
let new_load_to_resize_at = load_factor!(requested_capacity as isize);
// store the next load at which the table should resize to it's size_ctl field
// and thus release the initialization "lock"
self.size_ctl.store(new_load_to_resize_at, Ordering::SeqCst);
}
}
// ===
// the following methods require Clone, since they ultimately call `transfer`, which needs to be
// able to clone keys. however, they do _not_ need to require thread-safety bounds (Send + Sync +
// 'static) since if the bounds do not hold, the map is empty, so no keys or values will be
// transfered anyway.
// ===
impl<K, V, S> HashMap<K, V, S>
where
K: Clone,
{
/// Tries to presize table to accommodate the given number of elements.
fn try_presize(&self, size: usize, guard: &Guard) {
let requested_capacity = if size >= MAXIMUM_CAPACITY / 2 {
MAXIMUM_CAPACITY
} else {
// round the requested_capacity to the next power of to from 1.5 * size + 1
// TODO: find out if this is neccessary
let size = size + (size >> 1) + 1;
std::cmp::min(MAXIMUM_CAPACITY, size.next_power_of_two())
} as isize;
loop {
let size_ctl = self.size_ctl.load(Ordering::SeqCst);
if size_ctl < 0 {
break;
}
let table = self.table.load(Ordering::SeqCst, &guard);
// The current capacity == the number of bins in the current table
let current_capactity = if table.is_null() {
0
} else {
unsafe { table.deref() }.len()
};
if current_capactity == 0 {
// the table has not yet been initialized, so we can just create it
// with as many bins as were requested
// since the map is uninitialized, size_ctl describes the initial capacity
let initial_capacity = size_ctl;
// the new capacity is either the requested capacity or the initial capacity (size_ctl)
let new_capacity = requested_capacity.max(initial_capacity) as usize;
// try to aquire the initialization "lock" to indicate that we are initializing the table.
if self
.size_ctl
.compare_and_swap(size_ctl, -1, Ordering::SeqCst)
!= size_ctl
{
// somebody else is already initializing the table (or has already finished).
continue;
}
// we got the initialization `lock`; Make sure the table is still unitialized
// (or is the same table with 0 bins we read earlier, althought that should not be the case)
if self.table.load(Ordering::SeqCst, guard) != table {
// NOTE: this could probably be `!self.table.load(...).is_null()`
// if we decide that tables can never have 0 bins.
// the table is already initialized; Write the `size_ctl` value it had back to it's
// `size_ctl` field to release the initialization "lock"
self.size_ctl.store(size_ctl, Ordering::SeqCst);
continue;
}
// create a table with `new_capacity` empty bins
let new_table = Owned::new(Table::new(new_capacity)).into_shared(guard);
// store the new table to `self.table`
let old_table = self.table.swap(new_table, Ordering::SeqCst, &guard);
// old_table should be `null`, since we don't ever initialize a table with 0 bins
// and this branch only happens if table has not yet been initialized or it's length is 0.
assert!(old_table.is_null());
// TODO: if we allow tables with 0 bins. `defer_destroy` `old_table` if it's not `null`:
// if !old_table.is_null() {
// // TODO: safety argument, for why this is okay
// unsafe { guard.defer_destroy(old_table) }
// }
// resize the table once it is 75% full
let new_load_to_resize_at = load_factor!(new_capacity as isize);
// store the next load at which the table should resize to it's size_ctl field
// and thus release the initialization "lock"
self.size_ctl.store(new_load_to_resize_at, Ordering::SeqCst);
} else if requested_capacity <= size_ctl || current_capactity >= MAXIMUM_CAPACITY {
// Either the `requested_capacity` was smaller than or equal to the load we would resize at (size_ctl)
// and we don't need to resize, since our load factor will still be acceptable if we don't
// Or it was larger than the `MAXIMUM_CAPACITY` of the map and we refuse
// to resize to an invalid capacity
break;
} else if table == self.table.load(Ordering::SeqCst, &guard) {
// The table is initialized, try to resize it to the requested capacity
let rs: isize = Self::resize_stamp(current_capactity) << RESIZE_STAMP_SHIFT;
// TODO: see #29: `rs` is postive even though `resize_stamp` says:
// "Must be negative when shifted left by RESIZE_STAMP_SHIFT"
// and since our size_control field needs to be negative
// to indicate a resize this needs to be addressed
if self
.size_ctl
.compare_and_swap(size_ctl, rs + 2, Ordering::SeqCst)
== size_ctl
{
// someone else already started to resize the table
// TODO: can we `self.help_transfer`?
self.transfer(table, Shared::null(), &guard);
}
}
}
}
// NOTE: transfer requires that K and V are Send + Sync if it will actually transfer anything.
// If K/V aren't Send + Sync, the map must be empty, and therefore calling tansfer is fine.
#[inline(never)]
fn transfer<'g>(
&'g self,
table: Shared<'g, Table<K, V>>,
mut next_table: Shared<'g, Table<K, V>>,
guard: &'g Guard,
) {
// safety: table was read while `guard` was held. the code that drops table only drops it
// after it is no longer reachable, and any outstanding references are no longer active.
// this references is still active (marked by the guard), so the target of the references
// won't be dropped while the guard remains active.
let n = unsafe { table.deref() }.len();
let ncpu = num_cpus();
let stride = if ncpu > 1 { (n >> 3) / ncpu } else { n };
let stride = std::cmp::max(stride as isize, MIN_TRANSFER_STRIDE);
if next_table.is_null() {
// we are initiating a resize
let table = Owned::new(Table::new(n << 1));
let now_garbage = self.next_table.swap(table, Ordering::SeqCst, guard);
assert!(now_garbage.is_null());
self.transfer_index.store(n as isize, Ordering::SeqCst);
next_table = self.next_table.load(Ordering::Relaxed, guard);
}
// safety: same argument as for table above
let next_n = unsafe { next_table.deref() }.len();
let mut advance = true;
let mut finishing = false;
let mut i = 0;
let mut bound = 0;
loop {
// try to claim a range of bins for us to transfer
while advance {
i -= 1;
if i >= bound || finishing {
advance = false;
break;
}
let next_index = self.transfer_index.load(Ordering::SeqCst);
if next_index <= 0 {
i = -1;
advance = false;
break;
}
let next_bound = if next_index > stride {
next_index - stride
} else {
0
};
if self
.transfer_index
.compare_and_swap(next_index, next_bound, Ordering::SeqCst)
== next_index
{
bound = next_bound;
i = next_index;
advance = false;
break;
}
}
if i < 0 || i as usize >= n || i as usize + n >= next_n {
// the resize has finished
if finishing {
// this branch is only taken for one thread partaking in the resize!
self.next_table.store(Shared::null(), Ordering::SeqCst);
let now_garbage = self.table.swap(next_table, Ordering::SeqCst, guard);
// safety: need to guarantee that now_garbage is no longer reachable. more
// specifically, no thread that executes _after_ this line can ever get a
// reference to now_garbage.
//
// first, we need to argue that there is no _other_ way to get to now_garbage.
//
// - it is _not_ accessible through self.table any more
// - it is _not_ accessible through self.next_table any more
// - what about forwarding nodes (BinEntry::Moved)?
// the only BinEntry::Moved that point to now_garbage, are the ones in
// _previous_ tables. to get to those previous tables, one must ultimately
// have arrived through self.table (because that's where all operations
// start their search). since self.table has now changed, only "old" threads
// can still be accessing them. no new thread can get to past tables, and
// therefore they also cannot get to ::Moved that point to now_garbage, so
// we're fine.
//
// this means that no _future_ thread (i.e., in a later epoch where the value
// may be freed) can get a reference to now_garbage.
//
// next, let's talk about threads with _existing_ references to now_garbage.
// such a thread must have gotten that reference before the call to swap.
// because of this, that thread must be pinned to an epoch <= the epoch of our
// guard (since our guard is pinning the epoch). since the garbage is placed in
// our epoch, it won't be freed until the _next_ epoch, at which point, that
// thread must have dropped its guard, and with it, any reference to the value.
unsafe { guard.defer_destroy(now_garbage) };
self.size_ctl
.store(((n as isize) << 1) - ((n as isize) >> 1), Ordering::SeqCst);
return;
}
let sc = self.size_ctl.load(Ordering::SeqCst);
if self.size_ctl.compare_and_swap(sc, sc - 1, Ordering::SeqCst) == sc {
if (sc - 2) != Self::resize_stamp(n) << RESIZE_STAMP_SHIFT {
return;
}
// we are the chosen thread to finish the resize!
finishing = true;
// ???
advance = true;
// NOTE: the java code says "recheck before commit" here
i = n as isize;
}
continue;
}
let i = i as usize;
// safety: these were read while `guard` was held. the code that drops these, only
// drops them after a) they are no longer reachable, and b) any outstanding references
// are no longer active. these references are still active (marked by the guard), so
// the target of these references won't be dropped while the guard remains active.
let table = unsafe { table.deref() };
let bin = table.bin(i as usize, guard);
if bin.is_null() {
advance = table
.cas_bin(i, Shared::null(), table.get_moved(next_table, guard), guard)
.is_ok();
continue;
}
// safety: as for table above
let next_table = unsafe { next_table.deref() };
// safety: bin is a valid pointer.
//
// there are two cases when a bin pointer is invalidated:
//
// 1. if the table was resized, bin is a move entry, and the resize has completed. in
// that case, the table (and all its heads) will be dropped in the next epoch
// following that.
// 2. if the table is being resized, bin may be swapped with a move entry. the old bin
// will then be dropped in the following epoch after that happens.
//
// in both cases, we held the guard when we got the reference to the bin. if any such
// swap happened, it must have happened _after_ we read. since we did the read while
// pinning the epoch, the drop must happen in the _next_ epoch (i.e., the one that we
// are holding up by holding on to our guard).
match *unsafe { bin.deref() } {
BinEntry::Moved => {
// already processed
advance = true;
}
BinEntry::Node(ref head) => {
// bin is non-empty, need to link into it, so we must take the lock
let head_lock = head.lock.lock();
// need to check that this is _still_ the head
let current_head = table.bin(i, guard);
if current_head.as_raw() != bin.as_raw() {
// nope -- try again from the start
continue;
}
// yes, it is still the head, so we can now "own" the bin
// note that there can still be readers in the bin!
// TODO: TreeBin & ReservationNode
let mut run_bit = head.hash & n as u64;
let mut last_run = bin;
let mut p = bin;
loop {
// safety: p is a valid pointer.
//
// p is only dropped in the next epoch following when its bin is replaced
// with a move node (see safety comment near table.store_bin below). we
// read the bin, and got to p, so its bin has not yet been swapped with a
// move node. and, we have the epoch pinned, so the next epoch cannot have
// arrived yet. therefore, it will be dropped in a future epoch, and is
// safe to use now.
let node = unsafe { p.deref() }.as_node().unwrap();
let next = node.next.load(Ordering::SeqCst, guard);
let b = node.hash & n as u64;
if b != run_bit {
run_bit = b;
last_run = p;
}
if next.is_null() {
break;
}
p = next;
}
let mut low_bin = Shared::null();
let mut high_bin = Shared::null();
if run_bit == 0 {
// last run is all in the low bin
low_bin = last_run;
} else {
// last run is all in the high bin
high_bin = last_run;
}
p = bin;
while p != last_run {
// safety: p is a valid pointer.
//
// p is only dropped in the next epoch following when its bin is replaced
// with a move node (see safety comment near table.store_bin below). we
// read the bin, and got to p, so its bin has not yet been swapped with a
// move node. and, we have the epoch pinned, so the next epoch cannot have
// arrived yet. therefore, it will be dropped in a future epoch, and is
// safe to use now.
let node = unsafe { p.deref() }.as_node().unwrap();
let link = if node.hash & n as u64 == 0 {
// to the low bin!
&mut low_bin
} else {
// to the high bin!
&mut high_bin
};
*link = Owned::new(BinEntry::Node(Node {
hash: node.hash,
key: node.key.clone(),
lock: parking_lot::Mutex::new(()),
value: node.value.clone(),
next: Atomic::from(*link),
}))
.into_shared(guard);
p = node.next.load(Ordering::SeqCst, guard);
}
next_table.store_bin(i, low_bin);
next_table.store_bin(i + n, high_bin);
table.store_bin(
i,
table.get_moved(Shared::from(next_table as *const _), guard),
);
// everything up to last_run in the _old_ bin linked list is now garbage.
// those nodes have all been re-allocated in the new bin linked list.
p = bin;
while p != last_run {
// safety:
//
// we need to argue that there is no longer a way to access p. the only way
// to get to p is through table[i]. since table[i] has been replaced by a
// BinEntry::Moved, p is no longer accessible.
//
// any existing reference to p must have been taken before table.store_bin.
// at that time we had the epoch pinned, so any threads that have such a
// reference must be before or at our epoch. since the p isn't destroyed
// until the next epoch, those old references are fine since they are tied
// to those old threads' pins of the old epoch.
let next = unsafe { p.deref() }
.as_node()
.unwrap()
.next
.load(Ordering::SeqCst, guard);
unsafe { guard.defer_destroy(p) };
p = next;
}
advance = true;
drop(head_lock);
}
}
}
}
fn help_transfer<'g>(
&'g self,
table: Shared<'g, Table<K, V>>,
guard: &'g Guard,
) -> Shared<'g, Table<K, V>> {
if table.is_null() {
return table;
}
// safety: table is only dropped on the next epoch change after it is swapped to null.
// we read it as not null, so it must not be dropped until a subsequent epoch. since we
// held `guard` at the time, we know that the current epoch continues to persist, and that
// our reference is therefore valid.
let next_table = unsafe { table.deref() }.next_table(guard);
if next_table.is_null() {
return table;
}
// safety: same as above
let rs = Self::resize_stamp(unsafe { table.deref() }.len()) << RESIZE_STAMP_SHIFT;
while next_table == self.next_table.load(Ordering::SeqCst, guard)
&& table == self.table.load(Ordering::SeqCst, guard)
{
let sc = self.size_ctl.load(Ordering::SeqCst);
if sc >= 0
|| sc == rs + MAX_RESIZERS
|| sc == rs + 1
|| self.transfer_index.load(Ordering::SeqCst) <= 0
{
break;
}
if self.size_ctl.compare_and_swap(sc, sc + 1, Ordering::SeqCst) == sc {
self.transfer(table, next_table, guard);
break;
}
}
next_table
}
fn add_count(&self, n: isize, resize_hint: Option<usize>, guard: &Guard) {
// TODO: implement the Java CounterCell business here
use std::cmp;
let mut count = match n.cmp(&0) {
cmp::Ordering::Greater => {
let n = n as usize;
self.count.fetch_add(n, Ordering::SeqCst) + n
}
cmp::Ordering::Less => {
let n = n.abs() as usize;
self.count.fetch_sub(n, Ordering::SeqCst) - n
}
cmp::Ordering::Equal => self.count.load(Ordering::SeqCst),
};
// if resize_hint is None, it means the caller does not want us to consider a resize.
// if it is Some(n), the caller saw n entries in a bin
if resize_hint.is_none() {
return;
}
// TODO: use the resize hint
let _saw_bin_length = resize_hint.unwrap();
loop {
let sc = self.size_ctl.load(Ordering::SeqCst);
if (count as isize) < sc {
// we're not at the next resize point yet
break;
}
let table = self.table.load(Ordering::SeqCst, guard);
if table.is_null() {
// table will be initalized by another thread anyway
break;
}
// safety: table is only dropped on the next epoch change after it is swapped to null.
// we read it as not null, so it must not be dropped until a subsequent epoch. since we
// hold a Guard, we know that the current epoch will persist, and that our reference
// will therefore remain valid.
let n = unsafe { table.deref() }.len();
if n >= MAXIMUM_CAPACITY {
// can't resize any more anyway
break;
}
let rs = Self::resize_stamp(n) << RESIZE_STAMP_SHIFT;
if sc < 0 {
// ongoing resize! can we join the resize transfer?
if sc == rs + MAX_RESIZERS || sc == rs + 1 {
break;