33 #include "../common/TracyAlloc.hpp"
34 #include "../common/TracyForceInline.hpp"
35 #include "../common/TracySystem.hpp"
41 #pragma GCC diagnostic push
42 #pragma GCC diagnostic ignored "-Wconversion"
45 #if defined(__APPLE__)
46 #include "TargetConditionals.h"
54 #include <type_traits>
66 #ifndef MOODYCAMEL_EXCEPTIONS_ENABLED
67 #if (defined(_MSC_VER) && defined(_CPPUNWIND)) || (defined(__GNUC__) && defined(__EXCEPTIONS)) || (!defined(_MSC_VER) && !defined(__GNUC__))
68 #define MOODYCAMEL_EXCEPTIONS_ENABLED
71 #ifdef MOODYCAMEL_EXCEPTIONS_ENABLED
72 #define MOODYCAMEL_TRY try
73 #define MOODYCAMEL_CATCH(...) catch(__VA_ARGS__)
74 #define MOODYCAMEL_RETHROW throw
75 #define MOODYCAMEL_THROW(expr) throw (expr)
77 #define MOODYCAMEL_TRY if (true)
78 #define MOODYCAMEL_CATCH(...) else if (false)
79 #define MOODYCAMEL_RETHROW
80 #define MOODYCAMEL_THROW(expr)
83 #ifndef MOODYCAMEL_NOEXCEPT
84 #if !defined(MOODYCAMEL_EXCEPTIONS_ENABLED)
85 #define MOODYCAMEL_NOEXCEPT
86 #define MOODYCAMEL_NOEXCEPT_CTOR(type, valueType, expr) true
87 #define MOODYCAMEL_NOEXCEPT_ASSIGN(type, valueType, expr) true
88 #elif defined(_MSC_VER) && defined(_NOEXCEPT) && _MSC_VER < 1800
91 #define MOODYCAMEL_NOEXCEPT _NOEXCEPT
92 #define MOODYCAMEL_NOEXCEPT_CTOR(type, valueType, expr) (std::is_rvalue_reference<valueType>::value && std::is_move_constructible<type>::value ? std::is_trivially_move_constructible<type>::value : std::is_trivially_copy_constructible<type>::value)
93 #define MOODYCAMEL_NOEXCEPT_ASSIGN(type, valueType, expr) ((std::is_rvalue_reference<valueType>::value && std::is_move_assignable<type>::value ? std::is_trivially_move_assignable<type>::value || std::is_nothrow_move_assignable<type>::value : std::is_trivially_copy_assignable<type>::value || std::is_nothrow_copy_assignable<type>::value) && MOODYCAMEL_NOEXCEPT_CTOR(type, valueType, expr))
94 #elif defined(_MSC_VER) && defined(_NOEXCEPT) && _MSC_VER < 1900
95 #define MOODYCAMEL_NOEXCEPT _NOEXCEPT
96 #define MOODYCAMEL_NOEXCEPT_CTOR(type, valueType, expr) (std::is_rvalue_reference<valueType>::value && std::is_move_constructible<type>::value ? std::is_trivially_move_constructible<type>::value || std::is_nothrow_move_constructible<type>::value : std::is_trivially_copy_constructible<type>::value || std::is_nothrow_copy_constructible<type>::value)
97 #define MOODYCAMEL_NOEXCEPT_ASSIGN(type, valueType, expr) ((std::is_rvalue_reference<valueType>::value && std::is_move_assignable<type>::value ? std::is_trivially_move_assignable<type>::value || std::is_nothrow_move_assignable<type>::value : std::is_trivially_copy_assignable<type>::value || std::is_nothrow_copy_assignable<type>::value) && MOODYCAMEL_NOEXCEPT_CTOR(type, valueType, expr))
99 #define MOODYCAMEL_NOEXCEPT noexcept
100 #define MOODYCAMEL_NOEXCEPT_CTOR(type, valueType, expr) noexcept(expr)
101 #define MOODYCAMEL_NOEXCEPT_ASSIGN(type, valueType, expr) noexcept(expr)
107 #ifndef MOODYCAMEL_DELETE_FUNCTION
108 #if defined(_MSC_VER) && _MSC_VER < 1800
109 #define MOODYCAMEL_DELETE_FUNCTION
111 #define MOODYCAMEL_DELETE_FUNCTION = delete
116 namespace moodycamel {
namespace details {
117 #if defined(__GNUC__)
118 inline bool cqLikely(
bool x) {
return __builtin_expect((x),
true); }
119 inline bool cqUnlikely(
bool x) {
return __builtin_expect((x),
false); }
130 struct compile_time_condition
135 struct compile_time_condition<true>
137 static const bool value =
true;
141 namespace moodycamel {
146 static const T
value = std::numeric_limits<T>::is_signed
147 ? (
static_cast<T
>(1) << (
sizeof(T) * CHAR_BIT - 1)) -
static_cast<T
>(1)
148 : static_cast<T>(-1);
151 #if defined(__GLIBCXX__)
221 #if defined(malloc) || defined(free)
224 static inline void* WORKAROUND_malloc(
size_t size) {
return malloc(size); }
225 static inline void WORKAROUND_free(
void* ptr) {
return free(ptr); }
226 static inline void* (
malloc)(
size_t size) {
return WORKAROUND_malloc(size); }
227 static inline void (
free)(
void* ptr) {
return WORKAROUND_free(ptr); }
242 struct ProducerToken;
243 struct ConsumerToken;
245 template<
typename T,
typename Traits>
class ConcurrentQueue;
267 #pragma warning(push)
268 #pragma warning(disable: 4554)
271 return static_cast<T
>(a -
b) > static_cast<T>(static_cast<T>(1) <<
static_cast<T
>(
sizeof(T) * CHAR_BIT - 1));
281 return ptr + (alignment - (
reinterpret_cast<std::uintptr_t>(ptr) % alignment)) % alignment;
294 for (std::size_t i = 1; i <
sizeof(T); i <<= 1) {
304 T temp = std::move(left.load(std::memory_order_relaxed));
305 left.store(std::move(right.load(std::memory_order_relaxed)), std::memory_order_relaxed);
306 right.store(std::move(temp), std::memory_order_relaxed);
315 template<
bool Enable>
319 static inline T
const&
eval(T
const&
x)
330 -> decltype(std::forward<U>(
x))
332 return std::forward<U>(
x);
336 template<
typename It>
342 #if defined(__clang__) || !defined(__GNUC__) || __GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 8)
362 template<
typename T,
typename Traits>
368 other.producer =
nullptr;
369 if (producer !=
nullptr) {
370 producer->
token =
this;
386 if (other.producer !=
nullptr) {
417 details::ConcurrentQueueProducerTypelessBase*
producer;
423 template<
typename T,
typename Traits>
427 : initialOffset(other.initialOffset), lastKnownGlobalOffset(other.lastKnownGlobalOffset), itemsConsumedFromCurrent(other.itemsConsumedFromCurrent), currentProducer(other.currentProducer), desiredProducer(other.desiredProducer)
439 std::swap(initialOffset, other.initialOffset);
440 std::swap(lastKnownGlobalOffset, other.lastKnownGlobalOffset);
441 std::swap(itemsConsumedFromCurrent, other.itemsConsumedFromCurrent);
442 std::swap(currentProducer, other.currentProducer);
443 std::swap(desiredProducer, other.desiredProducer);
447 ConsumerToken(ConsumerToken
const&) MOODYCAMEL_DELETE_FUNCTION;
448 ConsumerToken& operator=(ConsumerToken const&) MOODYCAMEL_DELETE_FUNCTION;
451 template<typename T, typename Traits> friend class ConcurrentQueue;
457 details::ConcurrentQueueProducerTypelessBase* currentProducer;
458 details::ConcurrentQueueProducerTypelessBase* desiredProducer;
463 class ConcurrentQueue
474 static const size_t BLOCK_SIZE =
static_cast<size_t>(Traits::BLOCK_SIZE);
475 static const size_t EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD =
static_cast<size_t>(Traits::EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD);
476 static const size_t EXPLICIT_INITIAL_INDEX_SIZE =
static_cast<size_t>(Traits::EXPLICIT_INITIAL_INDEX_SIZE);
477 static const std::uint32_t EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE =
static_cast<std::uint32_t>(Traits::EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE);
479 #pragma warning(push)
480 #pragma warning(disable: 4307) // + integral constant overflow (that's what the ternary expression is for!)
481 #pragma warning(disable: 4309) // static_cast: Truncation of constant value
490 static_assert(
sizeof(index_t) >=
sizeof(
size_t),
"Traits::index_t must be at least as wide as Traits::size_t");
491 static_assert((BLOCK_SIZE > 1) && !(BLOCK_SIZE & (BLOCK_SIZE - 1)),
"Traits::BLOCK_SIZE must be a power of 2 (and at least 2)");
492 static_assert((EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD > 1) && !(EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD & (EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD - 1)),
"Traits::EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD must be a power of 2 (and greater than 1)");
493 static_assert((EXPLICIT_INITIAL_INDEX_SIZE > 1) && !(EXPLICIT_INITIAL_INDEX_SIZE & (EXPLICIT_INITIAL_INDEX_SIZE - 1)),
"Traits::EXPLICIT_INITIAL_INDEX_SIZE must be a power of 2 (and greater than 1)");
507 : producerListTail(nullptr),
509 initialBlockPoolIndex(0),
510 nextExplicitConsumerId(0),
511 globalExplicitConsumerOffset(0)
513 populate_initial_block_list(capacity / BLOCK_SIZE + ((capacity & (BLOCK_SIZE - 1)) == 0 ? 0 : 1));
520 : producerListTail(nullptr),
522 initialBlockPoolIndex(0),
523 nextExplicitConsumerId(0),
524 globalExplicitConsumerOffset(0)
526 size_t blocks = (((minCapacity + BLOCK_SIZE - 1) / BLOCK_SIZE) - 1) * (maxExplicitProducers + 1) + 2 * (maxExplicitProducers);
527 populate_initial_block_list(blocks);
536 auto ptr = producerListTail.load(std::memory_order_relaxed);
537 while (ptr !=
nullptr) {
538 auto next = ptr->next_prod();
539 if (ptr->token !=
nullptr) {
540 ptr->token->producer =
nullptr;
547 auto block = freeList.head_unsafe();
548 while (block !=
nullptr) {
549 auto next = block->freeListNext.load(std::memory_order_relaxed);
550 if (block->dynamicallyAllocated) {
557 destroy_array(initialBlockPool, initialBlockPoolSize);
562 ConcurrentQueue(ConcurrentQueue&& other) MOODYCAMEL_DELETE_FUNCTION;
563 ConcurrentQueue& operator=(ConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
564 ConcurrentQueue& operator=(ConcurrentQueue&& other) MOODYCAMEL_DELETE_FUNCTION;
572 template<
class NotifyThread,
class ProcessData>
573 size_t try_dequeue_bulk_single(consumer_token_t& token, NotifyThread notifyThread, ProcessData processData )
576 if (!update_current_producer_after_rotation(token)) {
584 auto tail = producerListTail.load(std::memory_order_acquire);
586 if (ptr ==
nullptr) {
592 auto dequeued = ptr->
dequeue_bulk(notifyThread, processData);
598 ptr = ptr->next_prod();
599 if (ptr ==
nullptr) {
620 size_t size_approx()
const
623 for (
auto ptr = producerListTail.load(std::memory_order_acquire); ptr !=
nullptr; ptr = ptr->next_prod()) {
624 size += ptr->size_approx();
633 static bool is_lock_free()
646 friend struct ConsumerToken;
654 inline bool update_current_producer_after_rotation(consumer_token_t& token)
657 auto tail = producerListTail.load(std::memory_order_acquire);
661 auto prodCount = producerCount.load(std::memory_order_relaxed);
662 auto globalOffset = globalExplicitConsumerOffset.load(std::memory_order_relaxed);
678 if (delta >= prodCount) {
679 delta = delta % prodCount;
699 template <
typename N>
715 FreeList(
FreeList&& other) : freeListHead(other.freeListHead.load(std::memory_order_relaxed)) { other.freeListHead.store(
nullptr, std::memory_order_relaxed); }
721 inline
void add(N* node)
725 if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST, std::memory_order_acq_rel) == 0) {
728 add_knowing_refcount_is_zero(node);
734 auto head = freeListHead.load(std::memory_order_acquire);
735 while (head !=
nullptr) {
736 auto prevHead = head;
737 auto refs = head->freeListRefs.load(std::memory_order_relaxed);
738 if ((refs & REFS_MASK) == 0 || !head->freeListRefs.compare_exchange_strong(refs, refs + 1, std::memory_order_acquire, std::memory_order_relaxed)) {
739 head = freeListHead.load(std::memory_order_acquire);
745 auto next = head->freeListNext.load(std::memory_order_relaxed);
746 if (freeListHead.compare_exchange_strong(head, next, std::memory_order_acquire, std::memory_order_relaxed)) {
749 assert((head->freeListRefs.load(std::memory_order_relaxed) & SHOULD_BE_ON_FREELIST) == 0);
752 head->freeListRefs.fetch_sub(2, std::memory_order_release);
759 refs = prevHead->freeListRefs.fetch_sub(1, std::memory_order_acq_rel);
760 if (refs == SHOULD_BE_ON_FREELIST + 1) {
761 add_knowing_refcount_is_zero(prevHead);
769 N*
head_unsafe()
const {
return freeListHead.load(std::memory_order_relaxed); }
772 inline void add_knowing_refcount_is_zero(N* node)
782 auto head = freeListHead.load(std::memory_order_relaxed);
784 node->
freeListNext.store(head, std::memory_order_relaxed);
785 node->freeListRefs.store(1, std::memory_order_release);
786 if (!freeListHead.compare_exchange_strong(head, node, std::memory_order_release, std::memory_order_relaxed)) {
788 if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST - 1, std::memory_order_release) == 1) {
812 : next(nullptr), elementsCompletelyDequeued(0), freeListRefs(0), freeListNext(nullptr), shouldBeOnFreeList(false), dynamicallyAllocated(true)
816 inline bool is_empty()
const
820 for (
size_t i = 0; i < BLOCK_SIZE; ++i) {
821 if (!emptyFlags[i].load(std::memory_order_relaxed)) {
827 std::atomic_thread_fence(std::memory_order_acquire);
832 if (elementsCompletelyDequeued.load(std::memory_order_relaxed) == BLOCK_SIZE) {
833 std::atomic_thread_fence(std::memory_order_acquire);
836 assert(elementsCompletelyDequeued.load(std::memory_order_relaxed) <= BLOCK_SIZE);
842 inline bool set_empty(index_t i)
844 if (BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD) {
846 assert(!emptyFlags[BLOCK_SIZE - 1 - static_cast<size_t>(i & static_cast<index_t>(BLOCK_SIZE - 1))].load(std::memory_order_relaxed));
847 emptyFlags[BLOCK_SIZE - 1 -
static_cast<size_t>(i &
static_cast<index_t
>(BLOCK_SIZE - 1))].store(
true, std::memory_order_release);
852 auto prevVal = elementsCompletelyDequeued.fetch_add(1, std::memory_order_release);
853 assert(prevVal < BLOCK_SIZE);
854 return prevVal == BLOCK_SIZE - 1;
860 inline bool set_many_empty(index_t i,
size_t count)
864 std::atomic_thread_fence(std::memory_order_release);
865 i = BLOCK_SIZE - 1 -
static_cast<size_t>(i &
static_cast<index_t
>(BLOCK_SIZE - 1)) - count + 1;
866 for (
size_t j = 0; j !=
count; ++j) {
867 assert(!emptyFlags[i + j].load(std::memory_order_relaxed));
868 emptyFlags[i + j].store(
true, std::memory_order_relaxed);
874 auto prevVal = elementsCompletelyDequeued.fetch_add(count, std::memory_order_release);
875 assert(prevVal + count <= BLOCK_SIZE);
876 return prevVal + count == BLOCK_SIZE;
880 inline void set_all_empty()
882 if (BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD) {
884 for (
size_t i = 0; i != BLOCK_SIZE; ++i) {
885 emptyFlags[i].store(
true, std::memory_order_relaxed);
890 elementsCompletelyDequeued.store(BLOCK_SIZE, std::memory_order_relaxed);
894 inline void reset_empty()
898 for (
size_t i = 0; i != BLOCK_SIZE; ++i) {
899 emptyFlags[i].store(
false, std::memory_order_relaxed);
904 elementsCompletelyDequeued.store(0, std::memory_order_relaxed);
908 inline T*
operator[](index_t idx)
MOODYCAMEL_NOEXCEPT {
return static_cast<T*
>(
static_cast<void*
>(elements)) +
static_cast<size_t>(idx &
static_cast<index_t
>(BLOCK_SIZE - 1)); }
909 inline T
const*
operator[](index_t idx)
const MOODYCAMEL_NOEXCEPT {
return static_cast<T const*
>(
static_cast<void const*
>(elements)) +
static_cast<size_t>(idx &
static_cast<index_t
>(BLOCK_SIZE - 1)); }
923 char elements[
sizeof(T) * BLOCK_SIZE];
929 std::atomic<bool> emptyFlags[BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD ? BLOCK_SIZE : 1];
948 dequeueOptimisticCount(0),
949 dequeueOvercommit(0),
957 template<
class NotifyThread,
class ProcessData>
958 inline size_t dequeue_bulk(NotifyThread notifyThread, ProcessData processData)
960 return static_cast<ExplicitProducer*
>(
this)->dequeue_bulk(notifyThread, processData);
965 inline size_t size_approx()
const
967 auto tail = tailIndex.load(std::memory_order_relaxed);
968 auto head = headIndex.load(std::memory_order_relaxed);
972 inline index_t
getTail()
const {
return tailIndex.load(std::memory_order_relaxed); }
996 pr_blockIndexSlotsUsed(0),
997 pr_blockIndexSize(EXPLICIT_INITIAL_INDEX_SIZE >> 1),
998 pr_blockIndexFront(0),
999 pr_blockIndexEntries(nullptr),
1000 pr_blockIndexRaw(nullptr)
1003 if (poolBasedIndexSize > pr_blockIndexSize) {
1004 pr_blockIndexSize = poolBasedIndexSize;
1015 if (this->tailBlock !=
nullptr) {
1017 Block* halfDequeuedBlock =
nullptr;
1018 if ((this->headIndex.load(std::memory_order_relaxed) &
static_cast<index_t
>(BLOCK_SIZE - 1)) != 0) {
1021 size_t i = (pr_blockIndexFront - pr_blockIndexSlotsUsed) & (pr_blockIndexSize - 1);
1022 while (details::circular_less_than<index_t>(pr_blockIndexEntries[i].base + BLOCK_SIZE, this->headIndex.load(std::memory_order_relaxed))) {
1023 i = (i + 1) & (pr_blockIndexSize - 1);
1025 assert(details::circular_less_than<index_t>(pr_blockIndexEntries[i].base, this->headIndex.load(std::memory_order_relaxed)));
1026 halfDequeuedBlock = pr_blockIndexEntries[i].block;
1030 auto block = this->tailBlock;
1032 block = block->
next;
1033 if (block->ConcurrentQueue::Block::is_empty()) {
1038 if (block == halfDequeuedBlock) {
1039 i =
static_cast<size_t>(this->headIndex.load(std::memory_order_relaxed) &
static_cast<index_t
>(BLOCK_SIZE - 1));
1043 auto lastValidIndex = (this->tailIndex.load(std::memory_order_relaxed) &
static_cast<index_t
>(BLOCK_SIZE - 1)) == 0 ? BLOCK_SIZE :
static_cast<size_t>(this->tailIndex.load(std::memory_order_relaxed) &
static_cast<index_t
>(BLOCK_SIZE - 1));
1044 while (i != BLOCK_SIZE && (block != this->tailBlock || i != lastValidIndex)) {
1045 (*block)[i++]->~T();
1047 }
while (block != this->tailBlock);
1051 if (this->tailBlock !=
nullptr) {
1052 auto block = this->tailBlock;
1054 auto nextBlock = block->
next;
1055 if (block->dynamicallyAllocated) {
1059 this->parent->add_block_to_free_list(block);
1062 }
while (block != this->tailBlock);
1067 while (header !=
nullptr) {
1069 header->~BlockIndexHeader();
1070 (Traits::free)(header);
1075 inline void enqueue_begin_alloc(index_t currentTailIndex)
1078 if (this->tailBlock !=
nullptr && this->tailBlock->next->ConcurrentQueue::Block::is_empty()) {
1080 this->tailBlock = this->tailBlock->next;
1081 this->tailBlock->ConcurrentQueue::Block::reset_empty();
1091 if (pr_blockIndexRaw ==
nullptr || pr_blockIndexSlotsUsed == pr_blockIndexSize) {
1095 new_block_index(pr_blockIndexSlotsUsed);
1099 auto newBlock = this->parent->ConcurrentQueue::requisition_block();
1100 newBlock->ConcurrentQueue::Block::reset_empty();
1101 if (this->tailBlock ==
nullptr) {
1102 newBlock->next = newBlock;
1105 newBlock->next = this->tailBlock->next;
1106 this->tailBlock->next = newBlock;
1108 this->tailBlock = newBlock;
1109 ++pr_blockIndexSlotsUsed;
1113 auto& entry = blockIndex.load(std::memory_order_relaxed)->entries[pr_blockIndexFront];
1114 entry.base = currentTailIndex;
1115 entry.block = this->tailBlock;
1116 blockIndex.load(std::memory_order_relaxed)->front.store(pr_blockIndexFront, std::memory_order_release);
1117 pr_blockIndexFront = (pr_blockIndexFront + 1) & (pr_blockIndexSize - 1);
1122 currentTailIndex = this->tailIndex.load(std::memory_order_relaxed);
1123 if (
details::cqUnlikely((currentTailIndex & static_cast<index_t>(BLOCK_SIZE - 1)) == 0)) {
1124 this->enqueue_begin_alloc(currentTailIndex);
1126 return (*this->tailBlock)[currentTailIndex];
1131 return this->tailIndex;
1134 template<
class NotifyThread,
class ProcessData>
1135 size_t dequeue_bulk(NotifyThread notifyThread, ProcessData processData)
1137 auto tail = this->tailIndex.load(std::memory_order_relaxed);
1138 auto overcommit = this->dequeueOvercommit.load(std::memory_order_relaxed);
1139 auto desiredCount =
static_cast<size_t>(tail - (this->dequeueOptimisticCount.load(std::memory_order_relaxed) - overcommit));
1140 if (details::circular_less_than<size_t>(0, desiredCount)) {
1141 desiredCount = desiredCount < 8192 ? desiredCount : 8192;
1142 std::atomic_thread_fence(std::memory_order_acquire);
1144 auto myDequeueCount = this->dequeueOptimisticCount.fetch_add(desiredCount, std::memory_order_relaxed);
1145 assert(overcommit <= myDequeueCount);
1147 tail = this->tailIndex.load(std::memory_order_acquire);
1148 auto actualCount =
static_cast<size_t>(tail - (myDequeueCount - overcommit));
1149 if (details::circular_less_than<size_t>(0, actualCount)) {
1150 actualCount = desiredCount < actualCount ? desiredCount : actualCount;
1151 if (actualCount < desiredCount) {
1152 this->dequeueOvercommit.fetch_add(desiredCount - actualCount, std::memory_order_release);
1157 auto firstIndex = this->headIndex.fetch_add(actualCount, std::memory_order_acq_rel);
1160 auto localBlockIndex = blockIndex.load(std::memory_order_acquire);
1161 auto localBlockIndexHead = localBlockIndex->front.load(std::memory_order_acquire);
1163 auto headBase = localBlockIndex->entries[localBlockIndexHead].base;
1164 auto firstBlockBaseIndex = firstIndex & ~static_cast<index_t>(BLOCK_SIZE - 1);
1166 auto indexIndex = (localBlockIndexHead +
offset) & (localBlockIndex->size - 1);
1168 notifyThread( this->threadId );
1171 auto index = firstIndex;
1173 auto firstIndexInBlock =
index;
1174 auto endIndex = (
index & ~static_cast<index_t>(BLOCK_SIZE - 1)) +
static_cast<index_t
>(BLOCK_SIZE);
1175 endIndex = details::circular_less_than<index_t>(firstIndex +
static_cast<index_t
>(actualCount), endIndex) ? firstIndex +
static_cast<index_t
>(actualCount) : endIndex;
1176 auto block = localBlockIndex->entries[indexIndex].block;
1178 const auto sz = endIndex -
index;
1179 processData( (*block)[index], sz );
1182 block->ConcurrentQueue::Block::set_many_empty(firstIndexInBlock, static_cast<size_t>(endIndex - firstIndexInBlock));
1183 indexIndex = (indexIndex + 1) & (localBlockIndex->size - 1);
1184 }
while (
index != firstIndex + actualCount);
1190 this->dequeueOvercommit.fetch_add(desiredCount, std::memory_order_release);
1213 bool new_block_index(
size_t numberOfFilledSlotsToExpose)
1215 auto prevBlockSizeMask = pr_blockIndexSize - 1;
1218 pr_blockIndexSize <<= 1;
1220 if (newRawPtr ==
nullptr) {
1221 pr_blockIndexSize >>= 1;
1229 if (pr_blockIndexSlotsUsed != 0) {
1230 auto i = (pr_blockIndexFront - pr_blockIndexSlotsUsed) & prevBlockSizeMask;
1232 newBlockIndexEntries[j++] = pr_blockIndexEntries[i];
1233 i = (i + 1) & prevBlockSizeMask;
1234 }
while (i != pr_blockIndexFront);
1239 header->
size = pr_blockIndexSize;
1240 header->
front.store(numberOfFilledSlotsToExpose - 1, std::memory_order_relaxed);
1241 header->entries = newBlockIndexEntries;
1242 header->prev = pr_blockIndexRaw;
1244 pr_blockIndexFront = j;
1245 pr_blockIndexEntries = newBlockIndexEntries;
1246 pr_blockIndexRaw = newRawPtr;
1247 blockIndex.store(header, std::memory_order_release);
1274 void populate_initial_block_list(
size_t blockCount)
1276 initialBlockPoolSize = blockCount;
1277 if (initialBlockPoolSize == 0) {
1278 initialBlockPool =
nullptr;
1282 initialBlockPool = create_array<Block>(blockCount);
1283 if (initialBlockPool ==
nullptr) {
1284 initialBlockPoolSize = 0;
1286 for (
size_t i = 0; i < initialBlockPoolSize; ++i) {
1287 initialBlockPool[i].dynamicallyAllocated =
false;
1291 inline Block* try_get_block_from_initial_pool()
1293 if (initialBlockPoolIndex.load(std::memory_order_relaxed) >= initialBlockPoolSize) {
1297 auto index = initialBlockPoolIndex.fetch_add(1, std::memory_order_relaxed);
1299 return index < initialBlockPoolSize ? (initialBlockPool +
index) :
nullptr;
1302 inline void add_block_to_free_list(
Block* block)
1304 freeList.add(block);
1307 inline void add_blocks_to_free_list(
Block* block)
1309 while (block !=
nullptr) {
1310 auto next = block->
next;
1311 add_block_to_free_list(block);
1316 inline Block* try_get_block_from_free_list()
1318 return freeList.try_get();
1324 auto block = try_get_block_from_initial_pool();
1325 if (block !=
nullptr) {
1329 block = try_get_block_from_free_list();
1330 if (block !=
nullptr) {
1334 return create<Block>();
1345 return recycle_or_create_producer(recycled);
1351 for (
auto ptr = producerListTail.load(std::memory_order_acquire); ptr !=
nullptr; ptr = ptr->next_prod()) {
1352 if (ptr->inactive.load(std::memory_order_relaxed)) {
1353 if( ptr->size_approx() == 0 )
1355 bool expected =
true;
1356 if (ptr->inactive.compare_exchange_strong(expected,
false, std::memory_order_acquire, std::memory_order_relaxed)) {
1366 return add_producer(static_cast<ProducerBase*>(create<ExplicitProducer>(
this)));
1372 if (producer ==
nullptr) {
1376 producerCount.fetch_add(1, std::memory_order_relaxed);
1379 auto prevTail = producerListTail.load(std::memory_order_relaxed);
1381 producer->
next = prevTail;
1382 }
while (!producerListTail.compare_exchange_weak(prevTail, producer, std::memory_order_release, std::memory_order_relaxed));
1387 void reown_producers()
1392 for (
auto ptr = producerListTail.load(std::memory_order_relaxed); ptr !=
nullptr; ptr = ptr->next_prod()) {
1401 template<
typename U>
1402 static inline U* create_array(
size_t count)
1405 return static_cast<U*
>((Traits::malloc)(
sizeof(U) *
count));
1408 template<
typename U>
1409 static inline void destroy_array(U*
p,
size_t count)
1418 template<
typename U>
1419 static inline U* create()
1421 auto p = (Traits::malloc)(
sizeof(U));
1425 template<
typename U,
typename A1>
1426 static inline U* create(A1&& a1)
1428 auto p = (Traits::malloc)(
sizeof(U));
1429 return new (
p) U(std::forward<A1>(a1));
1432 template<
typename U>
1433 static inline void destroy(U*
p)
1456 template<
typename T,
typename Traits>
1458 : producer(queue.recycle_or_create_producer())
1460 if (producer !=
nullptr) {
1461 producer->token =
this;
1466 template<
typename T,
typename Traits>
1468 : itemsConsumedFromCurrent(0), currentProducer(nullptr), desiredProducer(nullptr)
1474 template<
typename T,
typename Traits>
1494 #if defined(__GNUC__)
1495 #pragma GCC diagnostic pop