VR-Vantage 2.7 API Documentation
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
tracy_concurrentqueue.h
Go to the documentation of this file.
1 // Provides a C++11 implementation of a multi-producer, multi-consumer lock-free queue.
2 // An overview, including benchmark results, is provided here:
3 // http://moodycamel.com/blog/2014/a-fast-general-purpose-lock-free-queue-for-c++
4 // The full design is also described in excruciating detail at:
5 // http://moodycamel.com/blog/2014/detailed-design-of-a-lock-free-queue
6 
7 // Simplified BSD license:
8 // Copyright (c) 2013-2016, Cameron Desrochers.
9 // All rights reserved.
10 //
11 // Redistribution and use in source and binary forms, with or without modification,
12 // are permitted provided that the following conditions are met:
13 //
14 // - Redistributions of source code must retain the above copyright notice, this list of
15 // conditions and the following disclaimer.
16 // - Redistributions in binary form must reproduce the above copyright notice, this list of
17 // conditions and the following disclaimer in the documentation and/or other materials
18 // provided with the distribution.
19 //
20 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
21 // EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
22 // MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
23 // THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
24 // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
25 // OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
26 // HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
27 // TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
28 // EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29 
30 
31 #pragma once
32 
33 #include "../common/TracyAlloc.hpp"
34 #include "../common/TracyForceInline.hpp"
35 #include "../common/TracySystem.hpp"
36 
37 #if defined(__GNUC__)
38 // Disable -Wconversion warnings (spuriously triggered when Traits::size_t and
39 // Traits::index_t are set to < 32 bits, causing integer promotion, causing warnings
40 // upon assigning any computed values)
41 #pragma GCC diagnostic push
42 #pragma GCC diagnostic ignored "-Wconversion"
43 #endif
44 
45 #if defined(__APPLE__)
46 #include "TargetConditionals.h"
47 #endif
48 
49 #include <atomic> // Requires C++11. Sorry VS2010.
50 #include <cassert>
51 #include <cstddef> // for max_align_t
52 #include <cstdint>
53 #include <cstdlib>
54 #include <type_traits>
55 #include <algorithm>
56 #include <utility>
57 #include <limits>
58 #include <climits> // for CHAR_BIT
59 #include <array>
60 #include <thread> // partly for __WINPTHREADS_VERSION if on MinGW-w64 w/ POSIX threading
61 
62 namespace tracy
63 {
64 
65 // Exceptions
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
69 #endif
70 #endif
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)
76 #else
77 #define MOODYCAMEL_TRY if (true)
78 #define MOODYCAMEL_CATCH(...) else if (false)
79 #define MOODYCAMEL_RETHROW
80 #define MOODYCAMEL_THROW(expr)
81 #endif
82 
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
89 // VS2012's std::is_nothrow_[move_]constructible is broken and returns true when it shouldn't :-(
90 // We have to assume *all* non-trivial constructors may throw on VS2012!
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))
98 #else
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)
102 #endif
103 #endif
104 
105 // VS2012 doesn't support deleted functions.
106 // In this case, we declare the function normally but don't define it. A link error will be generated if the function is called.
107 #ifndef MOODYCAMEL_DELETE_FUNCTION
108 #if defined(_MSC_VER) && _MSC_VER < 1800
109 #define MOODYCAMEL_DELETE_FUNCTION
110 #else
111 #define MOODYCAMEL_DELETE_FUNCTION = delete
112 #endif
113 #endif
114 
115 // Compiler-specific likely/unlikely hints
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); }
120 #else
121  inline bool cqLikely(bool x) { return x; }
122  inline bool cqUnlikely(bool x) { return x; }
123 #endif
124 } }
125 
126 namespace
127 {
128  // to avoid MSVC warning 4127: conditional expression is constant
129  template <bool>
130  struct compile_time_condition
131  {
132  static const bool value = false;
133  };
134  template <>
135  struct compile_time_condition<true>
136  {
137  static const bool value = true;
138  };
139 }
140 
141 namespace moodycamel {
142 namespace details {
143  template<typename T>
145  static_assert(std::is_integral<T>::value, "const_numeric_max can only be used with integers");
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);
149  };
150 
151 #if defined(__GLIBCXX__)
152  typedef ::max_align_t std_max_align_t; // libstdc++ forgot to add it to std:: for a while
153 #else
154  typedef std::max_align_t std_max_align_t; // Others (e.g. MSVC) insist it can *only* be accessed via std::
155 #endif
156 
157  // Some platforms have incorrectly set max_align_t to a type with <8 bytes alignment even while supporting
158  // 8-byte aligned scalar values (*cough* 32-bit iOS). Work around this with our own union. See issue #64.
159  typedef union {
161  long long y;
162  void* z;
163  } max_align_t;
164 }
165 
166 // Default traits for the ConcurrentQueue. To change some of the
167 // traits without re-implementing all of them, inherit from this
168 // struct and shadow the declarations you wish to be different;
169 // since the traits are used as a template type parameter, the
170 // shadowed declarations will be used where defined, and the defaults
171 // otherwise.
173 {
174  // General-purpose size type. std::size_t is strongly recommended.
175  typedef std::size_t size_t;
176 
177  // The type used for the enqueue and dequeue indices. Must be at least as
178  // large as size_t. Should be significantly larger than the number of elements
179  // you expect to hold at once, especially if you have a high turnover rate;
180  // for example, on 32-bit x86, if you expect to have over a hundred million
181  // elements or pump several million elements through your queue in a very
182  // short space of time, using a 32-bit type *may* trigger a race condition.
183  // A 64-bit int type is recommended in that case, and in practice will
184  // prevent a race condition no matter the usage of the queue. Note that
185  // whether the queue is lock-free with a 64-int type depends on the whether
186  // std::atomic<std::uint64_t> is lock-free, which is platform-specific.
187  typedef std::size_t index_t;
188 
189  // Internally, all elements are enqueued and dequeued from multi-element
190  // blocks; this is the smallest controllable unit. If you expect few elements
191  // but many producers, a smaller block size should be favoured. For few producers
192  // and/or many elements, a larger block size is preferred. A sane default
193  // is provided. Must be a power of 2.
194  static const size_t BLOCK_SIZE = 64*1024;
195 
196  // For explicit producers (i.e. when using a producer token), the block is
197  // checked for being empty by iterating through a list of flags, one per element.
198  // For large block sizes, this is too inefficient, and switching to an atomic
199  // counter-based approach is faster. The switch is made for block sizes strictly
200  // larger than this threshold.
201  static const size_t EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD = 32;
202 
203  // How many full blocks can be expected for a single explicit producer? This should
204  // reflect that number's maximum for optimal performance. Must be a power of 2.
205  static const size_t EXPLICIT_INITIAL_INDEX_SIZE = 32;
206 
207  // Controls the number of items that an explicit consumer (i.e. one with a token)
208  // must consume before it causes all consumers to rotate and move on to the next
209  // internal queue.
211 
212  // The maximum number of elements (inclusive) that can be enqueued to a sub-queue.
213  // Enqueue operations that would cause this limit to be surpassed will fail. Note
214  // that this limit is enforced at the block level (for performance reasons), i.e.
215  // it's rounded up to the nearest block size.
217 
218 
219  // Memory allocation can be customized if needed.
220  // malloc should return nullptr on failure, and handle alignment like std::malloc.
221 #if defined(malloc) || defined(free)
222  // Gah, this is 2015, stop defining macros that break standard code already!
223  // Work around malloc/free being special macros:
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); }
228 #else
229  static inline void* malloc(size_t size) { return tracy::tracy_malloc(size); }
230  static inline void free(void* ptr) { return tracy::tracy_free(ptr); }
231 #endif
232 };
233 
234 
235 // When producing or consuming many elements, the most efficient way is to:
236 // 1) Use one of the bulk-operation methods of the queue with a token
237 // 2) Failing that, use the bulk-operation methods without a token
238 // 3) Failing that, create a token and use that with the single-item methods
239 // 4) Failing that, use the single-parameter methods of the queue
240 // Having said that, don't create tokens willy-nilly -- ideally there should be
241 // a maximum of one token per thread (of each kind).
242 struct ProducerToken;
243 struct ConsumerToken;
244 
245 template<typename T, typename Traits> class ConcurrentQueue;
246 
247 
248 namespace details
249 {
251  {
253  std::atomic<bool> inactive;
256 
258  : next(nullptr), inactive(false), token(nullptr), threadId(0)
259  {
260  }
261  };
262 
263  template<typename T>
264  static inline bool circular_less_than(T a, T b)
265  {
266 #ifdef _MSC_VER
267 #pragma warning(push)
268 #pragma warning(disable: 4554)
269 #endif
270  static_assert(std::is_integral<T>::value && !std::numeric_limits<T>::is_signed, "circular_less_than is intended to be used only with unsigned integer types");
271  return static_cast<T>(a - b) > static_cast<T>(static_cast<T>(1) << static_cast<T>(sizeof(T) * CHAR_BIT - 1));
272 #ifdef _MSC_VER
273 #pragma warning(pop)
274 #endif
275  }
276 
277  template<typename U>
278  static inline char* align_for(char* ptr)
279  {
280  const std::size_t alignment = std::alignment_of<U>::value;
281  return ptr + (alignment - (reinterpret_cast<std::uintptr_t>(ptr) % alignment)) % alignment;
282  }
283 
284  template<typename T>
285  static inline T ceil_to_pow_2(T x)
286  {
287  static_assert(std::is_integral<T>::value && !std::numeric_limits<T>::is_signed, "ceil_to_pow_2 is intended to be used only with unsigned integer types");
288 
289  // Adapted from http://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2
290  --x;
291  x |= x >> 1;
292  x |= x >> 2;
293  x |= x >> 4;
294  for (std::size_t i = 1; i < sizeof(T); i <<= 1) {
295  x |= x >> (i << 3);
296  }
297  ++x;
298  return x;
299  }
300 
301  template<typename T>
302  static inline void swap_relaxed(std::atomic<T>& left, std::atomic<T>& right)
303  {
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);
307  }
308 
309  template<typename T>
310  static inline T const& nomove(T const& x)
311  {
312  return x;
313  }
314 
315  template<bool Enable>
316  struct nomove_if
317  {
318  template<typename T>
319  static inline T const& eval(T const& x)
320  {
321  return x;
322  }
323  };
324 
325  template<>
326  struct nomove_if<false>
327  {
328  template<typename U>
329  static inline auto eval(U&& x)
330  -> decltype(std::forward<U>(x))
331  {
332  return std::forward<U>(x);
333  }
334  };
335 
336  template<typename It>
337  static inline auto deref_noexcept(It& it) MOODYCAMEL_NOEXCEPT -> decltype(*it)
338  {
339  return *it;
340  }
341 
342 #if defined(__clang__) || !defined(__GNUC__) || __GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 8)
343  template<typename T> struct is_trivially_destructible : std::is_trivially_destructible<T> { };
344 #else
345  template<typename T> struct is_trivially_destructible : std::has_trivial_destructor<T> { };
346 #endif
347 
348  template<typename T> struct static_is_lock_free_num { enum { value = 0 }; };
349  template<> struct static_is_lock_free_num<signed char> { enum { value = ATOMIC_CHAR_LOCK_FREE }; };
350  template<> struct static_is_lock_free_num<short> { enum { value = ATOMIC_SHORT_LOCK_FREE }; };
351  template<> struct static_is_lock_free_num<int> { enum { value = ATOMIC_INT_LOCK_FREE }; };
352  template<> struct static_is_lock_free_num<long> { enum { value = ATOMIC_LONG_LOCK_FREE }; };
353  template<> struct static_is_lock_free_num<long long> { enum { value = ATOMIC_LLONG_LOCK_FREE }; };
354  template<typename T> struct static_is_lock_free : static_is_lock_free_num<typename std::make_signed<T>::type> { };
355  template<> struct static_is_lock_free<bool> { enum { value = ATOMIC_BOOL_LOCK_FREE }; };
356  template<typename U> struct static_is_lock_free<U*> { enum { value = ATOMIC_POINTER_LOCK_FREE }; };
357 }
358 
359 
361 {
362  template<typename T, typename Traits>
363  explicit ProducerToken(ConcurrentQueue<T, Traits>& queue);
364 
366  : producer(other.producer)
367  {
368  other.producer = nullptr;
369  if (producer != nullptr) {
370  producer->token = this;
371  }
372  }
373 
375  {
376  swap(other);
377  return *this;
378  }
379 
381  {
382  std::swap(producer, other.producer);
383  if (producer != nullptr) {
384  producer->token = this;
385  }
386  if (other.producer != nullptr) {
387  other.producer->token = &other;
388  }
389  }
390 
391  // A token is always valid unless:
392  // 1) Memory allocation failed during construction
393  // 2) It was moved via the move constructor
394  // (Note: assignment does a swap, leaving both potentially valid)
395  // 3) The associated queue was destroyed
396  // Note that if valid() returns true, that only indicates
397  // that the token is valid for use with a specific queue,
398  // but not which one; that's up to the user to track.
399  inline bool valid() const { return producer != nullptr; }
400 
402  {
403  if (producer != nullptr) {
404  producer->token = nullptr;
405  producer->inactive.store(true, std::memory_order_release);
406  }
407  }
408 
409  // Disable copying and assignment
411  ProducerToken& operator=(ProducerToken const&) MOODYCAMEL_DELETE_FUNCTION;
412 
413 private:
414  template<typename T, typename Traits> friend class ConcurrentQueue;
415 
416 protected:
417  details::ConcurrentQueueProducerTypelessBase* producer;
418 };
419 
420 
422 {
423  template<typename T, typename Traits>
424  explicit ConsumerToken(ConcurrentQueue<T, Traits>& q);
425 
426  ConsumerToken(ConsumerToken&& other) MOODYCAMEL_NOEXCEPT
427  : initialOffset(other.initialOffset), lastKnownGlobalOffset(other.lastKnownGlobalOffset), itemsConsumedFromCurrent(other.itemsConsumedFromCurrent), currentProducer(other.currentProducer), desiredProducer(other.desiredProducer)
428  {
429  }
430 
431  inline ConsumerToken& operator=(ConsumerToken&& other) MOODYCAMEL_NOEXCEPT
432  {
433  swap(other);
434  return *this;
435  }
436 
437  void swap(ConsumerToken& other) MOODYCAMEL_NOEXCEPT
438  {
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);
444  }
445 
446  // Disable copying and assignment
447  ConsumerToken(ConsumerToken const&) MOODYCAMEL_DELETE_FUNCTION;
448  ConsumerToken& operator=(ConsumerToken const&) MOODYCAMEL_DELETE_FUNCTION;
449 
450 private:
451  template<typename T, typename Traits> friend class ConcurrentQueue;
452 
453 private: // but shared with ConcurrentQueue
454  std::uint32_t initialOffset;
455  std::uint32_t lastKnownGlobalOffset;
456  std::uint32_t itemsConsumedFromCurrent;
457  details::ConcurrentQueueProducerTypelessBase* currentProducer;
458  details::ConcurrentQueueProducerTypelessBase* desiredProducer;
459 };
460 
461 
462 template<typename T, typename Traits = ConcurrentQueueDefaultTraits>
463 class ConcurrentQueue
464 {
465 public:
467 
470 
471  typedef typename Traits::index_t index_t;
472  typedef typename Traits::size_t size_t;
473 
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);
478 #ifdef _MSC_VER
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
482 #endif
483  static const size_t MAX_SUBQUEUE_SIZE = (details::const_numeric_max<size_t>::value - static_cast<size_t>(Traits::MAX_SUBQUEUE_SIZE) < BLOCK_SIZE) ? details::const_numeric_max<size_t>::value : ((static_cast<size_t>(Traits::MAX_SUBQUEUE_SIZE) + (BLOCK_SIZE - 1)) / BLOCK_SIZE * BLOCK_SIZE);
484 #ifdef _MSC_VER
485 #pragma warning(pop)
486 #endif
487 
488  static_assert(!std::numeric_limits<size_t>::is_signed && std::is_integral<size_t>::value, "Traits::size_t must be an unsigned integral type");
489  static_assert(!std::numeric_limits<index_t>::is_signed && std::is_integral<index_t>::value, "Traits::index_t must be an unsigned integral type");
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)");
494 
495 public:
496  // Creates a queue with at least `capacity` element slots; note that the
497  // actual number of elements that can be inserted without additional memory
498  // allocation depends on the number of producers and the block size (e.g. if
499  // the block size is equal to `capacity`, only a single block will be allocated
500  // up-front, which means only a single producer will be able to enqueue elements
501  // without an extra allocation -- blocks aren't shared between producers).
502  // This method is not thread safe -- it is up to the user to ensure that the
503  // queue is fully constructed before it starts being used by other threads (this
504  // includes making the memory effects of construction visible, possibly with a
505  // memory barrier).
506  explicit ConcurrentQueue(size_t capacity = 6 * BLOCK_SIZE)
507  : producerListTail(nullptr),
508  producerCount(0),
509  initialBlockPoolIndex(0),
510  nextExplicitConsumerId(0),
511  globalExplicitConsumerOffset(0)
512  {
513  populate_initial_block_list(capacity / BLOCK_SIZE + ((capacity & (BLOCK_SIZE - 1)) == 0 ? 0 : 1));
514  }
515 
516  // Computes the correct amount of pre-allocated blocks for you based
517  // on the minimum number of elements you want available at any given
518  // time, and the maximum concurrent number of each type of producer.
519  ConcurrentQueue(size_t minCapacity, size_t maxExplicitProducers)
520  : producerListTail(nullptr),
521  producerCount(0),
522  initialBlockPoolIndex(0),
523  nextExplicitConsumerId(0),
524  globalExplicitConsumerOffset(0)
525  {
526  size_t blocks = (((minCapacity + BLOCK_SIZE - 1) / BLOCK_SIZE) - 1) * (maxExplicitProducers + 1) + 2 * (maxExplicitProducers);
527  populate_initial_block_list(blocks);
528  }
529 
530  // Note: The queue should not be accessed concurrently while it's
531  // being deleted. It's up to the user to synchronize this.
532  // This method is not thread safe.
533  ~ConcurrentQueue()
534  {
535  // Destroy producers
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;
541  }
542  destroy(ptr);
543  ptr = next;
544  }
545 
546  // Destroy global free list
547  auto block = freeList.head_unsafe();
548  while (block != nullptr) {
549  auto next = block->freeListNext.load(std::memory_order_relaxed);
550  if (block->dynamicallyAllocated) {
551  destroy(block);
552  }
553  block = next;
554  }
555 
556  // Destroy initial free list
557  destroy_array(initialBlockPool, initialBlockPoolSize);
558  }
559 
560  // Disable copying and copy assignment
561  ConcurrentQueue(ConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
562  ConcurrentQueue(ConcurrentQueue&& other) MOODYCAMEL_DELETE_FUNCTION;
563  ConcurrentQueue& operator=(ConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
564  ConcurrentQueue& operator=(ConcurrentQueue&& other) MOODYCAMEL_DELETE_FUNCTION;
565 
566 public:
567  tracy_force_inline T* enqueue_begin(producer_token_t const& token, index_t& currentTailIndex)
568  {
569  return static_cast<ExplicitProducer*>(token.producer)->ConcurrentQueue::ExplicitProducer::enqueue_begin(currentTailIndex);
570  }
571 
572  template<class NotifyThread, class ProcessData>
573  size_t try_dequeue_bulk_single(consumer_token_t& token, NotifyThread notifyThread, ProcessData processData )
574  {
575  if (token.desiredProducer == nullptr || token.lastKnownGlobalOffset != globalExplicitConsumerOffset.load(std::memory_order_relaxed)) {
576  if (!update_current_producer_after_rotation(token)) {
577  return 0;
578  }
579  }
580 
581  size_t count = static_cast<ProducerBase*>(token.currentProducer)->dequeue_bulk(notifyThread, processData);
582  token.itemsConsumedFromCurrent += static_cast<std::uint32_t>(count);
583 
584  auto tail = producerListTail.load(std::memory_order_acquire);
585  auto ptr = static_cast<ProducerBase*>(token.currentProducer)->next_prod();
586  if (ptr == nullptr) {
587  ptr = tail;
588  }
589  if( count == 0 )
590  {
591  while (ptr != static_cast<ProducerBase*>(token.currentProducer)) {
592  auto dequeued = ptr->dequeue_bulk(notifyThread, processData);
593  if (dequeued != 0) {
594  token.currentProducer = ptr;
595  token.itemsConsumedFromCurrent = static_cast<std::uint32_t>(dequeued);
596  return dequeued;
597  }
598  ptr = ptr->next_prod();
599  if (ptr == nullptr) {
600  ptr = tail;
601  }
602  }
603  return 0;
604  }
605  else
606  {
607  token.currentProducer = ptr;
608  token.itemsConsumedFromCurrent = 0;
609  return count;
610  }
611  }
612 
613 
614  // Returns an estimate of the total number of elements currently in the queue. This
615  // estimate is only accurate if the queue has completely stabilized before it is called
616  // (i.e. all enqueue and dequeue operations have completed and their memory effects are
617  // visible on the calling thread, and no further operations start while this method is
618  // being called).
619  // Thread-safe.
620  size_t size_approx() const
621  {
622  size_t size = 0;
623  for (auto ptr = producerListTail.load(std::memory_order_acquire); ptr != nullptr; ptr = ptr->next_prod()) {
624  size += ptr->size_approx();
625  }
626  return size;
627  }
628 
629 
630  // Returns true if the underlying atomic variables used by
631  // the queue are lock-free (they should be on most platforms).
632  // Thread-safe.
633  static bool is_lock_free()
634  {
635  return
641  }
642 
643 
644 private:
645  friend struct ProducerToken;
646  friend struct ConsumerToken;
647  friend struct ExplicitProducer;
648 
649 
651  // Queue methods
653 
654  inline bool update_current_producer_after_rotation(consumer_token_t& token)
655  {
656  // Ah, there's been a rotation, figure out where we should be!
657  auto tail = producerListTail.load(std::memory_order_acquire);
658  if (token.desiredProducer == nullptr && tail == nullptr) {
659  return false;
660  }
661  auto prodCount = producerCount.load(std::memory_order_relaxed);
662  auto globalOffset = globalExplicitConsumerOffset.load(std::memory_order_relaxed);
663  if (details::cqUnlikely(token.desiredProducer == nullptr)) {
664  // Aha, first time we're dequeueing anything.
665  // Figure out our local position
666  // Note: offset is from start, not end, but we're traversing from end -- subtract from count first
667  std::uint32_t offset = prodCount - 1 - (token.initialOffset % prodCount);
668  token.desiredProducer = tail;
669  for (std::uint32_t i = 0; i != offset; ++i) {
670  token.desiredProducer = static_cast<ProducerBase*>(token.desiredProducer)->next_prod();
671  if (token.desiredProducer == nullptr) {
672  token.desiredProducer = tail;
673  }
674  }
675  }
676 
677  std::uint32_t delta = globalOffset - token.lastKnownGlobalOffset;
678  if (delta >= prodCount) {
679  delta = delta % prodCount;
680  }
681  for (std::uint32_t i = 0; i != delta; ++i) {
682  token.desiredProducer = static_cast<ProducerBase*>(token.desiredProducer)->next_prod();
683  if (token.desiredProducer == nullptr) {
684  token.desiredProducer = tail;
685  }
686  }
687 
688  token.lastKnownGlobalOffset = globalOffset;
689  token.currentProducer = token.desiredProducer;
690  token.itemsConsumedFromCurrent = 0;
691  return true;
692  }
693 
694 
696  // Free list
698 
699  template <typename N>
701  {
702  FreeListNode() : freeListRefs(0), freeListNext(nullptr) { }
703 
704  std::atomic<std::uint32_t> freeListRefs;
705  std::atomic<N*> freeListNext;
706  };
707 
708  // A simple CAS-based lock-free free list. Not the fastest thing in the world under heavy contention, but
709  // simple and correct (assuming nodes are never freed until after the free list is destroyed), and fairly
710  // speedy under low contention.
711  template<typename N> // N must inherit FreeListNode or have the same fields (and initialization of them)
712  struct FreeList
713  {
714  FreeList() : freeListHead(nullptr) { }
715  FreeList(FreeList&& other) : freeListHead(other.freeListHead.load(std::memory_order_relaxed)) { other.freeListHead.store(nullptr, std::memory_order_relaxed); }
716  void swap(FreeList& other) { details::swap_relaxed(freeListHead, other.freeListHead); }
717 
718  FreeList(FreeList const&) MOODYCAMEL_DELETE_FUNCTION;
719  FreeList& operator=(FreeList const&) MOODYCAMEL_DELETE_FUNCTION;
720 
721  inline void add(N* node)
722  {
723  // We know that the should-be-on-freelist bit is 0 at this point, so it's safe to
724  // set it using a fetch_add
725  if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST, std::memory_order_acq_rel) == 0) {
726  // Oh look! We were the last ones referencing this node, and we know
727  // we want to add it to the free list, so let's do it!
728  add_knowing_refcount_is_zero(node);
729  }
730  }
731 
732  inline N* try_get()
733  {
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);
740  continue;
741  }
742 
743  // Good, reference count has been incremented (it wasn't at zero), which means we can read the
744  // next and not worry about it changing between now and the time we do the CAS
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)) {
747  // Yay, got the node. This means it was on the list, which means shouldBeOnFreeList must be false no
748  // matter the refcount (because nobody else knows it's been taken off yet, it can't have been put back on).
749  assert((head->freeListRefs.load(std::memory_order_relaxed) & SHOULD_BE_ON_FREELIST) == 0);
750 
751  // Decrease refcount twice, once for our ref, and once for the list's ref
752  head->freeListRefs.fetch_sub(2, std::memory_order_release);
753  return head;
754  }
755 
756  // OK, the head must have changed on us, but we still need to decrease the refcount we increased.
757  // Note that we don't need to release any memory effects, but we do need to ensure that the reference
758  // count decrement happens-after the CAS on the head.
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);
762  }
763  }
764 
765  return nullptr;
766  }
767 
768  // Useful for traversing the list when there's no contention (e.g. to destroy remaining nodes)
769  N* head_unsafe() const { return freeListHead.load(std::memory_order_relaxed); }
770 
771  private:
772  inline void add_knowing_refcount_is_zero(N* node)
773  {
774  // Since the refcount is zero, and nobody can increase it once it's zero (except us, and we run
775  // only one copy of this method per node at a time, i.e. the single thread case), then we know
776  // we can safely change the next pointer of the node; however, once the refcount is back above
777  // zero, then other threads could increase it (happens under heavy contention, when the refcount
778  // goes to zero in between a load and a refcount increment of a node in try_get, then back up to
779  // something non-zero, then the refcount increment is done by the other thread) -- so, if the CAS
780  // to add the node to the actual list fails, decrease the refcount and leave the add operation to
781  // the next thread who puts the refcount back at zero (which could be us, hence the loop).
782  auto head = freeListHead.load(std::memory_order_relaxed);
783  while (true) {
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)) {
787  // Hmm, the add failed, but we can only try again when the refcount goes back to zero
788  if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST - 1, std::memory_order_release) == 1) {
789  continue;
790  }
791  }
792  return;
793  }
794  }
795 
796  private:
797  // Implemented like a stack, but where node order doesn't matter (nodes are inserted out of order under contention)
798  std::atomic<N*> freeListHead;
799 
800  static const std::uint32_t REFS_MASK = 0x7FFFFFFF;
801  static const std::uint32_t SHOULD_BE_ON_FREELIST = 0x80000000;
802  };
803 
804 
806  // Block
808 
809  struct Block
810  {
812  : next(nullptr), elementsCompletelyDequeued(0), freeListRefs(0), freeListNext(nullptr), shouldBeOnFreeList(false), dynamicallyAllocated(true)
813  {
814  }
815 
816  inline bool is_empty() const
817  {
819  // Check flags
820  for (size_t i = 0; i < BLOCK_SIZE; ++i) {
821  if (!emptyFlags[i].load(std::memory_order_relaxed)) {
822  return false;
823  }
824  }
825 
826  // Aha, empty; make sure we have all other memory effects that happened before the empty flags were set
827  std::atomic_thread_fence(std::memory_order_acquire);
828  return true;
829  }
830  else {
831  // Check counter
832  if (elementsCompletelyDequeued.load(std::memory_order_relaxed) == BLOCK_SIZE) {
833  std::atomic_thread_fence(std::memory_order_acquire);
834  return true;
835  }
836  assert(elementsCompletelyDequeued.load(std::memory_order_relaxed) <= BLOCK_SIZE);
837  return false;
838  }
839  }
840 
841  // Returns true if the block is now empty (does not apply in explicit context)
842  inline bool set_empty(index_t i)
843  {
844  if (BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD) {
845  // Set flag
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);
848  return false;
849  }
850  else {
851  // Increment counter
852  auto prevVal = elementsCompletelyDequeued.fetch_add(1, std::memory_order_release);
853  assert(prevVal < BLOCK_SIZE);
854  return prevVal == BLOCK_SIZE - 1;
855  }
856  }
857 
858  // Sets multiple contiguous item statuses to 'empty' (assumes no wrapping and count > 0).
859  // Returns true if the block is now empty (does not apply in explicit context).
860  inline bool set_many_empty(index_t i, size_t count)
861  {
863  // Set flags
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);
869  }
870  return false;
871  }
872  else {
873  // Increment counter
874  auto prevVal = elementsCompletelyDequeued.fetch_add(count, std::memory_order_release);
875  assert(prevVal + count <= BLOCK_SIZE);
876  return prevVal + count == BLOCK_SIZE;
877  }
878  }
879 
880  inline void set_all_empty()
881  {
882  if (BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD) {
883  // Set all flags
884  for (size_t i = 0; i != BLOCK_SIZE; ++i) {
885  emptyFlags[i].store(true, std::memory_order_relaxed);
886  }
887  }
888  else {
889  // Reset counter
890  elementsCompletelyDequeued.store(BLOCK_SIZE, std::memory_order_relaxed);
891  }
892  }
893 
894  inline void reset_empty()
895  {
897  // Reset flags
898  for (size_t i = 0; i != BLOCK_SIZE; ++i) {
899  emptyFlags[i].store(false, std::memory_order_relaxed);
900  }
901  }
902  else {
903  // Reset counter
904  elementsCompletelyDequeued.store(0, std::memory_order_relaxed);
905  }
906  }
907 
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)); }
910 
911  private:
912  // IMPORTANT: This must be the first member in Block, so that if T depends on the alignment of
913  // addresses returned by malloc, that alignment will be preserved. Apparently clang actually
914  // generates code that uses this assumption for AVX instructions in some cases. Ideally, we
915  // should also align Block to the alignment of T in case it's higher than malloc's 16-byte
916  // alignment, but this is hard to do in a cross-platform way. Assert for this case:
917  static_assert(std::alignment_of<T>::value <= std::alignment_of<details::max_align_t>::value, "The queue does not support super-aligned types at this time");
918  // Additionally, we need the alignment of Block itself to be a multiple of max_align_t since
919  // otherwise the appropriate padding will not be added at the end of Block in order to make
920  // arrays of Blocks all be properly aligned (not just the first one). We use a union to force
921  // this.
922  union {
923  char elements[sizeof(T) * BLOCK_SIZE];
925  };
926  public:
928  std::atomic<size_t> elementsCompletelyDequeued;
929  std::atomic<bool> emptyFlags[BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD ? BLOCK_SIZE : 1];
930  public:
931  std::atomic<std::uint32_t> freeListRefs;
932  std::atomic<Block*> freeListNext;
933  std::atomic<bool> shouldBeOnFreeList;
934  bool dynamicallyAllocated; // Perhaps a better name for this would be 'isNotPartOfInitialBlockPool'
935  };
936  static_assert(std::alignment_of<Block>::value >= std::alignment_of<details::max_align_t>::value, "Internal error: Blocks must be at least as aligned as the type they are wrapping");
937 
938 
940  // Producer base
942 
944  {
945  ProducerBase(ConcurrentQueue* parent_) :
946  tailIndex(0),
947  headIndex(0),
948  dequeueOptimisticCount(0),
949  dequeueOvercommit(0),
950  tailBlock(nullptr),
951  parent(parent_)
952  {
953  }
954 
955  virtual ~ProducerBase() { };
956 
957  template<class NotifyThread, class ProcessData>
958  inline size_t dequeue_bulk(NotifyThread notifyThread, ProcessData processData)
959  {
960  return static_cast<ExplicitProducer*>(this)->dequeue_bulk(notifyThread, processData);
961  }
962 
963  inline ProducerBase* next_prod() const { return static_cast<ProducerBase*>(next); }
964 
965  inline size_t size_approx() const
966  {
967  auto tail = tailIndex.load(std::memory_order_relaxed);
968  auto head = headIndex.load(std::memory_order_relaxed);
969  return details::circular_less_than(head, tail) ? static_cast<size_t>(tail - head) : 0;
970  }
971 
972  inline index_t getTail() const { return tailIndex.load(std::memory_order_relaxed); }
973  protected:
974  std::atomic<index_t> tailIndex; // Where to enqueue to next
975  std::atomic<index_t> headIndex; // Where to dequeue from next
976 
977  std::atomic<index_t> dequeueOptimisticCount;
978  std::atomic<index_t> dequeueOvercommit;
979 
981 
982  public:
983  ConcurrentQueue* parent;
984  };
985 
986 
987  public:
989  // Explicit queue
992  {
993  explicit ExplicitProducer(ConcurrentQueue* _parent) :
994  ProducerBase(_parent),
995  blockIndex(nullptr),
996  pr_blockIndexSlotsUsed(0),
997  pr_blockIndexSize(EXPLICIT_INITIAL_INDEX_SIZE >> 1),
998  pr_blockIndexFront(0),
999  pr_blockIndexEntries(nullptr),
1000  pr_blockIndexRaw(nullptr)
1001  {
1002  size_t poolBasedIndexSize = details::ceil_to_pow_2(_parent->initialBlockPoolSize) >> 1;
1003  if (poolBasedIndexSize > pr_blockIndexSize) {
1004  pr_blockIndexSize = poolBasedIndexSize;
1005  }
1006 
1007  new_block_index(0); // This creates an index with double the number of current entries, i.e. EXPLICIT_INITIAL_INDEX_SIZE
1008  }
1009 
1011  {
1012  // Destruct any elements not yet dequeued.
1013  // Since we're in the destructor, we can assume all elements
1014  // are either completely dequeued or completely not (no halfways).
1015  if (this->tailBlock != nullptr) { // Note this means there must be a block index too
1016  // First find the block that's partially dequeued, if any
1017  Block* halfDequeuedBlock = nullptr;
1018  if ((this->headIndex.load(std::memory_order_relaxed) & static_cast<index_t>(BLOCK_SIZE - 1)) != 0) {
1019  // The head's not on a block boundary, meaning a block somewhere is partially dequeued
1020  // (or the head block is the tail block and was fully dequeued, but the head/tail are still not on a boundary)
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);
1024  }
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;
1027  }
1028 
1029  // Start at the head block (note the first line in the loop gives us the head from the tail on the first iteration)
1030  auto block = this->tailBlock;
1031  do {
1032  block = block->next;
1033  if (block->ConcurrentQueue::Block::is_empty()) {
1034  continue;
1035  }
1036 
1037  size_t i = 0; // Offset into block
1038  if (block == halfDequeuedBlock) {
1039  i = static_cast<size_t>(this->headIndex.load(std::memory_order_relaxed) & static_cast<index_t>(BLOCK_SIZE - 1));
1040  }
1041 
1042  // Walk through all the items in the block; if this is the tail block, we need to stop when we reach the tail index
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();
1046  }
1047  } while (block != this->tailBlock);
1048  }
1049 
1050  // Destroy all blocks that we own
1051  if (this->tailBlock != nullptr) {
1052  auto block = this->tailBlock;
1053  do {
1054  auto nextBlock = block->next;
1055  if (block->dynamicallyAllocated) {
1056  destroy(block);
1057  }
1058  else {
1059  this->parent->add_block_to_free_list(block);
1060  }
1061  block = nextBlock;
1062  } while (block != this->tailBlock);
1063  }
1064 
1065  // Destroy the block indices
1066  auto header = static_cast<BlockIndexHeader*>(pr_blockIndexRaw);
1067  while (header != nullptr) {
1068  auto prev = static_cast<BlockIndexHeader*>(header->prev);
1069  header->~BlockIndexHeader();
1070  (Traits::free)(header);
1071  header = prev;
1072  }
1073  }
1074 
1075  inline void enqueue_begin_alloc(index_t currentTailIndex)
1076  {
1077  // We reached the end of a block, start a new one
1078  if (this->tailBlock != nullptr && this->tailBlock->next->ConcurrentQueue::Block::is_empty()) {
1079  // We can re-use the block ahead of us, it's empty!
1080  this->tailBlock = this->tailBlock->next;
1081  this->tailBlock->ConcurrentQueue::Block::reset_empty();
1082 
1083  // We'll put the block on the block index (guaranteed to be room since we're conceptually removing the
1084  // last block from it first -- except instead of removing then adding, we can just overwrite).
1085  // Note that there must be a valid block index here, since even if allocation failed in the ctor,
1086  // it would have been re-attempted when adding the first block to the queue; since there is such
1087  // a block, a block index must have been successfully allocated.
1088  }
1089  else {
1090  // We're going to need a new block; check that the block index has room
1091  if (pr_blockIndexRaw == nullptr || pr_blockIndexSlotsUsed == pr_blockIndexSize) {
1092  // Hmm, the circular block index is already full -- we'll need
1093  // to allocate a new index. Note pr_blockIndexRaw can only be nullptr if
1094  // the initial allocation failed in the constructor.
1095  new_block_index(pr_blockIndexSlotsUsed);
1096  }
1097 
1098  // Insert a new block in the circular linked list
1099  auto newBlock = this->parent->ConcurrentQueue::requisition_block();
1100  newBlock->ConcurrentQueue::Block::reset_empty();
1101  if (this->tailBlock == nullptr) {
1102  newBlock->next = newBlock;
1103  }
1104  else {
1105  newBlock->next = this->tailBlock->next;
1106  this->tailBlock->next = newBlock;
1107  }
1108  this->tailBlock = newBlock;
1109  ++pr_blockIndexSlotsUsed;
1110  }
1111 
1112  // Add block to block index
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);
1118  }
1119 
1120  tracy_force_inline T* enqueue_begin(index_t& currentTailIndex)
1121  {
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);
1125  }
1126  return (*this->tailBlock)[currentTailIndex];
1127  }
1128 
1129  tracy_force_inline std::atomic<index_t>& get_tail_index()
1130  {
1131  return this->tailIndex;
1132  }
1133 
1134  template<class NotifyThread, class ProcessData>
1135  size_t dequeue_bulk(NotifyThread notifyThread, ProcessData processData)
1136  {
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);
1143 
1144  auto myDequeueCount = this->dequeueOptimisticCount.fetch_add(desiredCount, std::memory_order_relaxed);
1145  assert(overcommit <= myDequeueCount);
1146 
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);
1153  }
1154 
1155  // Get the first index. Note that since there's guaranteed to be at least actualCount elements, this
1156  // will never exceed tail.
1157  auto firstIndex = this->headIndex.fetch_add(actualCount, std::memory_order_acq_rel);
1158 
1159  // Determine which block the first element is in
1160  auto localBlockIndex = blockIndex.load(std::memory_order_acquire);
1161  auto localBlockIndexHead = localBlockIndex->front.load(std::memory_order_acquire);
1162 
1163  auto headBase = localBlockIndex->entries[localBlockIndexHead].base;
1164  auto firstBlockBaseIndex = firstIndex & ~static_cast<index_t>(BLOCK_SIZE - 1);
1165  auto offset = static_cast<size_t>(static_cast<typename std::make_signed<index_t>::type>(firstBlockBaseIndex - headBase) / BLOCK_SIZE);
1166  auto indexIndex = (localBlockIndexHead + offset) & (localBlockIndex->size - 1);
1167 
1168  notifyThread( this->threadId );
1169 
1170  // Iterate the blocks and dequeue
1171  auto index = firstIndex;
1172  do {
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;
1177 
1178  const auto sz = endIndex - index;
1179  processData( (*block)[index], sz );
1180  index += sz;
1181 
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);
1185 
1186  return actualCount;
1187  }
1188  else {
1189  // Wasn't anything to dequeue after all; make the effective dequeue count eventually consistent
1190  this->dequeueOvercommit.fetch_add(desiredCount, std::memory_order_release);
1191  }
1192  }
1193 
1194  return 0;
1195  }
1196 
1197  private:
1199  {
1200  index_t base;
1202  };
1203 
1205  {
1206  size_t size;
1207  std::atomic<size_t> front; // Current slot (not next, like pr_blockIndexFront)
1209  void* prev;
1210  };
1211 
1212 
1213  bool new_block_index(size_t numberOfFilledSlotsToExpose)
1214  {
1215  auto prevBlockSizeMask = pr_blockIndexSize - 1;
1216 
1217  // Create the new block
1218  pr_blockIndexSize <<= 1;
1219  auto newRawPtr = static_cast<char*>((Traits::malloc)(sizeof(BlockIndexHeader) + std::alignment_of<BlockIndexEntry>::value - 1 + sizeof(BlockIndexEntry) * pr_blockIndexSize));
1220  if (newRawPtr == nullptr) {
1221  pr_blockIndexSize >>= 1; // Reset to allow graceful retry
1222  return false;
1223  }
1224 
1225  auto newBlockIndexEntries = reinterpret_cast<BlockIndexEntry*>(details::align_for<BlockIndexEntry>(newRawPtr + sizeof(BlockIndexHeader)));
1226 
1227  // Copy in all the old indices, if any
1228  size_t j = 0;
1229  if (pr_blockIndexSlotsUsed != 0) {
1230  auto i = (pr_blockIndexFront - pr_blockIndexSlotsUsed) & prevBlockSizeMask;
1231  do {
1232  newBlockIndexEntries[j++] = pr_blockIndexEntries[i];
1233  i = (i + 1) & prevBlockSizeMask;
1234  } while (i != pr_blockIndexFront);
1235  }
1236 
1237  // Update everything
1238  auto header = new (newRawPtr) BlockIndexHeader;
1239  header->size = pr_blockIndexSize;
1240  header->front.store(numberOfFilledSlotsToExpose - 1, std::memory_order_relaxed);
1241  header->entries = newBlockIndexEntries;
1242  header->prev = pr_blockIndexRaw; // we link the new block to the old one so we can free it later
1243 
1244  pr_blockIndexFront = j;
1245  pr_blockIndexEntries = newBlockIndexEntries;
1246  pr_blockIndexRaw = newRawPtr;
1247  blockIndex.store(header, std::memory_order_release);
1248 
1249  return true;
1250  }
1251 
1252  private:
1253  std::atomic<BlockIndexHeader*> blockIndex;
1254 
1255  // To be used by producer only -- consumer must use the ones in referenced by blockIndex
1258  size_t pr_blockIndexFront; // Next slot (not current)
1261  };
1262 
1263  ExplicitProducer* get_explicit_producer(producer_token_t const& token)
1264  {
1265  return static_cast<ExplicitProducer*>(token.producer);
1266  }
1267 
1268  private:
1269 
1271  // Block pool manipulation
1273 
1274  void populate_initial_block_list(size_t blockCount)
1275  {
1276  initialBlockPoolSize = blockCount;
1277  if (initialBlockPoolSize == 0) {
1278  initialBlockPool = nullptr;
1279  return;
1280  }
1281 
1282  initialBlockPool = create_array<Block>(blockCount);
1283  if (initialBlockPool == nullptr) {
1284  initialBlockPoolSize = 0;
1285  }
1286  for (size_t i = 0; i < initialBlockPoolSize; ++i) {
1287  initialBlockPool[i].dynamicallyAllocated = false;
1288  }
1289  }
1290 
1291  inline Block* try_get_block_from_initial_pool()
1292  {
1293  if (initialBlockPoolIndex.load(std::memory_order_relaxed) >= initialBlockPoolSize) {
1294  return nullptr;
1295  }
1296 
1297  auto index = initialBlockPoolIndex.fetch_add(1, std::memory_order_relaxed);
1298 
1299  return index < initialBlockPoolSize ? (initialBlockPool + index) : nullptr;
1300  }
1301 
1302  inline void add_block_to_free_list(Block* block)
1303  {
1304  freeList.add(block);
1305  }
1306 
1307  inline void add_blocks_to_free_list(Block* block)
1308  {
1309  while (block != nullptr) {
1310  auto next = block->next;
1311  add_block_to_free_list(block);
1312  block = next;
1313  }
1314  }
1315 
1316  inline Block* try_get_block_from_free_list()
1317  {
1318  return freeList.try_get();
1319  }
1320 
1321  // Gets a free block from one of the memory pools, or allocates a new one (if applicable)
1322  Block* requisition_block()
1323  {
1324  auto block = try_get_block_from_initial_pool();
1325  if (block != nullptr) {
1326  return block;
1327  }
1328 
1329  block = try_get_block_from_free_list();
1330  if (block != nullptr) {
1331  return block;
1332  }
1333 
1334  return create<Block>();
1335  }
1336 
1337 
1339  // Producer list manipulation
1341 
1342  ProducerBase* recycle_or_create_producer()
1343  {
1344  bool recycled;
1345  return recycle_or_create_producer(recycled);
1346  }
1347 
1348  ProducerBase* recycle_or_create_producer(bool& recycled)
1349  {
1350  // Try to re-use one first
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 )
1354  {
1355  bool expected = true;
1356  if (ptr->inactive.compare_exchange_strong(expected, /* desired */ false, std::memory_order_acquire, std::memory_order_relaxed)) {
1357  // We caught one! It's been marked as activated, the caller can have it
1358  recycled = true;
1359  return ptr;
1360  }
1361  }
1362  }
1363  }
1364 
1365  recycled = false;
1366  return add_producer(static_cast<ProducerBase*>(create<ExplicitProducer>(this)));
1367  }
1368 
1369  ProducerBase* add_producer(ProducerBase* producer)
1370  {
1371  // Handle failed memory allocation
1372  if (producer == nullptr) {
1373  return nullptr;
1374  }
1375 
1376  producerCount.fetch_add(1, std::memory_order_relaxed);
1377 
1378  // Add it to the lock-free list
1379  auto prevTail = producerListTail.load(std::memory_order_relaxed);
1380  do {
1381  producer->next = prevTail;
1382  } while (!producerListTail.compare_exchange_weak(prevTail, producer, std::memory_order_release, std::memory_order_relaxed));
1383 
1384  return producer;
1385  }
1386 
1387  void reown_producers()
1388  {
1389  // After another instance is moved-into/swapped-with this one, all the
1390  // producers we stole still think their parents are the other queue.
1391  // So fix them up!
1392  for (auto ptr = producerListTail.load(std::memory_order_relaxed); ptr != nullptr; ptr = ptr->next_prod()) {
1393  ptr->parent = this;
1394  }
1395  }
1396 
1398  // Utility functions
1400 
1401  template<typename U>
1402  static inline U* create_array(size_t count)
1403  {
1404  assert(count > 0);
1405  return static_cast<U*>((Traits::malloc)(sizeof(U) * count));
1406  }
1407 
1408  template<typename U>
1409  static inline void destroy_array(U* p, size_t count)
1410  {
1411  ((void)count);
1412  if (p != nullptr) {
1413  assert(count > 0);
1414  (Traits::free)(p);
1415  }
1416  }
1417 
1418  template<typename U>
1419  static inline U* create()
1420  {
1421  auto p = (Traits::malloc)(sizeof(U));
1422  return new (p) U;
1423  }
1424 
1425  template<typename U, typename A1>
1426  static inline U* create(A1&& a1)
1427  {
1428  auto p = (Traits::malloc)(sizeof(U));
1429  return new (p) U(std::forward<A1>(a1));
1430  }
1431 
1432  template<typename U>
1433  static inline void destroy(U* p)
1434  {
1435  if (p != nullptr) {
1436  p->~U();
1437  }
1438  (Traits::free)(p);
1439  }
1440 
1441 private:
1442  std::atomic<ProducerBase*> producerListTail;
1443  std::atomic<std::uint32_t> producerCount;
1444 
1445  std::atomic<size_t> initialBlockPoolIndex;
1448 
1450 
1451  std::atomic<std::uint32_t> nextExplicitConsumerId;
1452  std::atomic<std::uint32_t> globalExplicitConsumerOffset;
1453 };
1454 
1455 
1456 template<typename T, typename Traits>
1458  : producer(queue.recycle_or_create_producer())
1459 {
1460  if (producer != nullptr) {
1461  producer->token = this;
1462  producer->threadId = detail::GetThreadHandleImpl();
1463  }
1464 }
1465 
1466 template<typename T, typename Traits>
1468  : itemsConsumedFromCurrent(0), currentProducer(nullptr), desiredProducer(nullptr)
1469 {
1470  initialOffset = queue.nextExplicitConsumerId.fetch_add(1, std::memory_order_release);
1471  lastKnownGlobalOffset = static_cast<std::uint32_t>(-1);
1472 }
1473 
1474 template<typename T, typename Traits>
1476 {
1477  a.swap(b);
1478 }
1479 
1481 {
1482  a.swap(b);
1483 }
1484 
1485 inline void swap(ConsumerToken& a, ConsumerToken& b) MOODYCAMEL_NOEXCEPT
1486 {
1487  a.swap(b);
1488 }
1489 
1490 }
1491 
1492 } /* namespace tracy */
1493 
1494 #if defined(__GNUC__)
1495 #pragma GCC diagnostic pop
1496 #endif


Copyright © 2005-2021 MAK Technologies. All Rights Reserved (www.mak.com)