VR-Vantage 3.1 API Documentation
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties 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 // Compiler-specific likely/unlikely hints
66 namespace moodycamel { namespace details {
67 #if defined(__GNUC__)
68  inline bool cqLikely(bool x) { return __builtin_expect((x), true); }
69  inline bool cqUnlikely(bool x) { return __builtin_expect((x), false); }
70 #else
71  inline bool cqLikely(bool x) { return x; }
72  inline bool cqUnlikely(bool x) { return x; }
73 #endif
74 } }
75 
76 namespace
77 {
78  // to avoid MSVC warning 4127: conditional expression is constant
79  template <bool>
80  struct compile_time_condition
81  {
82  static const bool value = false;
83  };
84  template <>
85  struct compile_time_condition<true>
86  {
87  static const bool value = true;
88  };
89 }
90 
91 namespace moodycamel {
92 namespace details {
93  template<typename T>
95  static_assert(std::is_integral<T>::value, "const_numeric_max can only be used with integers");
96  static const T value = std::numeric_limits<T>::is_signed
97  ? (static_cast<T>(1) << (sizeof(T) * CHAR_BIT - 1)) - static_cast<T>(1)
98  : static_cast<T>(-1);
99  };
100 
101 #if defined(__GLIBCXX__)
102  typedef ::max_align_t std_max_align_t; // libstdc++ forgot to add it to std:: for a while
103 #else
104  typedef std::max_align_t std_max_align_t; // Others (e.g. MSVC) insist it can *only* be accessed via std::
105 #endif
106 
107  // Some platforms have incorrectly set max_align_t to a type with <8 bytes alignment even while supporting
108  // 8-byte aligned scalar values (*cough* 32-bit iOS). Work around this with our own union. See issue #64.
109  typedef union {
111  long long y;
112  void* z;
113  } max_align_t;
114 }
115 
116 // Default traits for the ConcurrentQueue. To change some of the
117 // traits without re-implementing all of them, inherit from this
118 // struct and shadow the declarations you wish to be different;
119 // since the traits are used as a template type parameter, the
120 // shadowed declarations will be used where defined, and the defaults
121 // otherwise.
123 {
124  // General-purpose size type. std::size_t is strongly recommended.
125  typedef std::size_t size_t;
126 
127  // The type used for the enqueue and dequeue indices. Must be at least as
128  // large as size_t. Should be significantly larger than the number of elements
129  // you expect to hold at once, especially if you have a high turnover rate;
130  // for example, on 32-bit x86, if you expect to have over a hundred million
131  // elements or pump several million elements through your queue in a very
132  // short space of time, using a 32-bit type *may* trigger a race condition.
133  // A 64-bit int type is recommended in that case, and in practice will
134  // prevent a race condition no matter the usage of the queue. Note that
135  // whether the queue is lock-free with a 64-int type depends on the whether
136  // std::atomic<std::uint64_t> is lock-free, which is platform-specific.
137  typedef std::size_t index_t;
138 
139  // Internally, all elements are enqueued and dequeued from multi-element
140  // blocks; this is the smallest controllable unit. If you expect few elements
141  // but many producers, a smaller block size should be favoured. For few producers
142  // and/or many elements, a larger block size is preferred. A sane default
143  // is provided. Must be a power of 2.
144  static const size_t BLOCK_SIZE = 64*1024;
145 
146  // For explicit producers (i.e. when using a producer token), the block is
147  // checked for being empty by iterating through a list of flags, one per element.
148  // For large block sizes, this is too inefficient, and switching to an atomic
149  // counter-based approach is faster. The switch is made for block sizes strictly
150  // larger than this threshold.
151  static const size_t EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD = 32;
152 
153  // How many full blocks can be expected for a single explicit producer? This should
154  // reflect that number's maximum for optimal performance. Must be a power of 2.
155  static const size_t EXPLICIT_INITIAL_INDEX_SIZE = 32;
156 
157  // Controls the number of items that an explicit consumer (i.e. one with a token)
158  // must consume before it causes all consumers to rotate and move on to the next
159  // internal queue.
161 
162  // The maximum number of elements (inclusive) that can be enqueued to a sub-queue.
163  // Enqueue operations that would cause this limit to be surpassed will fail. Note
164  // that this limit is enforced at the block level (for performance reasons), i.e.
165  // it's rounded up to the nearest block size.
167 
168 
169  // Memory allocation can be customized if needed.
170  // malloc should return nullptr on failure, and handle alignment like std::malloc.
171 #if defined(malloc) || defined(free)
172  // Gah, this is 2015, stop defining macros that break standard code already!
173  // Work around malloc/free being special macros:
174  static inline void* WORKAROUND_malloc(size_t size) { return malloc(size); }
175  static inline void WORKAROUND_free(void* ptr) { return free(ptr); }
176  static inline void* (malloc)(size_t size) { return WORKAROUND_malloc(size); }
177  static inline void (free)(void* ptr) { return WORKAROUND_free(ptr); }
178 #else
179  static inline void* malloc(size_t size) { return tracy::tracy_malloc(size); }
180  static inline void free(void* ptr) { return tracy::tracy_free(ptr); }
181 #endif
182 };
183 
184 
185 // When producing or consuming many elements, the most efficient way is to:
186 // 1) Use one of the bulk-operation methods of the queue with a token
187 // 2) Failing that, use the bulk-operation methods without a token
188 // 3) Failing that, create a token and use that with the single-item methods
189 // 4) Failing that, use the single-parameter methods of the queue
190 // Having said that, don't create tokens willy-nilly -- ideally there should be
191 // a maximum of one token per thread (of each kind).
192 struct ProducerToken;
193 struct ConsumerToken;
194 
195 template<typename T, typename Traits> class ConcurrentQueue;
196 
197 
198 namespace details
199 {
201  {
203  std::atomic<bool> inactive;
206 
208  : next(nullptr), inactive(false), token(nullptr), threadId(0)
209  {
210  }
211  };
212 
213  template<typename T>
214  static inline bool circular_less_than(T a, T b)
215  {
216 #ifdef _MSC_VER
217 #pragma warning(push)
218 #pragma warning(disable: 4554)
219 #endif
220  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");
221  return static_cast<T>(a - b) > static_cast<T>(static_cast<T>(1) << static_cast<T>(sizeof(T) * CHAR_BIT - 1));
222 #ifdef _MSC_VER
223 #pragma warning(pop)
224 #endif
225  }
226 
227  template<typename U>
228  static inline char* align_for(char* ptr)
229  {
230  const std::size_t alignment = std::alignment_of<U>::value;
231  return ptr + (alignment - (reinterpret_cast<std::uintptr_t>(ptr) % alignment)) % alignment;
232  }
233 
234  template<typename T>
235  static inline T ceil_to_pow_2(T x)
236  {
237  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");
238 
239  // Adapted from http://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2
240  --x;
241  x |= x >> 1;
242  x |= x >> 2;
243  x |= x >> 4;
244  for (std::size_t i = 1; i < sizeof(T); i <<= 1) {
245  x |= x >> (i << 3);
246  }
247  ++x;
248  return x;
249  }
250 
251  template<typename T>
252  static inline void swap_relaxed(std::atomic<T>& left, std::atomic<T>& right)
253  {
254  T temp = std::move(left.load(std::memory_order_relaxed));
255  left.store(std::move(right.load(std::memory_order_relaxed)), std::memory_order_relaxed);
256  right.store(std::move(temp), std::memory_order_relaxed);
257  }
258 
259  template<typename T>
260  static inline T const& nomove(T const& x)
261  {
262  return x;
263  }
264 
265  template<bool Enable>
266  struct nomove_if
267  {
268  template<typename T>
269  static inline T const& eval(T const& x)
270  {
271  return x;
272  }
273  };
274 
275  template<>
276  struct nomove_if<false>
277  {
278  template<typename U>
279  static inline auto eval(U&& x)
280  -> decltype(std::forward<U>(x))
281  {
282  return std::forward<U>(x);
283  }
284  };
285 
286  template<typename It>
287  static inline auto deref_noexcept(It& it) noexcept -> decltype(*it)
288  {
289  return *it;
290  }
291 
292 #if defined(__clang__) || !defined(__GNUC__) || __GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 8)
293  template<typename T> struct is_trivially_destructible : std::is_trivially_destructible<T> { };
294 #else
295  template<typename T> struct is_trivially_destructible : std::has_trivial_destructor<T> { };
296 #endif
297 
298  template<typename T> struct static_is_lock_free_num { enum { value = 0 }; };
299  template<> struct static_is_lock_free_num<signed char> { enum { value = ATOMIC_CHAR_LOCK_FREE }; };
300  template<> struct static_is_lock_free_num<short> { enum { value = ATOMIC_SHORT_LOCK_FREE }; };
301  template<> struct static_is_lock_free_num<int> { enum { value = ATOMIC_INT_LOCK_FREE }; };
302  template<> struct static_is_lock_free_num<long> { enum { value = ATOMIC_LONG_LOCK_FREE }; };
303  template<> struct static_is_lock_free_num<long long> { enum { value = ATOMIC_LLONG_LOCK_FREE }; };
304  template<typename T> struct static_is_lock_free : static_is_lock_free_num<typename std::make_signed<T>::type> { };
305  template<> struct static_is_lock_free<bool> { enum { value = ATOMIC_BOOL_LOCK_FREE }; };
306  template<typename U> struct static_is_lock_free<U*> { enum { value = ATOMIC_POINTER_LOCK_FREE }; };
307 }
308 
309 
311 {
312  template<typename T, typename Traits>
313  explicit ProducerToken(ConcurrentQueue<T, Traits>& queue);
314 
315  ProducerToken(ProducerToken&& other) noexcept
316  : producer(other.producer)
317  {
318  other.producer = nullptr;
319  if (producer != nullptr) {
320  producer->token = this;
321  }
322  }
323 
324  inline ProducerToken& operator=(ProducerToken&& other) noexcept
325  {
326  swap(other);
327  return *this;
328  }
329 
330  void swap(ProducerToken& other) noexcept
331  {
332  std::swap(producer, other.producer);
333  if (producer != nullptr) {
334  producer->token = this;
335  }
336  if (other.producer != nullptr) {
337  other.producer->token = &other;
338  }
339  }
340 
341  // A token is always valid unless:
342  // 1) Memory allocation failed during construction
343  // 2) It was moved via the move constructor
344  // (Note: assignment does a swap, leaving both potentially valid)
345  // 3) The associated queue was destroyed
346  // Note that if valid() returns true, that only indicates
347  // that the token is valid for use with a specific queue,
348  // but not which one; that's up to the user to track.
349  inline bool valid() const { return producer != nullptr; }
350 
352  {
353  if (producer != nullptr) {
354  producer->token = nullptr;
355  producer->inactive.store(true, std::memory_order_release);
356  }
357  }
358 
359  // Disable copying and assignment
360  ProducerToken(ProducerToken const&) = delete;
361  ProducerToken& operator=(ProducerToken const&) = delete;
362 
363 private:
364  template<typename T, typename Traits> friend class ConcurrentQueue;
365 
366 protected:
368 };
369 
370 
372 {
373  template<typename T, typename Traits>
375 
376  ConsumerToken(ConsumerToken&& other) noexcept
378  {
379  }
380 
381  inline ConsumerToken& operator=(ConsumerToken&& other) noexcept
382  {
383  swap(other);
384  return *this;
385  }
386 
387  void swap(ConsumerToken& other) noexcept
388  {
389  std::swap(initialOffset, other.initialOffset);
390  std::swap(lastKnownGlobalOffset, other.lastKnownGlobalOffset);
391  std::swap(itemsConsumedFromCurrent, other.itemsConsumedFromCurrent);
392  std::swap(currentProducer, other.currentProducer);
393  std::swap(desiredProducer, other.desiredProducer);
394  }
395 
396  // Disable copying and assignment
397  ConsumerToken(ConsumerToken const&) = delete;
398  ConsumerToken& operator=(ConsumerToken const&) = delete;
399 
400 private:
401  template<typename T, typename Traits> friend class ConcurrentQueue;
402 
403 private: // but shared with ConcurrentQueue
409 };
410 
411 
412 template<typename T, typename Traits = ConcurrentQueueDefaultTraits>
414 {
415 public:
417 
420 
421  typedef typename Traits::index_t index_t;
422  typedef typename Traits::size_t size_t;
423 
424  static const size_t BLOCK_SIZE = static_cast<size_t>(Traits::BLOCK_SIZE);
426  static const size_t EXPLICIT_INITIAL_INDEX_SIZE = static_cast<size_t>(Traits::EXPLICIT_INITIAL_INDEX_SIZE);
428 #ifdef _MSC_VER
429 #pragma warning(push)
430 #pragma warning(disable: 4307) // + integral constant overflow (that's what the ternary expression is for!)
431 #pragma warning(disable: 4309) // static_cast: Truncation of constant value
432 #endif
434 #ifdef _MSC_VER
435 #pragma warning(pop)
436 #endif
437 
438  static_assert(!std::numeric_limits<size_t>::is_signed && std::is_integral<size_t>::value, "Traits::size_t must be an unsigned integral type");
439  static_assert(!std::numeric_limits<index_t>::is_signed && std::is_integral<index_t>::value, "Traits::index_t must be an unsigned integral type");
440  static_assert(sizeof(index_t) >= sizeof(size_t), "Traits::index_t must be at least as wide as Traits::size_t");
441  static_assert((BLOCK_SIZE > 1) && !(BLOCK_SIZE & (BLOCK_SIZE - 1)), "Traits::BLOCK_SIZE must be a power of 2 (and at least 2)");
442  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)");
443  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)");
444 
445 public:
446  // Creates a queue with at least `capacity` element slots; note that the
447  // actual number of elements that can be inserted without additional memory
448  // allocation depends on the number of producers and the block size (e.g. if
449  // the block size is equal to `capacity`, only a single block will be allocated
450  // up-front, which means only a single producer will be able to enqueue elements
451  // without an extra allocation -- blocks aren't shared between producers).
452  // This method is not thread safe -- it is up to the user to ensure that the
453  // queue is fully constructed before it starts being used by other threads (this
454  // includes making the memory effects of construction visible, possibly with a
455  // memory barrier).
456  explicit ConcurrentQueue(size_t capacity = 6 * BLOCK_SIZE)
457  : producerListTail(nullptr),
458  producerCount(0),
462  {
463  populate_initial_block_list(capacity / BLOCK_SIZE + ((capacity & (BLOCK_SIZE - 1)) == 0 ? 0 : 1));
464  }
465 
466  // Computes the correct amount of pre-allocated blocks for you based
467  // on the minimum number of elements you want available at any given
468  // time, and the maximum concurrent number of each type of producer.
469  ConcurrentQueue(size_t minCapacity, size_t maxExplicitProducers)
470  : producerListTail(nullptr),
471  producerCount(0),
475  {
476  size_t blocks = (((minCapacity + BLOCK_SIZE - 1) / BLOCK_SIZE) - 1) * (maxExplicitProducers + 1) + 2 * (maxExplicitProducers);
478  }
479 
480  // Note: The queue should not be accessed concurrently while it's
481  // being deleted. It's up to the user to synchronize this.
482  // This method is not thread safe.
484  {
485  // Destroy producers
486  auto ptr = producerListTail.load(std::memory_order_relaxed);
487  while (ptr != nullptr) {
488  auto next = ptr->next_prod();
489  if (ptr->token != nullptr) {
490  ptr->token->producer = nullptr;
491  }
492  destroy(ptr);
493  ptr = next;
494  }
495 
496  // Destroy global free list
497  auto block = freeList.head_unsafe();
498  while (block != nullptr) {
499  auto next = block->freeListNext.load(std::memory_order_relaxed);
500  if (block->dynamicallyAllocated) {
501  destroy(block);
502  }
503  block = next;
504  }
505 
506  // Destroy initial free list
508  }
509 
510  // Disable copying and copy assignment
511  ConcurrentQueue(ConcurrentQueue const&) = delete;
512  ConcurrentQueue(ConcurrentQueue&& other) = delete;
513  ConcurrentQueue& operator=(ConcurrentQueue const&) = delete;
514  ConcurrentQueue& operator=(ConcurrentQueue&& other) = delete;
515 
516 public:
517  tracy_force_inline T* enqueue_begin(producer_token_t const& token, index_t& currentTailIndex)
518  {
519  return static_cast<ExplicitProducer*>(token.producer)->ConcurrentQueue::ExplicitProducer::enqueue_begin(currentTailIndex);
520  }
521 
522  template<class NotifyThread, class ProcessData>
523  size_t try_dequeue_bulk_single(consumer_token_t& token, NotifyThread notifyThread, ProcessData processData )
524  {
525  if (token.desiredProducer == nullptr || token.lastKnownGlobalOffset != globalExplicitConsumerOffset.load(std::memory_order_relaxed)) {
527  return 0;
528  }
529  }
530 
531  size_t count = static_cast<ProducerBase*>(token.currentProducer)->dequeue_bulk(notifyThread, processData);
532  token.itemsConsumedFromCurrent += static_cast<std::uint32_t>(count);
533 
534  auto tail = producerListTail.load(std::memory_order_acquire);
535  auto ptr = static_cast<ProducerBase*>(token.currentProducer)->next_prod();
536  if (ptr == nullptr) {
537  ptr = tail;
538  }
539  if( count == 0 )
540  {
541  while (ptr != static_cast<ProducerBase*>(token.currentProducer)) {
542  auto dequeued = ptr->dequeue_bulk(notifyThread, processData);
543  if (dequeued != 0) {
544  token.currentProducer = ptr;
545  token.itemsConsumedFromCurrent = static_cast<std::uint32_t>(dequeued);
546  return dequeued;
547  }
548  ptr = ptr->next_prod();
549  if (ptr == nullptr) {
550  ptr = tail;
551  }
552  }
553  return 0;
554  }
555  else
556  {
557  token.currentProducer = ptr;
558  token.itemsConsumedFromCurrent = 0;
559  return count;
560  }
561  }
562 
563 
564  // Returns an estimate of the total number of elements currently in the queue. This
565  // estimate is only accurate if the queue has completely stabilized before it is called
566  // (i.e. all enqueue and dequeue operations have completed and their memory effects are
567  // visible on the calling thread, and no further operations start while this method is
568  // being called).
569  // Thread-safe.
570  size_t size_approx() const
571  {
572  size_t size = 0;
573  for (auto ptr = producerListTail.load(std::memory_order_acquire); ptr != nullptr; ptr = ptr->next_prod()) {
574  size += ptr->size_approx();
575  }
576  return size;
577  }
578 
579 
580  // Returns true if the underlying atomic variables used by
581  // the queue are lock-free (they should be on most platforms).
582  // Thread-safe.
583  static bool is_lock_free()
584  {
585  return
591  }
592 
593 
594 private:
595  friend struct ProducerToken;
596  friend struct ConsumerToken;
597  friend struct ExplicitProducer;
598 
599 
601  // Queue methods
603 
605  {
606  // Ah, there's been a rotation, figure out where we should be!
607  auto tail = producerListTail.load(std::memory_order_acquire);
608  if (token.desiredProducer == nullptr && tail == nullptr) {
609  return false;
610  }
611  auto prodCount = producerCount.load(std::memory_order_relaxed);
612  auto globalOffset = globalExplicitConsumerOffset.load(std::memory_order_relaxed);
613  if (details::cqUnlikely(token.desiredProducer == nullptr)) {
614  // Aha, first time we're dequeueing anything.
615  // Figure out our local position
616  // Note: offset is from start, not end, but we're traversing from end -- subtract from count first
617  std::uint32_t offset = prodCount - 1 - (token.initialOffset % prodCount);
618  token.desiredProducer = tail;
619  for (std::uint32_t i = 0; i != offset; ++i) {
620  token.desiredProducer = static_cast<ProducerBase*>(token.desiredProducer)->next_prod();
621  if (token.desiredProducer == nullptr) {
622  token.desiredProducer = tail;
623  }
624  }
625  }
626 
627  std::uint32_t delta = globalOffset - token.lastKnownGlobalOffset;
628  if (delta >= prodCount) {
629  delta = delta % prodCount;
630  }
631  for (std::uint32_t i = 0; i != delta; ++i) {
632  token.desiredProducer = static_cast<ProducerBase*>(token.desiredProducer)->next_prod();
633  if (token.desiredProducer == nullptr) {
634  token.desiredProducer = tail;
635  }
636  }
637 
638  token.lastKnownGlobalOffset = globalOffset;
639  token.currentProducer = token.desiredProducer;
640  token.itemsConsumedFromCurrent = 0;
641  return true;
642  }
643 
644 
646  // Free list
648 
649  template <typename N>
651  {
653 
654  std::atomic<std::uint32_t> freeListRefs;
655  std::atomic<N*> freeListNext;
656  };
657 
658  // A simple CAS-based lock-free free list. Not the fastest thing in the world under heavy contention, but
659  // simple and correct (assuming nodes are never freed until after the free list is destroyed), and fairly
660  // speedy under low contention.
661  template<typename N> // N must inherit FreeListNode or have the same fields (and initialization of them)
662  struct FreeList
663  {
664  FreeList() : freeListHead(nullptr) { }
665  FreeList(FreeList&& other) : freeListHead(other.freeListHead.load(std::memory_order_relaxed)) { other.freeListHead.store(nullptr, std::memory_order_relaxed); }
667 
668  FreeList(FreeList const&) = delete;
669  FreeList& operator=(FreeList const&) = delete;
670 
671  inline void add(N* node)
672  {
673  // We know that the should-be-on-freelist bit is 0 at this point, so it's safe to
674  // set it using a fetch_add
675  if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST, std::memory_order_acq_rel) == 0) {
676  // Oh look! We were the last ones referencing this node, and we know
677  // we want to add it to the free list, so let's do it!
679  }
680  }
681 
682  inline N* try_get()
683  {
684  auto head = freeListHead.load(std::memory_order_acquire);
685  while (head != nullptr) {
686  auto prevHead = head;
687  auto refs = head->freeListRefs.load(std::memory_order_relaxed);
688  if ((refs & REFS_MASK) == 0 || !head->freeListRefs.compare_exchange_strong(refs, refs + 1, std::memory_order_acquire, std::memory_order_relaxed)) {
689  head = freeListHead.load(std::memory_order_acquire);
690  continue;
691  }
692 
693  // Good, reference count has been incremented (it wasn't at zero), which means we can read the
694  // next and not worry about it changing between now and the time we do the CAS
695  auto next = head->freeListNext.load(std::memory_order_relaxed);
696  if (freeListHead.compare_exchange_strong(head, next, std::memory_order_acquire, std::memory_order_relaxed)) {
697  // Yay, got the node. This means it was on the list, which means shouldBeOnFreeList must be false no
698  // matter the refcount (because nobody else knows it's been taken off yet, it can't have been put back on).
699  assert((head->freeListRefs.load(std::memory_order_relaxed) & SHOULD_BE_ON_FREELIST) == 0);
700 
701  // Decrease refcount twice, once for our ref, and once for the list's ref
702  head->freeListRefs.fetch_sub(2, std::memory_order_release);
703  return head;
704  }
705 
706  // OK, the head must have changed on us, but we still need to decrease the refcount we increased.
707  // Note that we don't need to release any memory effects, but we do need to ensure that the reference
708  // count decrement happens-after the CAS on the head.
709  refs = prevHead->freeListRefs.fetch_sub(1, std::memory_order_acq_rel);
710  if (refs == SHOULD_BE_ON_FREELIST + 1) {
712  }
713  }
714 
715  return nullptr;
716  }
717 
718  // Useful for traversing the list when there's no contention (e.g. to destroy remaining nodes)
719  N* head_unsafe() const { return freeListHead.load(std::memory_order_relaxed); }
720 
721  private:
722  inline void add_knowing_refcount_is_zero(N* node)
723  {
724  // Since the refcount is zero, and nobody can increase it once it's zero (except us, and we run
725  // only one copy of this method per node at a time, i.e. the single thread case), then we know
726  // we can safely change the next pointer of the node; however, once the refcount is back above
727  // zero, then other threads could increase it (happens under heavy contention, when the refcount
728  // goes to zero in between a load and a refcount increment of a node in try_get, then back up to
729  // something non-zero, then the refcount increment is done by the other thread) -- so, if the CAS
730  // to add the node to the actual list fails, decrease the refcount and leave the add operation to
731  // the next thread who puts the refcount back at zero (which could be us, hence the loop).
732  auto head = freeListHead.load(std::memory_order_relaxed);
733  while (true) {
734  node->freeListNext.store(head, std::memory_order_relaxed);
735  node->freeListRefs.store(1, std::memory_order_release);
736  if (!freeListHead.compare_exchange_strong(head, node, std::memory_order_release, std::memory_order_relaxed)) {
737  // Hmm, the add failed, but we can only try again when the refcount goes back to zero
738  if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST - 1, std::memory_order_release) == 1) {
739  continue;
740  }
741  }
742  return;
743  }
744  }
745 
746  private:
747  // Implemented like a stack, but where node order doesn't matter (nodes are inserted out of order under contention)
748  std::atomic<N*> freeListHead;
749 
750  static const std::uint32_t REFS_MASK = 0x7FFFFFFF;
751  static const std::uint32_t SHOULD_BE_ON_FREELIST = 0x80000000;
752  };
753 
754 
756  // Block
758 
759  struct Block
760  {
763  {
764  }
765 
766  inline bool is_empty() const
767  {
769  // Check flags
770  for (size_t i = 0; i < BLOCK_SIZE; ++i) {
771  if (!emptyFlags[i].load(std::memory_order_relaxed)) {
772  return false;
773  }
774  }
775 
776  // Aha, empty; make sure we have all other memory effects that happened before the empty flags were set
777  std::atomic_thread_fence(std::memory_order_acquire);
778  return true;
779  }
780  else {
781  // Check counter
782  if (elementsCompletelyDequeued.load(std::memory_order_relaxed) == BLOCK_SIZE) {
783  std::atomic_thread_fence(std::memory_order_acquire);
784  return true;
785  }
786  assert(elementsCompletelyDequeued.load(std::memory_order_relaxed) <= BLOCK_SIZE);
787  return false;
788  }
789  }
790 
791  // Returns true if the block is now empty (does not apply in explicit context)
792  inline bool set_empty(index_t i)
793  {
794  if (BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD) {
795  // Set flag
796  assert(!emptyFlags[BLOCK_SIZE - 1 - static_cast<size_t>(i & static_cast<index_t>(BLOCK_SIZE - 1))].load(std::memory_order_relaxed));
797  emptyFlags[BLOCK_SIZE - 1 - static_cast<size_t>(i & static_cast<index_t>(BLOCK_SIZE - 1))].store(true, std::memory_order_release);
798  return false;
799  }
800  else {
801  // Increment counter
802  auto prevVal = elementsCompletelyDequeued.fetch_add(1, std::memory_order_release);
803  assert(prevVal < BLOCK_SIZE);
804  return prevVal == BLOCK_SIZE - 1;
805  }
806  }
807 
808  // Sets multiple contiguous item statuses to 'empty' (assumes no wrapping and count > 0).
809  // Returns true if the block is now empty (does not apply in explicit context).
810  inline bool set_many_empty(index_t i, size_t count)
811  {
813  // Set flags
814  std::atomic_thread_fence(std::memory_order_release);
815  i = BLOCK_SIZE - 1 - static_cast<size_t>(i & static_cast<index_t>(BLOCK_SIZE - 1)) - count + 1;
816  for (size_t j = 0; j != count; ++j) {
817  assert(!emptyFlags[i + j].load(std::memory_order_relaxed));
818  emptyFlags[i + j].store(true, std::memory_order_relaxed);
819  }
820  return false;
821  }
822  else {
823  // Increment counter
824  auto prevVal = elementsCompletelyDequeued.fetch_add(count, std::memory_order_release);
825  assert(prevVal + count <= BLOCK_SIZE);
826  return prevVal + count == BLOCK_SIZE;
827  }
828  }
829 
830  inline void set_all_empty()
831  {
832  if (BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD) {
833  // Set all flags
834  for (size_t i = 0; i != BLOCK_SIZE; ++i) {
835  emptyFlags[i].store(true, std::memory_order_relaxed);
836  }
837  }
838  else {
839  // Reset counter
840  elementsCompletelyDequeued.store(BLOCK_SIZE, std::memory_order_relaxed);
841  }
842  }
843 
844  inline void reset_empty()
845  {
847  // Reset flags
848  for (size_t i = 0; i != BLOCK_SIZE; ++i) {
849  emptyFlags[i].store(false, std::memory_order_relaxed);
850  }
851  }
852  else {
853  // Reset counter
854  elementsCompletelyDequeued.store(0, std::memory_order_relaxed);
855  }
856  }
857 
858  inline T* operator[](index_t idx) noexcept { return static_cast<T*>(static_cast<void*>(elements)) + static_cast<size_t>(idx & static_cast<index_t>(BLOCK_SIZE - 1)); }
859  inline T const* operator[](index_t idx) const noexcept { return static_cast<T const*>(static_cast<void const*>(elements)) + static_cast<size_t>(idx & static_cast<index_t>(BLOCK_SIZE - 1)); }
860 
861  private:
862  // IMPORTANT: This must be the first member in Block, so that if T depends on the alignment of
863  // addresses returned by malloc, that alignment will be preserved. Apparently clang actually
864  // generates code that uses this assumption for AVX instructions in some cases. Ideally, we
865  // should also align Block to the alignment of T in case it's higher than malloc's 16-byte
866  // alignment, but this is hard to do in a cross-platform way. Assert for this case:
867  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");
868  // Additionally, we need the alignment of Block itself to be a multiple of max_align_t since
869  // otherwise the appropriate padding will not be added at the end of Block in order to make
870  // arrays of Blocks all be properly aligned (not just the first one). We use a union to force
871  // this.
872  union {
873  char elements[sizeof(T) * BLOCK_SIZE];
875  };
876  public:
878  std::atomic<size_t> elementsCompletelyDequeued;
879  std::atomic<bool> emptyFlags[BLOCK_SIZE <= EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD ? BLOCK_SIZE : 1];
880  public:
881  std::atomic<std::uint32_t> freeListRefs;
882  std::atomic<Block*> freeListNext;
883  std::atomic<bool> shouldBeOnFreeList;
884  bool dynamicallyAllocated; // Perhaps a better name for this would be 'isNotPartOfInitialBlockPool'
885  };
886  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");
887 
888 
890  // Producer base
892 
894  {
896  tailIndex(0),
897  headIndex(0),
900  tailBlock(nullptr),
901  parent(parent_)
902  {
903  }
904 
905  virtual ~ProducerBase() { };
906 
907  template<class NotifyThread, class ProcessData>
908  inline size_t dequeue_bulk(NotifyThread notifyThread, ProcessData processData)
909  {
910  return static_cast<ExplicitProducer*>(this)->dequeue_bulk(notifyThread, processData);
911  }
912 
913  inline ProducerBase* next_prod() const { return static_cast<ProducerBase*>(next); }
914 
915  inline size_t size_approx() const
916  {
917  auto tail = tailIndex.load(std::memory_order_relaxed);
918  auto head = headIndex.load(std::memory_order_relaxed);
919  return details::circular_less_than(head, tail) ? static_cast<size_t>(tail - head) : 0;
920  }
921 
922  inline index_t getTail() const { return tailIndex.load(std::memory_order_relaxed); }
923  protected:
924  std::atomic<index_t> tailIndex; // Where to enqueue to next
925  std::atomic<index_t> headIndex; // Where to dequeue from next
926 
927  std::atomic<index_t> dequeueOptimisticCount;
928  std::atomic<index_t> dequeueOvercommit;
929 
931 
932  public:
934  };
935 
936 
937  public:
939  // Explicit queue
942  {
943  explicit ExplicitProducer(ConcurrentQueue* _parent) :
944  ProducerBase(_parent),
945  blockIndex(nullptr),
949  pr_blockIndexEntries(nullptr),
950  pr_blockIndexRaw(nullptr)
951  {
952  size_t poolBasedIndexSize = details::ceil_to_pow_2(_parent->initialBlockPoolSize) >> 1;
953  if (poolBasedIndexSize > pr_blockIndexSize) {
954  pr_blockIndexSize = poolBasedIndexSize;
955  }
956 
957  new_block_index(0); // This creates an index with double the number of current entries, i.e. EXPLICIT_INITIAL_INDEX_SIZE
958  }
959 
961  {
962  // Destruct any elements not yet dequeued.
963  // Since we're in the destructor, we can assume all elements
964  // are either completely dequeued or completely not (no halfways).
965  if (this->tailBlock != nullptr) { // Note this means there must be a block index too
966  // First find the block that's partially dequeued, if any
967  Block* halfDequeuedBlock = nullptr;
968  if ((this->headIndex.load(std::memory_order_relaxed) & static_cast<index_t>(BLOCK_SIZE - 1)) != 0) {
969  // The head's not on a block boundary, meaning a block somewhere is partially dequeued
970  // (or the head block is the tail block and was fully dequeued, but the head/tail are still not on a boundary)
972  while (details::circular_less_than<index_t>(pr_blockIndexEntries[i].base + BLOCK_SIZE, this->headIndex.load(std::memory_order_relaxed))) {
973  i = (i + 1) & (pr_blockIndexSize - 1);
974  }
975  assert(details::circular_less_than<index_t>(pr_blockIndexEntries[i].base, this->headIndex.load(std::memory_order_relaxed)));
976  halfDequeuedBlock = pr_blockIndexEntries[i].block;
977  }
978 
979  // Start at the head block (note the first line in the loop gives us the head from the tail on the first iteration)
980  auto block = this->tailBlock;
981  do {
982  block = block->next;
983  if (block->ConcurrentQueue::Block::is_empty()) {
984  continue;
985  }
986 
987  size_t i = 0; // Offset into block
988  if (block == halfDequeuedBlock) {
989  i = static_cast<size_t>(this->headIndex.load(std::memory_order_relaxed) & static_cast<index_t>(BLOCK_SIZE - 1));
990  }
991 
992  // Walk through all the items in the block; if this is the tail block, we need to stop when we reach the tail index
993  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));
994  while (i != BLOCK_SIZE && (block != this->tailBlock || i != lastValidIndex)) {
995  (*block)[i++]->~T();
996  }
997  } while (block != this->tailBlock);
998  }
999 
1000  // Destroy all blocks that we own
1001  if (this->tailBlock != nullptr) {
1002  auto block = this->tailBlock;
1003  do {
1004  auto nextBlock = block->next;
1005  if (block->dynamicallyAllocated) {
1006  destroy(block);
1007  }
1008  else {
1009  this->parent->add_block_to_free_list(block);
1010  }
1011  block = nextBlock;
1012  } while (block != this->tailBlock);
1013  }
1014 
1015  // Destroy the block indices
1016  auto header = static_cast<BlockIndexHeader*>(pr_blockIndexRaw);
1017  while (header != nullptr) {
1018  auto prev = static_cast<BlockIndexHeader*>(header->prev);
1019  header->~BlockIndexHeader();
1020  (Traits::free)(header);
1021  header = prev;
1022  }
1023  }
1024 
1025  inline void enqueue_begin_alloc(index_t currentTailIndex)
1026  {
1027  // We reached the end of a block, start a new one
1028  if (this->tailBlock != nullptr && this->tailBlock->next->ConcurrentQueue::Block::is_empty()) {
1029  // We can re-use the block ahead of us, it's empty!
1030  this->tailBlock = this->tailBlock->next;
1031  this->tailBlock->ConcurrentQueue::Block::reset_empty();
1032 
1033  // We'll put the block on the block index (guaranteed to be room since we're conceptually removing the
1034  // last block from it first -- except instead of removing then adding, we can just overwrite).
1035  // Note that there must be a valid block index here, since even if allocation failed in the ctor,
1036  // it would have been re-attempted when adding the first block to the queue; since there is such
1037  // a block, a block index must have been successfully allocated.
1038  }
1039  else {
1040  // We're going to need a new block; check that the block index has room
1042  // Hmm, the circular block index is already full -- we'll need
1043  // to allocate a new index. Note pr_blockIndexRaw can only be nullptr if
1044  // the initial allocation failed in the constructor.
1046  }
1047 
1048  // Insert a new block in the circular linked list
1049  auto newBlock = this->parent->ConcurrentQueue::requisition_block();
1050  newBlock->ConcurrentQueue::Block::reset_empty();
1051  if (this->tailBlock == nullptr) {
1052  newBlock->next = newBlock;
1053  }
1054  else {
1055  newBlock->next = this->tailBlock->next;
1056  this->tailBlock->next = newBlock;
1057  }
1058  this->tailBlock = newBlock;
1060  }
1061 
1062  // Add block to block index
1063  auto& entry = blockIndex.load(std::memory_order_relaxed)->entries[pr_blockIndexFront];
1064  entry.base = currentTailIndex;
1065  entry.block = this->tailBlock;
1066  blockIndex.load(std::memory_order_relaxed)->front.store(pr_blockIndexFront, std::memory_order_release);
1068  }
1069 
1071  {
1072  currentTailIndex = this->tailIndex.load(std::memory_order_relaxed);
1073  if (details::cqUnlikely((currentTailIndex & static_cast<index_t>(BLOCK_SIZE - 1)) == 0)) {
1074  this->enqueue_begin_alloc(currentTailIndex);
1075  }
1076  return (*this->tailBlock)[currentTailIndex];
1077  }
1078 
1079  tracy_force_inline std::atomic<index_t>& get_tail_index()
1080  {
1081  return this->tailIndex;
1082  }
1083 
1084  template<class NotifyThread, class ProcessData>
1085  size_t dequeue_bulk(NotifyThread notifyThread, ProcessData processData)
1086  {
1087  auto tail = this->tailIndex.load(std::memory_order_relaxed);
1088  auto overcommit = this->dequeueOvercommit.load(std::memory_order_relaxed);
1089  auto desiredCount = static_cast<size_t>(tail - (this->dequeueOptimisticCount.load(std::memory_order_relaxed) - overcommit));
1090  if (details::circular_less_than<size_t>(0, desiredCount)) {
1091  desiredCount = desiredCount < 8192 ? desiredCount : 8192;
1092  std::atomic_thread_fence(std::memory_order_acquire);
1093 
1094  auto myDequeueCount = this->dequeueOptimisticCount.fetch_add(desiredCount, std::memory_order_relaxed);
1095  assert(overcommit <= myDequeueCount);
1096 
1097  tail = this->tailIndex.load(std::memory_order_acquire);
1098  auto actualCount = static_cast<size_t>(tail - (myDequeueCount - overcommit));
1099  if (details::circular_less_than<size_t>(0, actualCount)) {
1100  actualCount = desiredCount < actualCount ? desiredCount : actualCount;
1101  if (actualCount < desiredCount) {
1102  this->dequeueOvercommit.fetch_add(desiredCount - actualCount, std::memory_order_release);
1103  }
1104 
1105  // Get the first index. Note that since there's guaranteed to be at least actualCount elements, this
1106  // will never exceed tail.
1107  auto firstIndex = this->headIndex.fetch_add(actualCount, std::memory_order_acq_rel);
1108 
1109  // Determine which block the first element is in
1110  auto localBlockIndex = blockIndex.load(std::memory_order_acquire);
1111  auto localBlockIndexHead = localBlockIndex->front.load(std::memory_order_acquire);
1112 
1113  auto headBase = localBlockIndex->entries[localBlockIndexHead].base;
1114  auto firstBlockBaseIndex = firstIndex & ~static_cast<index_t>(BLOCK_SIZE - 1);
1115  auto offset = static_cast<size_t>(static_cast<typename std::make_signed<index_t>::type>(firstBlockBaseIndex - headBase) / BLOCK_SIZE);
1116  auto indexIndex = (localBlockIndexHead + offset) & (localBlockIndex->size - 1);
1117 
1118  notifyThread( this->threadId );
1119 
1120  // Iterate the blocks and dequeue
1121  auto index = firstIndex;
1122  do {
1123  auto firstIndexInBlock = index;
1124  auto endIndex = (index & ~static_cast<index_t>(BLOCK_SIZE - 1)) + static_cast<index_t>(BLOCK_SIZE);
1125  endIndex = details::circular_less_than<index_t>(firstIndex + static_cast<index_t>(actualCount), endIndex) ? firstIndex + static_cast<index_t>(actualCount) : endIndex;
1126  auto block = localBlockIndex->entries[indexIndex].block;
1127 
1128  const auto sz = endIndex - index;
1129  processData( (*block)[index], sz );
1130  index += sz;
1131 
1132  block->ConcurrentQueue::Block::set_many_empty(firstIndexInBlock, static_cast<size_t>(endIndex - firstIndexInBlock));
1133  indexIndex = (indexIndex + 1) & (localBlockIndex->size - 1);
1134  } while (index != firstIndex + actualCount);
1135 
1136  return actualCount;
1137  }
1138  else {
1139  // Wasn't anything to dequeue after all; make the effective dequeue count eventually consistent
1140  this->dequeueOvercommit.fetch_add(desiredCount, std::memory_order_release);
1141  }
1142  }
1143 
1144  return 0;
1145  }
1146 
1147  private:
1149  {
1152  };
1153 
1155  {
1156  size_t size;
1157  std::atomic<size_t> front; // Current slot (not next, like pr_blockIndexFront)
1159  void* prev;
1160  };
1161 
1162 
1163  bool new_block_index(size_t numberOfFilledSlotsToExpose)
1164  {
1165  auto prevBlockSizeMask = pr_blockIndexSize - 1;
1166 
1167  // Create the new block
1168  pr_blockIndexSize <<= 1;
1169  auto newRawPtr = static_cast<char*>((Traits::malloc)(sizeof(BlockIndexHeader) + std::alignment_of<BlockIndexEntry>::value - 1 + sizeof(BlockIndexEntry) * pr_blockIndexSize));
1170  if (newRawPtr == nullptr) {
1171  pr_blockIndexSize >>= 1; // Reset to allow graceful retry
1172  return false;
1173  }
1174 
1175  auto newBlockIndexEntries = reinterpret_cast<BlockIndexEntry*>(details::align_for<BlockIndexEntry>(newRawPtr + sizeof(BlockIndexHeader)));
1176 
1177  // Copy in all the old indices, if any
1178  size_t j = 0;
1179  if (pr_blockIndexSlotsUsed != 0) {
1180  auto i = (pr_blockIndexFront - pr_blockIndexSlotsUsed) & prevBlockSizeMask;
1181  do {
1182  newBlockIndexEntries[j++] = pr_blockIndexEntries[i];
1183  i = (i + 1) & prevBlockSizeMask;
1184  } while (i != pr_blockIndexFront);
1185  }
1186 
1187  // Update everything
1188  auto header = new (newRawPtr) BlockIndexHeader;
1189  header->size = pr_blockIndexSize;
1190  header->front.store(numberOfFilledSlotsToExpose - 1, std::memory_order_relaxed);
1191  header->entries = newBlockIndexEntries;
1192  header->prev = pr_blockIndexRaw; // we link the new block to the old one so we can free it later
1193 
1194  pr_blockIndexFront = j;
1195  pr_blockIndexEntries = newBlockIndexEntries;
1196  pr_blockIndexRaw = newRawPtr;
1197  blockIndex.store(header, std::memory_order_release);
1198 
1199  return true;
1200  }
1201 
1202  private:
1203  std::atomic<BlockIndexHeader*> blockIndex;
1204 
1205  // To be used by producer only -- consumer must use the ones in referenced by blockIndex
1208  size_t pr_blockIndexFront; // Next slot (not current)
1211  };
1212 
1214  {
1215  return static_cast<ExplicitProducer*>(token.producer);
1216  }
1217 
1218  private:
1219 
1221  // Block pool manipulation
1223 
1224  void populate_initial_block_list(size_t blockCount)
1225  {
1226  initialBlockPoolSize = blockCount;
1227  if (initialBlockPoolSize == 0) {
1228  initialBlockPool = nullptr;
1229  return;
1230  }
1231 
1232  initialBlockPool = create_array<Block>(blockCount);
1233  if (initialBlockPool == nullptr) {
1235  }
1236  for (size_t i = 0; i < initialBlockPoolSize; ++i) {
1238  }
1239  }
1240 
1242  {
1243  if (initialBlockPoolIndex.load(std::memory_order_relaxed) >= initialBlockPoolSize) {
1244  return nullptr;
1245  }
1246 
1247  auto index = initialBlockPoolIndex.fetch_add(1, std::memory_order_relaxed);
1248 
1249  return index < initialBlockPoolSize ? (initialBlockPool + index) : nullptr;
1250  }
1251 
1252  inline void add_block_to_free_list(Block* block)
1253  {
1254  freeList.add(block);
1255  }
1256 
1257  inline void add_blocks_to_free_list(Block* block)
1258  {
1259  while (block != nullptr) {
1260  auto next = block->next;
1261  add_block_to_free_list(block);
1262  block = next;
1263  }
1264  }
1265 
1267  {
1268  return freeList.try_get();
1269  }
1270 
1271  // Gets a free block from one of the memory pools, or allocates a new one (if applicable)
1273  {
1274  auto block = try_get_block_from_initial_pool();
1275  if (block != nullptr) {
1276  return block;
1277  }
1278 
1279  block = try_get_block_from_free_list();
1280  if (block != nullptr) {
1281  return block;
1282  }
1283 
1284  return create<Block>();
1285  }
1286 
1287 
1289  // Producer list manipulation
1291 
1293  {
1294  bool recycled;
1295  return recycle_or_create_producer(recycled);
1296  }
1297 
1299  {
1300  // Try to re-use one first
1301  for (auto ptr = producerListTail.load(std::memory_order_acquire); ptr != nullptr; ptr = ptr->next_prod()) {
1302  if (ptr->inactive.load(std::memory_order_relaxed)) {
1303  if( ptr->size_approx() == 0 )
1304  {
1305  bool expected = true;
1306  if (ptr->inactive.compare_exchange_strong(expected, /* desired */ false, std::memory_order_acquire, std::memory_order_relaxed)) {
1307  // We caught one! It's been marked as activated, the caller can have it
1308  recycled = true;
1309  return ptr;
1310  }
1311  }
1312  }
1313  }
1314 
1315  recycled = false;
1316  return add_producer(static_cast<ProducerBase*>(create<ExplicitProducer>(this)));
1317  }
1318 
1320  {
1321  // Handle failed memory allocation
1322  if (producer == nullptr) {
1323  return nullptr;
1324  }
1325 
1326  producerCount.fetch_add(1, std::memory_order_relaxed);
1327 
1328  // Add it to the lock-free list
1329  auto prevTail = producerListTail.load(std::memory_order_relaxed);
1330  do {
1331  producer->next = prevTail;
1332  } while (!producerListTail.compare_exchange_weak(prevTail, producer, std::memory_order_release, std::memory_order_relaxed));
1333 
1334  return producer;
1335  }
1336 
1338  {
1339  // After another instance is moved-into/swapped-with this one, all the
1340  // producers we stole still think their parents are the other queue.
1341  // So fix them up!
1342  for (auto ptr = producerListTail.load(std::memory_order_relaxed); ptr != nullptr; ptr = ptr->next_prod()) {
1343  ptr->parent = this;
1344  }
1345  }
1346 
1348  // Utility functions
1350 
1351  template<typename U>
1352  static inline U* create_array(size_t count)
1353  {
1354  assert(count > 0);
1355  return static_cast<U*>((Traits::malloc)(sizeof(U) * count));
1356  }
1357 
1358  template<typename U>
1359  static inline void destroy_array(U* p, size_t count)
1360  {
1361  ((void)count);
1362  if (p != nullptr) {
1363  assert(count > 0);
1364  (Traits::free)(p);
1365  }
1366  }
1367 
1368  template<typename U>
1369  static inline U* create()
1370  {
1371  auto p = (Traits::malloc)(sizeof(U));
1372  return new (p) U;
1373  }
1374 
1375  template<typename U, typename A1>
1376  static inline U* create(A1&& a1)
1377  {
1378  auto p = (Traits::malloc)(sizeof(U));
1379  return new (p) U(std::forward<A1>(a1));
1380  }
1381 
1382  template<typename U>
1383  static inline void destroy(U* p)
1384  {
1385  if (p != nullptr) {
1386  p->~U();
1387  }
1388  (Traits::free)(p);
1389  }
1390 
1391 private:
1392  std::atomic<ProducerBase*> producerListTail;
1393  std::atomic<std::uint32_t> producerCount;
1394 
1395  std::atomic<size_t> initialBlockPoolIndex;
1398 
1400 
1401  std::atomic<std::uint32_t> nextExplicitConsumerId;
1402  std::atomic<std::uint32_t> globalExplicitConsumerOffset;
1403 };
1404 
1405 
1406 template<typename T, typename Traits>
1408  : producer(queue.recycle_or_create_producer())
1409 {
1410  if (producer != nullptr) {
1411  producer->token = this;
1413  }
1414 }
1415 
1416 template<typename T, typename Traits>
1418  : itemsConsumedFromCurrent(0), currentProducer(nullptr), desiredProducer(nullptr)
1419 {
1420  initialOffset = queue.nextExplicitConsumerId.fetch_add(1, std::memory_order_release);
1421  lastKnownGlobalOffset = static_cast<std::uint32_t>(-1);
1422 }
1423 
1424 template<typename T, typename Traits>
1426 {
1427  a.swap(b);
1428 }
1429 
1430 inline void swap(ProducerToken& a, ProducerToken& b) noexcept
1431 {
1432  a.swap(b);
1433 }
1434 
1435 inline void swap(ConsumerToken& a, ConsumerToken& b) noexcept
1436 {
1437  a.swap(b);
1438 }
1439 
1440 }
1441 
1442 } /* namespace tracy */
1443 
1444 #if defined(__GNUC__)
1445 #pragma GCC diagnostic pop
1446 #endif


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