-
Notifications
You must be signed in to change notification settings - Fork 1k
/
concurrent_hash_map.h
1664 lines (1459 loc) · 69.9 KB
/
concurrent_hash_map.h
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
/*
Copyright (c) 2005-2022 Intel Corporation
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
#ifndef __TBB_concurrent_hash_map_H
#define __TBB_concurrent_hash_map_H
#include "detail/_namespace_injection.h"
#include "detail/_utils.h"
#include "detail/_assert.h"
#include "detail/_allocator_traits.h"
#include "detail/_containers_helpers.h"
#include "detail/_template_helpers.h"
#include "detail/_hash_compare.h"
#include "detail/_range_common.h"
#include "tbb_allocator.h"
#include "spin_rw_mutex.h"
#include <atomic>
#include <initializer_list>
#include <tuple>
#include <iterator>
#include <utility> // Need std::pair
#include <cstring> // Need std::memset
namespace tbb {
namespace detail {
namespace d2 {
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS && __TBB_CPP20_CONCEPTS_PRESENT
template <typename Mutex>
concept ch_map_rw_scoped_lockable = rw_scoped_lockable<Mutex> &&
requires(const typename Mutex::scoped_lock& sl) {
{ sl.is_writer() } -> std::convertible_to<bool>;
};
#endif
template <typename MutexType>
struct hash_map_node_base : no_copy {
using mutex_type = MutexType;
// Scoped lock type for mutex
using scoped_type = typename MutexType::scoped_lock;
// Next node in chain
hash_map_node_base* next;
mutex_type mutex;
};
// Incompleteness flag value
static void* const rehash_req_flag = reinterpret_cast<void*>(std::size_t(3));
// Rehashed empty bucket flag
static void* const empty_rehashed_flag = reinterpret_cast<void*>(std::size_t(0));
template <typename MutexType>
bool rehash_required( hash_map_node_base<MutexType>* node_ptr ) {
return reinterpret_cast<void*>(node_ptr) == rehash_req_flag;
}
#if TBB_USE_ASSERT
template <typename MutexType>
bool empty_rehashed( hash_map_node_base<MutexType>* node_ptr ) {
return reinterpret_cast<void*>(node_ptr) == empty_rehashed_flag;
}
#endif
// base class of concurrent_hash_map
template <typename Allocator, typename MutexType>
class hash_map_base {
public:
using size_type = std::size_t;
using hashcode_type = std::size_t;
using segment_index_type = std::size_t;
using node_base = hash_map_node_base<MutexType>;
struct bucket : no_copy {
using mutex_type = MutexType;
using scoped_type = typename mutex_type::scoped_lock;
bucket() : node_list(nullptr) {}
bucket( node_base* ptr ) : node_list(ptr) {}
mutex_type mutex;
std::atomic<node_base*> node_list;
};
using allocator_type = Allocator;
using allocator_traits_type = tbb::detail::allocator_traits<allocator_type>;
using bucket_allocator_type = typename allocator_traits_type::template rebind_alloc<bucket>;
using bucket_allocator_traits = tbb::detail::allocator_traits<bucket_allocator_type>;
// Count of segments in the first block
static constexpr size_type embedded_block = 1;
// Count of segments in the first block
static constexpr size_type embedded_buckets = 1 << embedded_block;
// Count of segments in the first block
static constexpr size_type first_block = 8; //including embedded_block. perfect with bucket size 16, so the allocations are power of 4096
// Size of a pointer / table size
static constexpr size_type pointers_per_table = sizeof(segment_index_type) * 8; // one segment per bit
using segment_ptr_type = bucket*;
using atomic_segment_type = std::atomic<segment_ptr_type>;
using segments_table_type = atomic_segment_type[pointers_per_table];
hash_map_base( const allocator_type& alloc ) : my_allocator(alloc), my_mask(embedded_buckets - 1), my_size(0) {
for (size_type i = 0; i != embedded_buckets; ++i) {
my_embedded_segment[i].node_list.store(nullptr, std::memory_order_relaxed);
}
for (size_type segment_index = 0; segment_index < pointers_per_table; ++segment_index) {
auto argument = segment_index < embedded_block ? my_embedded_segment + segment_base(segment_index) : nullptr;
my_table[segment_index].store(argument, std::memory_order_relaxed);
}
__TBB_ASSERT( embedded_block <= first_block, "The first block number must include embedded blocks");
}
// segment index of given index in the array
static segment_index_type segment_index_of( size_type index ) {
return segment_index_type(tbb::detail::log2( index|1 ));
}
// the first array index of given segment
static segment_index_type segment_base( segment_index_type k ) {
return (segment_index_type(1) << k & ~segment_index_type(1));
}
// segment size except for k == 0
static size_type segment_size( segment_index_type k ) {
return size_type(1) << k; // fake value for k==0
}
// true if ptr is valid pointer
static bool is_valid( void* ptr ) {
return reinterpret_cast<uintptr_t>(ptr) > uintptr_t(63);
}
template <typename... Args>
void init_buckets_impl( segment_ptr_type ptr, size_type sz, const Args&... args ) {
for (size_type i = 0; i < sz; ++i) {
bucket_allocator_traits::construct(my_allocator, ptr + i, args...);
}
}
// Initialize buckets
void init_buckets( segment_ptr_type ptr, size_type sz, bool is_initial ) {
if (is_initial) {
init_buckets_impl(ptr, sz);
} else {
init_buckets_impl(ptr, sz, reinterpret_cast<node_base*>(rehash_req_flag));
}
}
// Add node n to bucket b
static void add_to_bucket( bucket* b, node_base* n ) {
__TBB_ASSERT(!rehash_required(b->node_list.load(std::memory_order_relaxed)), nullptr);
n->next = b->node_list.load(std::memory_order_relaxed);
b->node_list.store(n, std::memory_order_relaxed); // its under lock and flag is set
}
const bucket_allocator_type& get_allocator() const {
return my_allocator;
}
bucket_allocator_type& get_allocator() {
return my_allocator;
}
// Enable segment
void enable_segment( segment_index_type k, bool is_initial = false ) {
__TBB_ASSERT( k, "Zero segment must be embedded" );
size_type sz;
__TBB_ASSERT( !is_valid(my_table[k].load(std::memory_order_relaxed)), "Wrong concurrent assignment");
if (k >= first_block) {
sz = segment_size(k);
segment_ptr_type ptr = nullptr;
try_call( [&] {
ptr = bucket_allocator_traits::allocate(my_allocator, sz);
} ).on_exception( [&] {
my_table[k].store(nullptr, std::memory_order_relaxed);
});
__TBB_ASSERT(ptr, nullptr);
init_buckets(ptr, sz, is_initial);
my_table[k].store(ptr, std::memory_order_release);
sz <<= 1;// double it to get entire capacity of the container
} else { // the first block
__TBB_ASSERT( k == embedded_block, "Wrong segment index" );
sz = segment_size(first_block);
segment_ptr_type ptr = nullptr;
try_call( [&] {
ptr = bucket_allocator_traits::allocate(my_allocator, sz - embedded_buckets);
} ).on_exception( [&] {
my_table[k].store(nullptr, std::memory_order_relaxed);
});
__TBB_ASSERT(ptr, nullptr);
init_buckets(ptr, sz - embedded_buckets, is_initial);
ptr -= segment_base(embedded_block);
for(segment_index_type i = embedded_block; i < first_block; i++) // calc the offsets
my_table[i].store(ptr + segment_base(i), std::memory_order_release);
}
my_mask.store(sz-1, std::memory_order_release);
}
void delete_segment( segment_index_type s ) {
segment_ptr_type buckets_ptr = my_table[s].load(std::memory_order_relaxed);
size_type sz = segment_size( s ? s : 1 );
size_type deallocate_size = 0;
if (s >= first_block) { // the first segment or the next
deallocate_size = sz;
} else if (s == embedded_block && embedded_block != first_block) {
deallocate_size = segment_size(first_block) - embedded_buckets;
}
for (size_type i = 0; i < deallocate_size; ++i) {
bucket_allocator_traits::destroy(my_allocator, buckets_ptr + i);
}
if (deallocate_size != 0) {
bucket_allocator_traits::deallocate(my_allocator, buckets_ptr, deallocate_size);
}
if (s >= embedded_block) my_table[s].store(nullptr, std::memory_order_relaxed);
}
// Get bucket by (masked) hashcode
bucket *get_bucket( hashcode_type h ) const noexcept {
segment_index_type s = segment_index_of( h );
h -= segment_base(s);
segment_ptr_type seg = my_table[s].load(std::memory_order_acquire);
__TBB_ASSERT( is_valid(seg), "hashcode must be cut by valid mask for allocated segments" );
return &seg[h];
}
// detail serial rehashing helper
void mark_rehashed_levels( hashcode_type h ) noexcept {
segment_index_type s = segment_index_of( h );
while (segment_ptr_type seg = my_table[++s].load(std::memory_order_relaxed))
if (rehash_required(seg[h].node_list.load(std::memory_order_relaxed))) {
seg[h].node_list.store(reinterpret_cast<node_base*>(empty_rehashed_flag), std::memory_order_relaxed);
mark_rehashed_levels( h + ((hashcode_type)1<<s) ); // optimized segment_base(s)
}
}
// Check for mask race
// Splitting into two functions should help inlining
inline bool check_mask_race( const hashcode_type h, hashcode_type &m ) const {
hashcode_type m_now, m_old = m;
m_now = my_mask.load(std::memory_order_acquire);
if (m_old != m_now) {
return check_rehashing_collision(h, m_old, m = m_now);
}
return false;
}
// Process mask race, check for rehashing collision
bool check_rehashing_collision( const hashcode_type h, hashcode_type m_old, hashcode_type m ) const {
__TBB_ASSERT(m_old != m, nullptr); // TODO?: m arg could be optimized out by passing h = h&m
if( (h & m_old) != (h & m) ) { // mask changed for this hashcode, rare event
// condition above proves that 'h' has some other bits set beside 'm_old'
// find next applicable mask after m_old //TODO: look at bsl instruction
for( ++m_old; !(h & m_old); m_old <<= 1 ) // at maximum few rounds depending on the first block size
;
m_old = (m_old<<1) - 1; // get full mask from a bit
__TBB_ASSERT((m_old&(m_old+1))==0 && m_old <= m, nullptr);
// check whether it is rehashing/ed
if (!rehash_required(get_bucket(h & m_old)->node_list.load(std::memory_order_acquire))) {
return true;
}
}
return false;
}
// Insert a node and check for load factor. @return segment index to enable.
segment_index_type insert_new_node( bucket *b, node_base *n, hashcode_type mask ) {
size_type sz = ++my_size; // prefix form is to enforce allocation after the first item inserted
add_to_bucket( b, n );
// check load factor
if( sz >= mask ) { // TODO: add custom load_factor
segment_index_type new_seg = tbb::detail::log2( mask+1 ); //optimized segment_index_of
__TBB_ASSERT( is_valid(my_table[new_seg-1].load(std::memory_order_relaxed)), "new allocations must not publish new mask until segment has allocated");
static const segment_ptr_type is_allocating = segment_ptr_type(2);
segment_ptr_type disabled = nullptr;
if (!(my_table[new_seg].load(std::memory_order_acquire))
&& my_table[new_seg].compare_exchange_strong(disabled, is_allocating))
return new_seg; // The value must be processed
}
return 0;
}
// Prepare enough segments for number of buckets
void reserve(size_type buckets) {
if( !buckets-- ) return;
bool is_initial = !my_size.load(std::memory_order_relaxed);
for (size_type m = my_mask.load(std::memory_order_relaxed); buckets > m;
m = my_mask.load(std::memory_order_relaxed))
{
enable_segment( segment_index_of( m+1 ), is_initial );
}
}
// Swap hash_map_bases
void internal_swap_content(hash_map_base &table) {
using std::swap;
swap_atomics_relaxed(my_mask, table.my_mask);
swap_atomics_relaxed(my_size, table.my_size);
for(size_type i = 0; i < embedded_buckets; i++) {
auto temp = my_embedded_segment[i].node_list.load(std::memory_order_relaxed);
my_embedded_segment[i].node_list.store(table.my_embedded_segment[i].node_list.load(std::memory_order_relaxed),
std::memory_order_relaxed);
table.my_embedded_segment[i].node_list.store(temp, std::memory_order_relaxed);
}
for(size_type i = embedded_block; i < pointers_per_table; i++) {
auto temp = my_table[i].load(std::memory_order_relaxed);
my_table[i].store(table.my_table[i].load(std::memory_order_relaxed),
std::memory_order_relaxed);
table.my_table[i].store(temp, std::memory_order_relaxed);
}
}
void internal_move(hash_map_base&& other) {
my_mask.store(other.my_mask.load(std::memory_order_relaxed), std::memory_order_relaxed);
other.my_mask.store(embedded_buckets - 1, std::memory_order_relaxed);
my_size.store(other.my_size.load(std::memory_order_relaxed), std::memory_order_relaxed);
other.my_size.store(0, std::memory_order_relaxed);
for (size_type i = 0; i < embedded_buckets; ++i) {
my_embedded_segment[i].node_list.store(other.my_embedded_segment[i].node_list, std::memory_order_relaxed);
other.my_embedded_segment[i].node_list.store(nullptr, std::memory_order_relaxed);
}
for (size_type i = embedded_block; i < pointers_per_table; ++i) {
my_table[i].store(other.my_table[i].load(std::memory_order_relaxed),
std::memory_order_relaxed);
other.my_table[i].store(nullptr, std::memory_order_relaxed);
}
}
protected:
bucket_allocator_type my_allocator;
// Hash mask = sum of allocated segment sizes - 1
std::atomic<hashcode_type> my_mask;
// Size of container in stored items
std::atomic<size_type> my_size; // It must be in separate cache line from my_mask due to performance effects
// Zero segment
bucket my_embedded_segment[embedded_buckets];
// Segment pointers table. Also prevents false sharing between my_mask and my_size
segments_table_type my_table;
};
template <typename Iterator>
class hash_map_range;
// Meets requirements of a forward iterator for STL
// Value is either the T or const T type of the container.
template <typename Container, typename Value>
class hash_map_iterator {
using map_type = Container;
using node = typename Container::node;
using map_base = typename Container::base_type;
using node_base = typename map_base::node_base;
using bucket = typename map_base::bucket;
public:
using value_type = Value;
using size_type = typename Container::size_type;
using difference_type = typename Container::difference_type;
using pointer = value_type*;
using reference = value_type&;
using iterator_category = std::forward_iterator_tag;
// Construct undefined iterator
hash_map_iterator(): my_map(), my_index(), my_bucket(), my_node() {}
hash_map_iterator( const hash_map_iterator<Container, typename Container::value_type>& other ) :
my_map(other.my_map),
my_index(other.my_index),
my_bucket(other.my_bucket),
my_node(other.my_node)
{}
hash_map_iterator& operator=( const hash_map_iterator<Container, typename Container::value_type>& other ) {
my_map = other.my_map;
my_index = other.my_index;
my_bucket = other.my_bucket;
my_node = other.my_node;
return *this;
}
Value& operator*() const {
__TBB_ASSERT( map_base::is_valid(my_node), "iterator uninitialized or at end of container?" );
return my_node->value();
}
Value* operator->() const {return &operator*();}
hash_map_iterator& operator++() {
my_node = static_cast<node*>( my_node->next );
if( !my_node ) advance_to_next_bucket();
return *this;
}
// Post increment
hash_map_iterator operator++(int) {
hash_map_iterator old(*this);
operator++();
return old;
}
private:
template <typename C, typename T, typename U>
friend bool operator==( const hash_map_iterator<C,T>& i, const hash_map_iterator<C,U>& j );
template <typename C, typename T, typename U>
friend bool operator!=( const hash_map_iterator<C,T>& i, const hash_map_iterator<C,U>& j );
template <typename C, typename T, typename U>
friend ptrdiff_t operator-( const hash_map_iterator<C,T>& i, const hash_map_iterator<C,U>& j );
template <typename C, typename U>
friend class hash_map_iterator;
template <typename I>
friend class hash_map_range;
void advance_to_next_bucket() { // TODO?: refactor to iterator_base class
size_t k = my_index+1;
__TBB_ASSERT( my_bucket, "advancing an invalid iterator?");
while (k <= my_map->my_mask.load(std::memory_order_relaxed)) {
// Following test uses 2's-complement wizardry
if( k&(k-2) ) // not the beginning of a segment
++my_bucket;
else my_bucket = my_map->get_bucket( k );
node_base *n = my_bucket->node_list.load(std::memory_order_relaxed);
if( map_base::is_valid(n) ) {
my_node = static_cast<node*>(n);
my_index = k;
return;
}
++k;
}
my_bucket = nullptr; my_node = nullptr; my_index = k; // the end
}
template <typename Key, typename T, typename HashCompare, typename A
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS
, typename M
>
__TBB_requires(tbb::detail::hash_compare<HashCompare, Key> &&
ch_map_rw_scoped_lockable<M>)
#else
>
__TBB_requires(tbb::detail::hash_compare<HashCompare, Key>)
#endif
friend class concurrent_hash_map;
hash_map_iterator( const Container &map, std::size_t index, const bucket *b, node_base *n ) :
my_map(&map), my_index(index), my_bucket(b), my_node(static_cast<node*>(n))
{
if( b && !map_base::is_valid(n) )
advance_to_next_bucket();
}
// concurrent_hash_map over which we are iterating.
const Container *my_map;
// Index in hash table for current item
size_t my_index;
// Pointer to bucket
const bucket* my_bucket;
// Pointer to node that has current item
node* my_node;
};
template <typename Container, typename T, typename U>
bool operator==( const hash_map_iterator<Container,T>& i, const hash_map_iterator<Container,U>& j ) {
return i.my_node == j.my_node && i.my_map == j.my_map;
}
template <typename Container, typename T, typename U>
bool operator!=( const hash_map_iterator<Container,T>& i, const hash_map_iterator<Container,U>& j ) {
return i.my_node != j.my_node || i.my_map != j.my_map;
}
// Range class used with concurrent_hash_map
template <typename Iterator>
class hash_map_range {
using map_type = typename Iterator::map_type;
public:
// Type for size of a range
using size_type = std::size_t;
using value_type = typename Iterator::value_type;
using reference = typename Iterator::reference;
using difference_type = typename Iterator::difference_type;
using iterator = Iterator;
// True if range is empty.
bool empty() const { return my_begin == my_end; }
// True if range can be partitioned into two subranges.
bool is_divisible() const {
return my_midpoint != my_end;
}
// Split range.
hash_map_range( hash_map_range& r, split ) :
my_end(r.my_end),
my_grainsize(r.my_grainsize)
{
r.my_end = my_begin = r.my_midpoint;
__TBB_ASSERT( !empty(), "Splitting despite the range is not divisible" );
__TBB_ASSERT( !r.empty(), "Splitting despite the range is not divisible" );
set_midpoint();
r.set_midpoint();
}
// Init range with container and grainsize specified
hash_map_range( const map_type &map, size_type grainsize_ = 1 ) :
my_begin( Iterator( map, 0, map.my_embedded_segment, map.my_embedded_segment->node_list.load(std::memory_order_relaxed) ) ),
my_end( Iterator( map, map.my_mask.load(std::memory_order_relaxed) + 1, nullptr, nullptr ) ),
my_grainsize( grainsize_ )
{
__TBB_ASSERT( grainsize_>0, "grainsize must be positive" );
set_midpoint();
}
Iterator begin() const { return my_begin; }
Iterator end() const { return my_end; }
// The grain size for this range.
size_type grainsize() const { return my_grainsize; }
private:
Iterator my_begin;
Iterator my_end;
mutable Iterator my_midpoint;
size_t my_grainsize;
// Set my_midpoint to point approximately half way between my_begin and my_end.
void set_midpoint() const;
template <typename U> friend class hash_map_range;
};
template <typename Iterator>
void hash_map_range<Iterator>::set_midpoint() const {
// Split by groups of nodes
size_t m = my_end.my_index-my_begin.my_index;
if( m > my_grainsize ) {
m = my_begin.my_index + m/2u;
auto b = my_begin.my_map->get_bucket(m);
my_midpoint = Iterator(*my_begin.my_map,m,b,b->node_list.load(std::memory_order_relaxed));
} else {
my_midpoint = my_end;
}
__TBB_ASSERT( my_begin.my_index <= my_midpoint.my_index,
"my_begin is after my_midpoint" );
__TBB_ASSERT( my_midpoint.my_index <= my_end.my_index,
"my_midpoint is after my_end" );
__TBB_ASSERT( my_begin != my_midpoint || my_begin == my_end,
"[my_begin, my_midpoint) range should not be empty" );
}
template <typename Key, typename T,
typename HashCompare = d1::tbb_hash_compare<Key>,
typename Allocator = tbb_allocator<std::pair<const Key, T>>
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS
, typename MutexType = spin_rw_mutex
>
__TBB_requires(tbb::detail::hash_compare<HashCompare, Key> &&
ch_map_rw_scoped_lockable<MutexType>)
#else
>
__TBB_requires(tbb::detail::hash_compare<HashCompare, Key>)
#endif
class concurrent_hash_map
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS
: protected hash_map_base<Allocator, MutexType>
#else
: protected hash_map_base<Allocator, spin_rw_mutex>
#endif
{
template <typename Container, typename Value>
friend class hash_map_iterator;
template <typename I>
friend class hash_map_range;
using allocator_traits_type = tbb::detail::allocator_traits<Allocator>;
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS
using base_type = hash_map_base<Allocator, MutexType>;
#else
using base_type = hash_map_base<Allocator, spin_rw_mutex>;
#endif
public:
using key_type = Key;
using mapped_type = T;
// type_identity is needed to disable implicit deduction guides for std::initializer_list constructors
// and copy/move constructor with explicit allocator argument
using allocator_type = tbb::detail::type_identity_t<Allocator>;
using hash_compare_type = tbb::detail::type_identity_t<HashCompare>;
using value_type = std::pair<const Key, T>;
using size_type = typename base_type::size_type;
using difference_type = std::ptrdiff_t;
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS
using mutex_type = MutexType;
#endif
using pointer = typename allocator_traits_type::pointer;
using const_pointer = typename allocator_traits_type::const_pointer;
using reference = value_type&;
using const_reference = const value_type&;
using iterator = hash_map_iterator<concurrent_hash_map, value_type>;
using const_iterator = hash_map_iterator<concurrent_hash_map, const value_type>;
using range_type = hash_map_range<iterator>;
using const_range_type = hash_map_range<const_iterator>;
protected:
static_assert(std::is_same<value_type, typename Allocator::value_type>::value,
"value_type of the container must be the same as its allocator's");
friend class const_accessor;
class node;
using segment_index_type = typename base_type::segment_index_type;
using segment_ptr_type = typename base_type::segment_ptr_type;
using node_base = typename base_type::node_base;
using bucket = typename base_type::bucket;
using hashcode_type = typename base_type::hashcode_type;
using bucket_allocator_type = typename base_type::bucket_allocator_type;
using node_allocator_type = typename base_type::allocator_traits_type::template rebind_alloc<node>;
using node_allocator_traits = tbb::detail::allocator_traits<node_allocator_type>;
hash_compare_type my_hash_compare;
class node : public node_base {
public:
node() {}
~node() {}
pointer storage() { return &my_value; }
value_type& value() { return *storage(); }
private:
union {
value_type my_value;
};
};
void delete_node( node_base *n ) {
node_allocator_type node_allocator(this->get_allocator());
node_allocator_traits::destroy(node_allocator, static_cast<node*>(n)->storage());
node_allocator_traits::destroy(node_allocator, static_cast<node*>(n));
node_allocator_traits::deallocate(node_allocator, static_cast<node*>(n), 1);
}
template <typename... Args>
static node* create_node(bucket_allocator_type& allocator, Args&&... args) {
node_allocator_type node_allocator(allocator);
node* node_ptr = node_allocator_traits::allocate(node_allocator, 1);
auto guard = make_raii_guard([&] {
node_allocator_traits::destroy(node_allocator, node_ptr);
node_allocator_traits::deallocate(node_allocator, node_ptr, 1);
});
node_allocator_traits::construct(node_allocator, node_ptr);
node_allocator_traits::construct(node_allocator, node_ptr->storage(), std::forward<Args>(args)...);
guard.dismiss();
return node_ptr;
}
static node* allocate_node_copy_construct(bucket_allocator_type& allocator, const Key &key, const T * t){
return create_node(allocator, key, *t);
}
static node* allocate_node_move_construct(bucket_allocator_type& allocator, const Key &key, const T * t){
return create_node(allocator, key, std::move(*const_cast<T*>(t)));
}
template <typename K = Key>
static node* allocate_node_default_construct(bucket_allocator_type& allocator, const K &key, const T * ){
// Emplace construct an empty T object inside the pair
return create_node(allocator, std::piecewise_construct,
std::forward_as_tuple(key), std::forward_as_tuple());
}
static node* do_not_allocate_node(bucket_allocator_type& , const Key &, const T * ){
__TBB_ASSERT(false,"this dummy function should not be called");
return nullptr;
}
template <typename K>
node *search_bucket( const K &key, bucket *b ) const {
node *n = static_cast<node*>( b->node_list.load(std::memory_order_relaxed) );
while (this->is_valid(n) && !my_hash_compare.equal(key, n->value().first))
n = static_cast<node*>( n->next );
__TBB_ASSERT(!rehash_required(n), "Search can be executed only for rehashed bucket");
return n;
}
// bucket accessor is to find, rehash, acquire a lock, and access a bucket
class bucket_accessor : public bucket::scoped_type {
bucket *my_b;
public:
bucket_accessor( concurrent_hash_map *base, const hashcode_type h, bool writer = false ) { acquire( base, h, writer ); }
// find a bucket by masked hashcode, optionally rehash, and acquire the lock
inline void acquire( concurrent_hash_map *base, const hashcode_type h, bool writer = false ) {
my_b = base->get_bucket( h );
// TODO: actually, notification is unnecessary here, just hiding double-check
if (rehash_required(my_b->node_list.load(std::memory_order_acquire))
&& bucket::scoped_type::try_acquire( my_b->mutex, /*write=*/true ) )
{
if (rehash_required(my_b->node_list.load(std::memory_order_relaxed))) base->rehash_bucket(my_b, h); // recursive rehashing
}
else bucket::scoped_type::acquire( my_b->mutex, writer );
__TBB_ASSERT(!rehash_required(my_b->node_list.load(std::memory_order_relaxed)), nullptr);
}
// get bucket pointer
bucket *operator() () { return my_b; }
};
// TODO refactor to hash_base
void rehash_bucket( bucket *b_new, const hashcode_type hash ) {
__TBB_ASSERT( hash > 1, "The lowermost buckets can't be rehashed" );
b_new->node_list.store(reinterpret_cast<node_base*>(empty_rehashed_flag), std::memory_order_release); // mark rehashed
hashcode_type mask = (hashcode_type(1) << tbb::detail::log2(hash)) - 1; // get parent mask from the topmost bit
bucket_accessor b_old( this, hash & mask );
mask = (mask<<1) | 1; // get full mask for new bucket
__TBB_ASSERT( (mask&(mask+1))==0 && (hash & mask) == hash, nullptr );
restart:
node_base* prev = nullptr;
node_base* curr = b_old()->node_list.load(std::memory_order_acquire);
while (this->is_valid(curr)) {
hashcode_type curr_node_hash = my_hash_compare.hash(static_cast<node*>(curr)->value().first);
if ((curr_node_hash & mask) == hash) {
if (!b_old.is_writer()) {
if (!b_old.upgrade_to_writer()) {
goto restart; // node ptr can be invalid due to concurrent erase
}
}
node_base* next = curr->next;
// exclude from b_old
if (prev == nullptr) {
b_old()->node_list.store(curr->next, std::memory_order_relaxed);
} else {
prev->next = curr->next;
}
this->add_to_bucket(b_new, curr);
curr = next;
} else {
prev = curr;
curr = curr->next;
}
}
}
template <typename U>
using hash_compare_is_transparent = dependent_bool<comp_is_transparent<hash_compare_type>, U>;
public:
class accessor;
// Combines data access, locking, and garbage collection.
class const_accessor : private node::scoped_type /*which derived from no_copy*/ {
#if __TBB_PREVIEW_CONCURRENT_HASH_MAP_EXTENSIONS
friend class concurrent_hash_map<Key,T,HashCompare,Allocator,MutexType>;
#else
friend class concurrent_hash_map<Key,T,HashCompare,Allocator>;
#endif
friend class accessor;
public:
// Type of value
using value_type = const typename concurrent_hash_map::value_type;
// True if result is empty.
bool empty() const { return !my_node; }
// Set to null
void release() {
if( my_node ) {
node::scoped_type::release();
my_node = nullptr;
}
}
// Return reference to associated value in hash table.
const_reference operator*() const {
__TBB_ASSERT( my_node, "attempt to dereference empty accessor" );
return my_node->value();
}
// Return pointer to associated value in hash table.
const_pointer operator->() const {
return &operator*();
}
// Create empty result
const_accessor() : my_node(nullptr), my_hash() {}
// Destroy result after releasing the underlying reference.
~const_accessor() {
my_node = nullptr; // scoped lock's release() is called in its destructor
}
protected:
bool is_writer() { return node::scoped_type::is_writer(); }
node *my_node;
hashcode_type my_hash;
};
// Allows write access to elements and combines data access, locking, and garbage collection.
class accessor: public const_accessor {
public:
// Type of value
using value_type = typename concurrent_hash_map::value_type;
// Return reference to associated value in hash table.
reference operator*() const {
__TBB_ASSERT( this->my_node, "attempt to dereference empty accessor" );
return this->my_node->value();
}
// Return pointer to associated value in hash table.
pointer operator->() const {
return &operator*();
}
};
explicit concurrent_hash_map( const hash_compare_type& compare, const allocator_type& a = allocator_type() )
: base_type(a)
, my_hash_compare(compare)
{}
concurrent_hash_map() : concurrent_hash_map(hash_compare_type()) {}
explicit concurrent_hash_map( const allocator_type& a )
: concurrent_hash_map(hash_compare_type(), a)
{}
// Construct empty table with n preallocated buckets. This number serves also as initial concurrency level.
concurrent_hash_map( size_type n, const allocator_type &a = allocator_type() )
: concurrent_hash_map(a)
{
this->reserve(n);
}
concurrent_hash_map( size_type n, const hash_compare_type& compare, const allocator_type& a = allocator_type() )
: concurrent_hash_map(compare, a)
{
this->reserve(n);
}
// Copy constructor
concurrent_hash_map( const concurrent_hash_map &table )
: concurrent_hash_map(node_allocator_traits::select_on_container_copy_construction(table.get_allocator()))
{
try_call( [&] {
internal_copy(table);
}).on_exception( [&] {
this->clear();
});
}
concurrent_hash_map( const concurrent_hash_map &table, const allocator_type &a)
: concurrent_hash_map(a)
{
try_call( [&] {
internal_copy(table);
}).on_exception( [&] {
this->clear();
});
}
// Move constructor
concurrent_hash_map( concurrent_hash_map &&table )
: concurrent_hash_map(std::move(table.get_allocator()))
{
this->internal_move(std::move(table));
}
// Move constructor
concurrent_hash_map( concurrent_hash_map &&table, const allocator_type &a )
: concurrent_hash_map(a)
{
using is_equal_type = typename node_allocator_traits::is_always_equal;
internal_move_construct_with_allocator(std::move(table), a, is_equal_type());
}
// Construction with copying iteration range and given allocator instance
template <typename I>
concurrent_hash_map( I first, I last, const allocator_type &a = allocator_type() )
: concurrent_hash_map(a)
{
try_call( [&] {
internal_copy(first, last, std::distance(first, last));
}).on_exception( [&] {
this->clear();
});
}
template <typename I>
concurrent_hash_map( I first, I last, const hash_compare_type& compare, const allocator_type& a = allocator_type() )
: concurrent_hash_map(compare, a)
{
try_call( [&] {
internal_copy(first, last, std::distance(first, last));
}).on_exception( [&] {
this->clear();
});
}
concurrent_hash_map( std::initializer_list<value_type> il, const hash_compare_type& compare = hash_compare_type(), const allocator_type& a = allocator_type() )
: concurrent_hash_map(compare, a)
{
try_call( [&] {
internal_copy(il.begin(), il.end(), il.size());
}).on_exception( [&] {
this->clear();
});
}
concurrent_hash_map( std::initializer_list<value_type> il, const allocator_type& a )
: concurrent_hash_map(il, hash_compare_type(), a) {}
// Assignment
concurrent_hash_map& operator=( const concurrent_hash_map &table ) {
if( this != &table ) {
clear();
copy_assign_allocators(this->my_allocator, table.my_allocator);
internal_copy(table);
}
return *this;
}
// Move Assignment
concurrent_hash_map& operator=( concurrent_hash_map &&table ) {
if( this != &table ) {
using pocma_type = typename node_allocator_traits::propagate_on_container_move_assignment;
using is_equal_type = typename node_allocator_traits::is_always_equal;
move_assign_allocators(this->my_allocator, table.my_allocator);
internal_move_assign(std::move(table), tbb::detail::disjunction<is_equal_type, pocma_type>());
}
return *this;
}
// Assignment
concurrent_hash_map& operator=( std::initializer_list<value_type> il ) {
clear();
internal_copy(il.begin(), il.end(), il.size());
return *this;
}
// Rehashes and optionally resizes the whole table.
/** Useful to optimize performance before or after concurrent operations.
Also enables using of find() and count() concurrent methods in serial context. */
void rehash(size_type sz = 0) {
this->reserve(sz); // TODO: add reduction of number of buckets as well
hashcode_type mask = this->my_mask.load(std::memory_order_relaxed);
hashcode_type b = (mask+1)>>1; // size or first index of the last segment
__TBB_ASSERT((b&(b-1))==0, nullptr); // zero or power of 2
bucket *bp = this->get_bucket( b ); // only the last segment should be scanned for rehashing
for(; b <= mask; b++, bp++ ) {
node_base *n = bp->node_list.load(std::memory_order_relaxed);
__TBB_ASSERT( this->is_valid(n) || empty_rehashed(n) || rehash_required(n), "Broken internal structure" );
__TBB_ASSERT( *reinterpret_cast<intptr_t*>(&bp->mutex) == 0, "concurrent or unexpectedly terminated operation during rehash() execution" );
if (rehash_required(n)) { // rehash bucket, conditional because rehashing of a previous bucket may affect this one
hashcode_type h = b; bucket *b_old = bp;
do {
__TBB_ASSERT( h > 1, "The lowermost buckets can't be rehashed" );
hashcode_type m = ( hashcode_type(1) << tbb::detail::log2( h ) ) - 1; // get parent mask from the topmost bit
b_old = this->get_bucket( h &= m );
} while( rehash_required(b_old->node_list.load(std::memory_order_relaxed)) );
// now h - is index of the root rehashed bucket b_old
this->mark_rehashed_levels( h ); // mark all non-rehashed children recursively across all segments
node_base* prev = nullptr;
node_base* curr = b_old->node_list.load(std::memory_order_relaxed);
while (this->is_valid(curr)) {
hashcode_type curr_node_hash = my_hash_compare.hash(static_cast<node*>(curr)->value().first);
if ((curr_node_hash & mask) != h) { // should be rehashed
node_base* next = curr->next;
// exclude from b_old
if (prev == nullptr) {
b_old->node_list.store(curr->next, std::memory_order_relaxed);
} else {
prev->next = curr->next;
}
bucket *b_new = this->get_bucket(curr_node_hash & mask);
__TBB_ASSERT(!rehash_required(b_new->node_list.load(std::memory_order_relaxed)), "hash() function changed for key in table or internal error");
this->add_to_bucket(b_new, curr);
curr = next;
} else {
prev = curr;
curr = curr->next;