-
Notifications
You must be signed in to change notification settings - Fork 5.5k
/
MPMCQueue.h
1497 lines (1371 loc) · 54.6 KB
/
MPMCQueue.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) Meta Platforms, Inc. and affiliates.
*
* 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.
*/
#pragma once
#include <algorithm>
#include <atomic>
#include <cassert>
#include <cstring>
#include <limits>
#include <type_traits>
#include <folly/Traits.h>
#include <folly/concurrency/CacheLocality.h>
#include <folly/detail/TurnSequencer.h>
#include <folly/portability/Unistd.h>
namespace folly {
namespace detail {
template <typename T, template <typename> class Atom>
struct SingleElementQueue;
template <typename T>
class MPMCPipelineStageImpl;
/// MPMCQueue base CRTP template
template <typename>
class MPMCQueueBase;
} // namespace detail
/// MPMCQueue<T> is a high-performance bounded concurrent queue that
/// supports multiple producers, multiple consumers, and optional blocking.
/// The queue has a fixed capacity, for which all memory will be allocated
/// up front. The bulk of the work of enqueuing and dequeuing can be
/// performed in parallel.
///
/// MPMCQueue is linearizable. That means that if a call to write(A)
/// returns before a call to write(B) begins, then A will definitely end up
/// in the queue before B, and if a call to read(X) returns before a call
/// to read(Y) is started, that X will be something from earlier in the
/// queue than Y. This also means that if a read call returns a value, you
/// can be sure that all previous elements of the queue have been assigned
/// a reader (that reader might not yet have returned, but it exists).
///
/// The underlying implementation uses a ticket dispenser for the head and
/// the tail, spreading accesses across N single-element queues to produce
/// a queue with capacity N. The ticket dispensers use atomic increment,
/// which is more robust to contention than a CAS loop. Each of the
/// single-element queues uses its own CAS to serialize access, with an
/// adaptive spin cutoff. When spinning fails on a single-element queue
/// it uses futex()'s _BITSET operations to reduce unnecessary wakeups
/// even if multiple waiters are present on an individual queue (such as
/// when the MPMCQueue's capacity is smaller than the number of enqueuers
/// or dequeuers).
///
/// In benchmarks (contained in tao/queues/ConcurrentQueueTests)
/// it handles 1 to 1, 1 to N, N to 1, and N to M thread counts better
/// than any of the alternatives present in fbcode, for both small (~10)
/// and large capacities. In these benchmarks it is also faster than
/// tbb::concurrent_bounded_queue for all configurations. When there are
/// many more threads than cores, MPMCQueue is _much_ faster than the tbb
/// queue because it uses futex() to block and unblock waiting threads,
/// rather than spinning with sched_yield.
///
/// NOEXCEPT INTERACTION: tl;dr; If it compiles you're fine. Ticket-based
/// queues separate the assignment of queue positions from the actual
/// construction of the in-queue elements, which means that the T
/// constructor used during enqueue must not throw an exception. This is
/// enforced at compile time using type traits, which requires that T be
/// adorned with accurate noexcept information. If your type does not
/// use noexcept, you will have to wrap it in something that provides
/// the guarantee. We provide an alternate safe implementation for types
/// that don't use noexcept but that are marked folly::IsRelocatable
/// and std::is_nothrow_constructible, which is common for folly types.
/// In particular, if you can declare FOLLY_ASSUME_FBVECTOR_COMPATIBLE
/// then your type can be put in MPMCQueue.
///
/// If you have a pool of N queue consumers that you want to shut down
/// after the queue has drained, one way is to enqueue N sentinel values
/// to the queue. If the producer doesn't know how many consumers there
/// are you can enqueue one sentinel and then have each consumer requeue
/// two sentinels after it receives it (by requeuing 2 the shutdown can
/// complete in O(log P) time instead of O(P)).
template <
typename T,
template <typename> class Atom = std::atomic,
bool Dynamic = false>
class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T, Atom, Dynamic>> {
friend class detail::MPMCPipelineStageImpl<T>;
using Slot = detail::SingleElementQueue<T, Atom>;
public:
explicit MPMCQueue(size_t queueCapacity)
: detail::MPMCQueueBase<MPMCQueue<T, Atom, Dynamic>>(queueCapacity) {
this->stride_ = this->computeStride(queueCapacity);
this->slots_ = new Slot[queueCapacity + 2 * this->kSlotPadding];
}
MPMCQueue() noexcept {}
};
/// *** The dynamic version of MPMCQueue is deprecated. ***
/// Use UnboundedQueue instead.
/// The dynamic version of MPMCQueue allows dynamic expansion of queue
/// capacity, such that a queue may start with a smaller capacity than
/// specified and expand only if needed. Users may optionally specify
/// the initial capacity and the expansion multiplier.
///
/// The design uses a seqlock to enforce mutual exclusion among
/// expansion attempts. Regular operations read up-to-date queue
/// information (slots array, capacity, stride) inside read-only
/// seqlock sections, which are unimpeded when no expansion is in
/// progress.
///
/// An expansion computes a new capacity, allocates a new slots array,
/// and updates stride. No information needs to be copied from the
/// current slots array to the new one. When this happens, new slots
/// will not have sequence numbers that match ticket numbers. The
/// expansion needs to compute a ticket offset such that operations
/// that use new arrays can adjust the calculations of slot indexes
/// and sequence numbers that take into account that the new slots
/// start with sequence numbers of zero. The current ticket offset is
/// packed with the seqlock in an atomic 64-bit integer. The initial
/// offset is zero.
///
/// Lagging write and read operations with tickets lower than the
/// ticket offset of the current slots array (i.e., the minimum ticket
/// number that can be served by the current array) must use earlier
/// closed arrays instead of the current one. Information about closed
/// slots arrays (array address, capacity, stride, and offset) is
/// maintained in a logarithmic-sized structure. Each entry in that
/// structure never needs to be changed once set. The number of closed
/// arrays is half the value of the seqlock (when unlocked).
///
/// The acquisition of the seqlock to perform an expansion does not
/// prevent the issuing of new push and pop tickets concurrently. The
/// expansion must set the new ticket offset to a value that couldn't
/// have been issued to an operation that has already gone through a
/// seqlock read-only section (and hence obtained information for
/// older closed arrays).
///
/// Note that the total queue capacity can temporarily exceed the
/// specified capacity when there are lagging consumers that haven't
/// yet consumed all the elements in closed arrays. Users should not
/// rely on the capacity of dynamic queues for synchronization, e.g.,
/// they should not expect that a thread will definitely block on a
/// call to blockingWrite() when the queue size is known to be equal
/// to its capacity.
///
/// Note that some writeIfNotFull() and tryWriteUntil() operations may
/// fail even if the size of the queue is less than its maximum
/// capacity and despite the success of expansion, if the operation
/// happens to acquire a ticket that belongs to a closed array. This
/// is a transient condition. Typically, one or two ticket values may
/// be subject to such condition per expansion.
///
/// The dynamic version is a partial specialization of MPMCQueue with
/// Dynamic == true
template <typename T, template <typename> class Atom>
class MPMCQueue<T, Atom, true>
: public detail::MPMCQueueBase<MPMCQueue<T, Atom, true>> {
friend class detail::MPMCQueueBase<MPMCQueue<T, Atom, true>>;
using Slot = detail::SingleElementQueue<T, Atom>;
struct ClosedArray {
uint64_t offset_{0};
Slot* slots_{nullptr};
size_t capacity_{0};
int stride_{0};
};
public:
explicit MPMCQueue(size_t queueCapacity)
: detail::MPMCQueueBase<MPMCQueue<T, Atom, true>>(queueCapacity) {
size_t cap = std::min<size_t>(kDefaultMinDynamicCapacity, queueCapacity);
initQueue(cap, kDefaultExpansionMultiplier);
}
explicit MPMCQueue(
size_t queueCapacity, size_t minCapacity, size_t expansionMultiplier)
: detail::MPMCQueueBase<MPMCQueue<T, Atom, true>>(queueCapacity) {
minCapacity = std::max<size_t>(1, minCapacity);
size_t cap = std::min<size_t>(minCapacity, queueCapacity);
expansionMultiplier = std::max<size_t>(2, expansionMultiplier);
initQueue(cap, expansionMultiplier);
}
MPMCQueue() noexcept {
dmult_ = 0;
closed_ = nullptr;
}
MPMCQueue(MPMCQueue<T, Atom, true>&& rhs) noexcept {
this->capacity_ = rhs.capacity_;
new (&this->dslots_)
Atom<Slot*>(rhs.dslots_.load(std::memory_order_relaxed));
new (&this->dstride_)
Atom<int>(rhs.dstride_.load(std::memory_order_relaxed));
this->dstate_.store(
rhs.dstate_.load(std::memory_order_relaxed), std::memory_order_relaxed);
this->dcapacity_.store(
rhs.dcapacity_.load(std::memory_order_relaxed),
std::memory_order_relaxed);
this->pushTicket_.store(
rhs.pushTicket_.load(std::memory_order_relaxed),
std::memory_order_relaxed);
this->popTicket_.store(
rhs.popTicket_.load(std::memory_order_relaxed),
std::memory_order_relaxed);
this->pushSpinCutoff_.store(
rhs.pushSpinCutoff_.load(std::memory_order_relaxed),
std::memory_order_relaxed);
this->popSpinCutoff_.store(
rhs.popSpinCutoff_.load(std::memory_order_relaxed),
std::memory_order_relaxed);
dmult_ = rhs.dmult_;
closed_ = rhs.closed_;
rhs.capacity_ = 0;
rhs.dslots_.store(nullptr, std::memory_order_relaxed);
rhs.dstride_.store(0, std::memory_order_relaxed);
rhs.dstate_.store(0, std::memory_order_relaxed);
rhs.dcapacity_.store(0, std::memory_order_relaxed);
rhs.pushTicket_.store(0, std::memory_order_relaxed);
rhs.popTicket_.store(0, std::memory_order_relaxed);
rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
rhs.dmult_ = 0;
rhs.closed_ = nullptr;
}
MPMCQueue<T, Atom, true> const& operator=(MPMCQueue<T, Atom, true>&& rhs) {
if (this != &rhs) {
this->~MPMCQueue();
new (this) MPMCQueue(std::move(rhs));
}
return *this;
}
~MPMCQueue() {
if (closed_ != nullptr) {
for (int i = getNumClosed(this->dstate_.load()) - 1; i >= 0; --i) {
delete[] closed_[i].slots_;
}
delete[] closed_;
}
using AtomInt = Atom<int>;
this->dstride_.~AtomInt();
using AtomSlot = Atom<Slot*>;
// Sort of a hack to get ~MPMCQueueBase to free dslots_
auto slots = this->dslots_.load();
this->dslots_.~AtomSlot();
this->slots_ = slots;
}
size_t allocatedCapacity() const noexcept {
return this->dcapacity_.load(std::memory_order_relaxed);
}
template <typename... Args>
void blockingWrite(Args&&... args) noexcept {
uint64_t ticket = this->pushTicket_++;
Slot* slots;
size_t cap;
int stride;
uint64_t state;
uint64_t offset;
do {
if (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause();
continue;
}
if (maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride)) {
// There was an expansion after this ticket was issued.
break;
}
if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
this->turn(ticket - offset, cap))) {
// A slot is ready. No need to expand.
break;
} else if (
this->popTicket_.load(std::memory_order_relaxed) + cap > ticket) {
// May block, but a pop is in progress. No need to expand.
// Get seqlock read section info again in case an expansion
// occurred with an equal or higher ticket.
continue;
} else {
// May block. See if we can expand.
if (tryExpand(state, cap)) {
// This or another thread started an expansion. Get updated info.
continue;
} else {
// Can't expand.
break;
}
}
} while (true);
this->enqueueWithTicketBase(
ticket - offset, slots, cap, stride, std::forward<Args>(args)...);
}
void blockingReadWithTicket(uint64_t& ticket, T& elem) noexcept {
ticket = this->popTicket_++;
Slot* slots;
size_t cap;
int stride;
uint64_t state;
uint64_t offset;
while (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause();
}
// If there was an expansion after the corresponding push ticket
// was issued, adjust accordingly
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
this->dequeueWithTicketBase(ticket - offset, slots, cap, stride, elem);
}
private:
enum {
kSeqlockBits = 6,
kDefaultMinDynamicCapacity = 10,
kDefaultExpansionMultiplier = 10,
};
size_t dmult_;
// Info about closed slots arrays for use by lagging operations
ClosedArray* closed_;
void initQueue(const size_t cap, const size_t mult) {
new (&this->dstride_) Atom<int>(this->computeStride(cap));
Slot* slots = new Slot[cap + 2 * this->kSlotPadding];
new (&this->dslots_) Atom<Slot*>(slots);
this->dstate_.store(0);
this->dcapacity_.store(cap);
dmult_ = mult;
size_t maxClosed = 0;
for (size_t expanded = cap; expanded < this->capacity_; expanded *= mult) {
++maxClosed;
}
closed_ = (maxClosed > 0) ? new ClosedArray[maxClosed] : nullptr;
}
bool tryObtainReadyPushTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride) noexcept {
uint64_t state;
do {
ticket = this->pushTicket_.load(std::memory_order_acquire); // A
if (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause();
continue;
}
// If there was an expansion with offset greater than this ticket,
// adjust accordingly
uint64_t offset;
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
this->turn(ticket - offset, cap))) {
// A slot is ready.
if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
// Adjust ticket
ticket -= offset;
return true;
} else {
continue;
}
} else {
if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B
// Try again. Ticket changed.
continue;
}
// Likely to block.
// Try to expand unless the ticket is for a closed array
if (offset == getOffset(state)) {
if (tryExpand(state, cap)) {
// This or another thread started an expansion. Get up-to-date info.
continue;
}
}
return false;
}
} while (true);
}
bool tryObtainPromisedPushTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride) noexcept {
uint64_t state;
do {
ticket = this->pushTicket_.load(std::memory_order_acquire);
auto numPops = this->popTicket_.load(std::memory_order_acquire);
if (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause();
continue;
}
const auto curCap = cap;
// If there was an expansion with offset greater than this ticket,
// adjust accordingly
uint64_t offset;
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
int64_t n = ticket - numPops;
if (n >= static_cast<ssize_t>(cap)) {
if ((cap == curCap) && tryExpand(state, cap)) {
// This or another thread started an expansion. Start over.
continue;
}
// Can't expand.
ticket -= offset;
return false;
}
if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
// Adjust ticket
ticket -= offset;
return true;
}
} while (true);
}
bool tryObtainReadyPopTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride) noexcept {
uint64_t state;
do {
ticket = this->popTicket_.load(std::memory_order_relaxed);
if (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause();
continue;
}
// If there was an expansion after the corresponding push ticket
// was issued, adjust accordingly
uint64_t offset;
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
if (slots[this->idx((ticket - offset), cap, stride)].mayDequeue(
this->turn(ticket - offset, cap))) {
if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
// Adjust ticket
ticket -= offset;
return true;
}
} else {
return false;
}
} while (true);
}
bool tryObtainPromisedPopTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride) noexcept {
uint64_t state;
do {
ticket = this->popTicket_.load(std::memory_order_acquire);
auto numPushes = this->pushTicket_.load(std::memory_order_acquire);
if (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause();
continue;
}
uint64_t offset;
// If there was an expansion after the corresponding push
// ticket was issued, adjust accordingly
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
if (ticket >= numPushes) {
ticket -= offset;
return false;
}
if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
ticket -= offset;
return true;
}
} while (true);
}
/// Enqueues an element with a specific ticket number
template <typename... Args>
void enqueueWithTicket(const uint64_t ticket, Args&&... args) noexcept {
Slot* slots;
size_t cap;
int stride;
uint64_t state;
uint64_t offset;
while (!trySeqlockReadSection(state, slots, cap, stride)) {
}
// If there was an expansion after this ticket was issued, adjust
// accordingly
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
this->enqueueWithTicketBase(
ticket - offset, slots, cap, stride, std::forward<Args>(args)...);
}
uint64_t getOffset(const uint64_t state) const noexcept {
return state >> kSeqlockBits;
}
int getNumClosed(const uint64_t state) const noexcept {
return (state & ((1 << kSeqlockBits) - 1)) >> 1;
}
/// Try to expand the queue. Returns true if this expansion was
/// successful or a concurrent expansion is in progress. Returns
/// false if the queue has reached its maximum capacity or
/// allocation has failed.
bool tryExpand(const uint64_t state, const size_t cap) noexcept {
if (cap == this->capacity_) {
return false;
}
// Acquire seqlock
uint64_t oldval = state;
assert((state & 1) == 0);
if (this->dstate_.compare_exchange_strong(oldval, state + 1)) {
assert(cap == this->dcapacity_.load());
uint64_t ticket =
1 + std::max(this->pushTicket_.load(), this->popTicket_.load());
size_t newCapacity = std::min(dmult_ * cap, this->capacity_);
Slot* newSlots =
new (std::nothrow) Slot[newCapacity + 2 * this->kSlotPadding];
if (newSlots == nullptr) {
// Expansion failed. Restore the seqlock
this->dstate_.store(state);
return false;
}
// Successful expansion
// calculate the current ticket offset
uint64_t offset = getOffset(state);
// calculate index in closed array
int index = getNumClosed(state);
assert((index << 1) < (1 << kSeqlockBits));
// fill the info for the closed slots array
closed_[index].offset_ = offset;
closed_[index].slots_ = this->dslots_.load();
closed_[index].capacity_ = cap;
closed_[index].stride_ = this->dstride_.load();
// update the new slots array info
this->dslots_.store(newSlots);
this->dcapacity_.store(newCapacity);
this->dstride_.store(this->computeStride(newCapacity));
// Release the seqlock and record the new ticket offset
this->dstate_.store((ticket << kSeqlockBits) + (2 * (index + 1)));
return true;
} else { // failed to acquire seqlock
// Someone acquired the seqlock. Go back to the caller and get
// up-to-date info.
return true;
}
}
/// Seqlock read-only section
bool trySeqlockReadSection(
uint64_t& state, Slot*& slots, size_t& cap, int& stride) noexcept {
state = this->dstate_.load(std::memory_order_acquire);
if (state & 1) {
// Locked.
return false;
}
// Start read-only section.
slots = this->dslots_.load(std::memory_order_relaxed);
cap = this->dcapacity_.load(std::memory_order_relaxed);
stride = this->dstride_.load(std::memory_order_relaxed);
// End of read-only section. Validate seqlock.
std::atomic_thread_fence(std::memory_order_acquire);
return (state == this->dstate_.load(std::memory_order_relaxed));
}
/// If there was an expansion after ticket was issued, update local variables
/// of the lagging operation using the most recent closed array with
/// offset <= ticket and return true. Otherwise, return false;
bool maybeUpdateFromClosed(
const uint64_t state,
const uint64_t ticket,
uint64_t& offset,
Slot*& slots,
size_t& cap,
int& stride) noexcept {
offset = getOffset(state);
if (ticket >= offset) {
return false;
}
for (int i = getNumClosed(state) - 1; i >= 0; --i) {
offset = closed_[i].offset_;
if (offset <= ticket) {
slots = closed_[i].slots_;
cap = closed_[i].capacity_;
stride = closed_[i].stride_;
return true;
}
}
// A closed array with offset <= ticket should have been found
assert(false);
return false;
}
};
namespace detail {
/// CRTP specialization of MPMCQueueBase
template <
template <typename T, template <typename> class Atom, bool Dynamic>
class Derived,
typename T,
template <typename>
class Atom,
bool Dynamic>
class MPMCQueueBase<Derived<T, Atom, Dynamic>> {
// Note: Using CRTP static casts in several functions of this base
// template instead of making called functions virtual or duplicating
// the code of calling functions in the derived partially specialized
// template
static_assert(
std::is_nothrow_constructible<T, T&&>::value ||
folly::IsRelocatable<T>::value,
"T must be relocatable or have a noexcept move constructor");
public:
typedef T value_type;
using Slot = detail::SingleElementQueue<T, Atom>;
explicit MPMCQueueBase(size_t queueCapacity)
: capacity_(queueCapacity),
dstate_(0),
dcapacity_(0),
pushTicket_(0),
popTicket_(0),
pushSpinCutoff_(0),
popSpinCutoff_(0) {
if (queueCapacity == 0) {
throw std::invalid_argument(
"MPMCQueue with explicit capacity 0 is impossible"
// Stride computation in derived classes would sigfpe if capacity is 0
);
}
// ideally this would be a static assert, but g++ doesn't allow it
assert(
alignof(MPMCQueue<T, Atom>) >= hardware_destructive_interference_size);
assert(
static_cast<uint8_t*>(static_cast<void*>(&popTicket_)) -
static_cast<uint8_t*>(static_cast<void*>(&pushTicket_)) >=
static_cast<ptrdiff_t>(hardware_destructive_interference_size));
}
/// A default-constructed queue is useful because a usable (non-zero
/// capacity) queue can be moved onto it or swapped with it
MPMCQueueBase() noexcept
: capacity_(0),
slots_(nullptr),
stride_(0),
dstate_(0),
dcapacity_(0),
pushTicket_(0),
popTicket_(0),
pushSpinCutoff_(0),
popSpinCutoff_(0) {}
/// IMPORTANT: The move constructor is here to make it easier to perform
/// the initialization phase, it is not safe to use when there are any
/// concurrent accesses (this is not checked).
MPMCQueueBase(MPMCQueueBase<Derived<T, Atom, Dynamic>>&& rhs) noexcept
: capacity_(rhs.capacity_),
slots_(rhs.slots_),
stride_(rhs.stride_),
dstate_(rhs.dstate_.load(std::memory_order_relaxed)),
dcapacity_(rhs.dcapacity_.load(std::memory_order_relaxed)),
pushTicket_(rhs.pushTicket_.load(std::memory_order_relaxed)),
popTicket_(rhs.popTicket_.load(std::memory_order_relaxed)),
pushSpinCutoff_(rhs.pushSpinCutoff_.load(std::memory_order_relaxed)),
popSpinCutoff_(rhs.popSpinCutoff_.load(std::memory_order_relaxed)) {
// relaxed ops are okay for the previous reads, since rhs queue can't
// be in concurrent use
// zero out rhs
rhs.capacity_ = 0;
rhs.slots_ = nullptr;
rhs.stride_ = 0;
rhs.dstate_.store(0, std::memory_order_relaxed);
rhs.dcapacity_.store(0, std::memory_order_relaxed);
rhs.pushTicket_.store(0, std::memory_order_relaxed);
rhs.popTicket_.store(0, std::memory_order_relaxed);
rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
}
/// IMPORTANT: The move operator is here to make it easier to perform
/// the initialization phase, it is not safe to use when there are any
/// concurrent accesses (this is not checked).
MPMCQueueBase<Derived<T, Atom, Dynamic>> const& operator=(
MPMCQueueBase<Derived<T, Atom, Dynamic>>&& rhs) {
if (this != &rhs) {
this->~MPMCQueueBase();
new (this) MPMCQueueBase(std::move(rhs));
}
return *this;
}
MPMCQueueBase(const MPMCQueueBase&) = delete;
MPMCQueueBase& operator=(const MPMCQueueBase&) = delete;
/// MPMCQueue can only be safely destroyed when there are no
/// pending enqueuers or dequeuers (this is not checked).
~MPMCQueueBase() { delete[] slots_; }
/// Returns the number of writes (including threads that are blocked waiting
/// to write) minus the number of reads (including threads that are blocked
/// waiting to read). So effectively, it becomes:
/// elements in queue + pending(calls to write) - pending(calls to read).
/// If nothing is pending, then the method returns the actual number of
/// elements in the queue.
/// The returned value can be negative if there are no writers and the queue
/// is empty, but there is one reader that is blocked waiting to read (in
/// which case, the returned size will be -1).
ssize_t size() const noexcept {
// since both pushes and pops increase monotonically, we can get a
// consistent snapshot either by bracketing a read of popTicket_ with
// two reads of pushTicket_ that return the same value, or the other
// way around. We maximize our chances by alternately attempting
// both bracketings.
uint64_t pushes = pushTicket_.load(std::memory_order_acquire); // A
uint64_t pops = popTicket_.load(std::memory_order_acquire); // B
while (true) {
uint64_t nextPushes = pushTicket_.load(std::memory_order_acquire); // C
if (pushes == nextPushes) {
// pushTicket_ didn't change from A (or the previous C) to C,
// so we can linearize at B (or D)
return ssize_t(pushes - pops);
}
pushes = nextPushes;
uint64_t nextPops = popTicket_.load(std::memory_order_acquire); // D
if (pops == nextPops) {
// popTicket_ didn't chance from B (or the previous D), so we
// can linearize at C
return ssize_t(pushes - pops);
}
pops = nextPops;
}
}
/// Returns true if there are no items available for dequeue
bool isEmpty() const noexcept { return size() <= 0; }
/// Returns true if there is currently no empty space to enqueue
bool isFull() const noexcept {
// careful with signed -> unsigned promotion, since size can be negative
return size() >= static_cast<ssize_t>(capacity_);
}
/// Returns is a guess at size() for contexts that don't need a precise
/// value, such as stats. More specifically, it returns the number of writes
/// minus the number of reads, but after reading the number of writes, more
/// writers could have came before the number of reads was sampled,
/// and this method doesn't protect against such case.
/// The returned value can be negative.
ssize_t sizeGuess() const noexcept { return writeCount() - readCount(); }
/// Doesn't change
size_t capacity() const noexcept { return capacity_; }
/// Doesn't change for non-dynamic
size_t allocatedCapacity() const noexcept { return capacity_; }
/// Returns the total number of calls to blockingWrite or successful
/// calls to write, including those blockingWrite calls that are
/// currently blocking
uint64_t writeCount() const noexcept {
return pushTicket_.load(std::memory_order_acquire);
}
/// Returns the total number of calls to blockingRead or successful
/// calls to read, including those blockingRead calls that are currently
/// blocking
uint64_t readCount() const noexcept {
return popTicket_.load(std::memory_order_acquire);
}
/// Enqueues a T constructed from args, blocking until space is
/// available. Note that this method signature allows enqueue via
/// move, if args is a T rvalue, via copy, if args is a T lvalue, or
/// via emplacement if args is an initializer list that can be passed
/// to a T constructor.
template <typename... Args>
void blockingWrite(Args&&... args) noexcept {
enqueueWithTicketBase(
pushTicket_++, slots_, capacity_, stride_, std::forward<Args>(args)...);
}
/// If an item can be enqueued with no blocking, does so and returns
/// true, otherwise returns false. This method is similar to
/// writeIfNotFull, but if you don't have a specific need for that
/// method you should use this one.
///
/// One of the common usages of this method is to enqueue via the
/// move constructor, something like q.write(std::move(x)). If write
/// returns false because the queue is full then x has not actually been
/// consumed, which looks strange. To understand why it is actually okay
/// to use x afterward, remember that std::move is just a typecast that
/// provides an rvalue reference that enables use of a move constructor
/// or operator. std::move doesn't actually move anything. It could
/// more accurately be called std::rvalue_cast or std::move_permission.
template <typename... Args>
bool write(Args&&... args) noexcept {
uint64_t ticket;
Slot* slots;
size_t cap;
int stride;
if (static_cast<Derived<T, Atom, Dynamic>*>(this)->tryObtainReadyPushTicket(
ticket, slots, cap, stride)) {
// we have pre-validated that the ticket won't block
enqueueWithTicketBase(
ticket, slots, cap, stride, std::forward<Args>(args)...);
return true;
} else {
return false;
}
}
template <class Clock, typename... Args>
bool tryWriteUntil(
const std::chrono::time_point<Clock>& when, Args&&... args) noexcept {
uint64_t ticket;
Slot* slots;
size_t cap;
int stride;
if (tryObtainPromisedPushTicketUntil(ticket, slots, cap, stride, when)) {
// we have pre-validated that the ticket won't block, or rather that
// it won't block longer than it takes another thread to dequeue an
// element from the slot it identifies.
enqueueWithTicketBase(
ticket, slots, cap, stride, std::forward<Args>(args)...);
return true;
} else {
return false;
}
}
/// If the queue is not full, enqueues and returns true, otherwise
/// returns false. Unlike write this method can be blocked by another
/// thread, specifically a read that has linearized (been assigned
/// a ticket) but not yet completed. If you don't really need this
/// function you should probably use write.
///
/// MPMCQueue isn't lock-free, so just because a read operation has
/// linearized (and isFull is false) doesn't mean that space has been
/// made available for another write. In this situation write will
/// return false, but writeIfNotFull will wait for the dequeue to finish.
/// This method is required if you are composing queues and managing
/// your own wakeup, because it guarantees that after every successful
/// write a readIfNotEmpty will succeed.
template <typename... Args>
bool writeIfNotFull(Args&&... args) noexcept {
uint64_t ticket;
Slot* slots;
size_t cap;
int stride;
if (static_cast<Derived<T, Atom, Dynamic>*>(this)
->tryObtainPromisedPushTicket(ticket, slots, cap, stride)) {
// some other thread is already dequeuing the slot into which we
// are going to enqueue, but we might have to wait for them to finish
enqueueWithTicketBase(
ticket, slots, cap, stride, std::forward<Args>(args)...);
return true;
} else {
return false;
}
}
/// Moves a dequeued element onto elem, blocking until an element
/// is available
void blockingRead(T& elem) noexcept {
uint64_t ticket;
static_cast<Derived<T, Atom, Dynamic>*>(this)->blockingReadWithTicket(
ticket, elem);
}
/// Same as blockingRead() but also records the ticket nunmer
void blockingReadWithTicket(uint64_t& ticket, T& elem) noexcept {
assert(capacity_ != 0);
ticket = popTicket_++;
dequeueWithTicketBase(ticket, slots_, capacity_, stride_, elem);
}
/// If an item can be dequeued with no blocking, does so and returns true,
/// otherwise returns false.
///
/// Note that if the matching write is still in progress, this may return
/// false even if writes that have been started later have already
/// completed. If an external mechanism is used for counting completed writes
/// (for example a semaphore) to determine when an element is ready to
/// dequeue, readIfNotEmpty() should be used instead, which will wait for the
/// write in progress.
bool read(T& elem) noexcept {
uint64_t ticket;
return readAndGetTicket(ticket, elem);
}
/// Same as read() but also records the ticket nunmer
bool readAndGetTicket(uint64_t& ticket, T& elem) noexcept {
Slot* slots;
size_t cap;
int stride;
if (static_cast<Derived<T, Atom, Dynamic>*>(this)->tryObtainReadyPopTicket(
ticket, slots, cap, stride)) {
// the ticket has been pre-validated to not block
dequeueWithTicketBase(ticket, slots, cap, stride, elem);
return true;
} else {
return false;
}
}
template <class Clock, typename... Args>
bool tryReadUntil(
const std::chrono::time_point<Clock>& when, T& elem) noexcept {
uint64_t ticket;
Slot* slots;
size_t cap;
int stride;
if (tryObtainPromisedPopTicketUntil(ticket, slots, cap, stride, when)) {
// we have pre-validated that the ticket won't block, or rather that
// it won't block longer than it takes another thread to enqueue an
// element on the slot it identifies.
dequeueWithTicketBase(ticket, slots, cap, stride, elem);
return true;
} else {
return false;
}
}
/// If the queue is not empty, dequeues and returns true, otherwise
/// returns false. If the matching write is still in progress then this
/// method may block waiting for it. If you don't rely on being able
/// to dequeue (such as by counting completed write) then you should
/// prefer read.
bool readIfNotEmpty(T& elem) noexcept {
uint64_t ticket;
Slot* slots;
size_t cap;
int stride;
if (static_cast<Derived<T, Atom, Dynamic>*>(this)
->tryObtainPromisedPopTicket(ticket, slots, cap, stride)) {
// the matching enqueue already has a ticket, but might not be done
dequeueWithTicketBase(ticket, slots, cap, stride, elem);
return true;
} else {
return false;
}
}
protected:
enum {
/// Once every kAdaptationFreq we will spin longer, to try to estimate
/// the proper spin backoff
kAdaptationFreq = 128,
/// To avoid false sharing in slots_ with neighboring memory
/// allocations, we pad it with this many SingleElementQueue-s at
/// each end
kSlotPadding =
(hardware_destructive_interference_size - 1) / sizeof(Slot) + 1
};
/// The maximum number of items in the queue at once
alignas(hardware_destructive_interference_size) size_t capacity_;
/// Anonymous union for use when Dynamic = false and true, respectively
union {
/// An array of capacity_ SingleElementQueue-s, each of which holds
/// either 0 or 1 item. We over-allocate by 2 * kSlotPadding and don't
/// touch the slots at either end, to avoid false sharing
Slot* slots_;
/// Current dynamic slots array of dcapacity_ SingleElementQueue-s
Atom<Slot*> dslots_;
};
/// Anonymous union for use when Dynamic = false and true, respectively
union {