folly/MPMCQueue.h

   1 /*
   2  * Copyright 2017 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #pragma once
  18
  19 #include <algorithm>
  20 #include <atomic>
  21 #include <cassert>
  22 #include <cstring>
  23 #include <limits>
  24 #include <type_traits>
  25
  26 #include <boost/noncopyable.hpp>
  27
  28 #include <folly/Traits.h>
  29 #include <folly/concurrency/CacheLocality.h>
  30 #include <folly/detail/TurnSequencer.h>
  31 #include <folly/portability/Unistd.h>
  32
  33 namespace folly {
  34
  35 namespace detail {
  36
  37 template <typename T, template <typename> class Atom>
  38 struct SingleElementQueue;
  39
  40 template <typename T> class MPMCPipelineStageImpl;
  41
  42 /// MPMCQueue base CRTP template
  43 template <typename> class MPMCQueueBase;
  44
  45 } // namespace detail
  46
  47 /// MPMCQueue<T> is a high-performance bounded concurrent queue that
  48 /// supports multiple producers, multiple consumers, and optional blocking.
  49 /// The queue has a fixed capacity, for which all memory will be allocated
  50 /// up front.  The bulk of the work of enqueuing and dequeuing can be
  51 /// performed in parallel.
  52 ///
  53 /// MPMCQueue is linearizable.  That means that if a call to write(A)
  54 /// returns before a call to write(B) begins, then A will definitely end up
  55 /// in the queue before B, and if a call to read(X) returns before a call
  56 /// to read(Y) is started, that X will be something from earlier in the
  57 /// queue than Y.  This also means that if a read call returns a value, you
  58 /// can be sure that all previous elements of the queue have been assigned
  59 /// a reader (that reader might not yet have returned, but it exists).
  60 ///
  61 /// The underlying implementation uses a ticket dispenser for the head and
  62 /// the tail, spreading accesses across N single-element queues to produce
  63 /// a queue with capacity N.  The ticket dispensers use atomic increment,
  64 /// which is more robust to contention than a CAS loop.  Each of the
  65 /// single-element queues uses its own CAS to serialize access, with an
  66 /// adaptive spin cutoff.  When spinning fails on a single-element queue
  67 /// it uses futex()'s _BITSET operations to reduce unnecessary wakeups
  68 /// even if multiple waiters are present on an individual queue (such as
  69 /// when the MPMCQueue's capacity is smaller than the number of enqueuers
  70 /// or dequeuers).
  71 ///
  72 /// In benchmarks (contained in tao/queues/ConcurrentQueueTests)
  73 /// it handles 1 to 1, 1 to N, N to 1, and N to M thread counts better
  74 /// than any of the alternatives present in fbcode, for both small (~10)
  75 /// and large capacities.  In these benchmarks it is also faster than
  76 /// tbb::concurrent_bounded_queue for all configurations.  When there are
  77 /// many more threads than cores, MPMCQueue is _much_ faster than the tbb
  78 /// queue because it uses futex() to block and unblock waiting threads,
  79 /// rather than spinning with sched_yield.
  80 ///
  81 /// NOEXCEPT INTERACTION: tl;dr; If it compiles you're fine.  Ticket-based
  82 /// queues separate the assignment of queue positions from the actual
  83 /// construction of the in-queue elements, which means that the T
  84 /// constructor used during enqueue must not throw an exception.  This is
  85 /// enforced at compile time using type traits, which requires that T be
  86 /// adorned with accurate noexcept information.  If your type does not
  87 /// use noexcept, you will have to wrap it in something that provides
  88 /// the guarantee.  We provide an alternate safe implementation for types
  89 /// that don't use noexcept but that are marked folly::IsRelocatable
  90 /// and std::is_nothrow_constructible, which is common for folly types.
  91 /// In particular, if you can declare FOLLY_ASSUME_FBVECTOR_COMPATIBLE
  92 /// then your type can be put in MPMCQueue.
  93 ///
  94 /// If you have a pool of N queue consumers that you want to shut down
  95 /// after the queue has drained, one way is to enqueue N sentinel values
  96 /// to the queue.  If the producer doesn't know how many consumers there
  97 /// are you can enqueue one sentinel and then have each consumer requeue
  98 /// two sentinels after it receives it (by requeuing 2 the shutdown can
  99 /// complete in O(log P) time instead of O(P)).
 100 template <
 101     typename T,
 102     template <typename> class Atom = std::atomic,
 103     bool Dynamic = false>
 104 class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> {
 105   friend class detail::MPMCPipelineStageImpl<T>;
 106   using Slot = detail::SingleElementQueue<T,Atom>;
 107  public:
 108
 109   explicit MPMCQueue(size_t queueCapacity)
 110     : detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>>(queueCapacity)
 111   {
 112     this->stride_ = this->computeStride(queueCapacity);
 113     this->slots_ = new Slot[queueCapacity + 2 * this->kSlotPadding];
 114   }
 115
 116   MPMCQueue() noexcept { }
 117 };
 118
 119 /// The dynamic version of MPMCQueue allows dynamic expansion of queue
 120 /// capacity, such that a queue may start with a smaller capacity than
 121 /// specified and expand only if needed. Users may optionally specify
 122 /// the initial capacity and the expansion multiplier.
 123 ///
 124 /// The design uses a seqlock to enforce mutual exclusion among
 125 /// expansion attempts. Regular operations read up-to-date queue
 126 /// information (slots array, capacity, stride) inside read-only
 127 /// seqlock sections, which are unimpeded when no expansion is in
 128 /// progress.
 129 ///
 130 /// An expansion computes a new capacity, allocates a new slots array,
 131 /// and updates stride. No information needs to be copied from the
 132 /// current slots array to the new one. When this happens, new slots
 133 /// will not have sequence numbers that match ticket numbers. The
 134 /// expansion needs to compute a ticket offset such that operations
 135 /// that use new arrays can adjust the calculations of slot indexes
 136 /// and sequence numbers that take into account that the new slots
 137 /// start with sequence numbers of zero. The current ticket offset is
 138 /// packed with the seqlock in an atomic 64-bit integer. The initial
 139 /// offset is zero.
 140 ///
 141 /// Lagging write and read operations with tickets lower than the
 142 /// ticket offset of the current slots array (i.e., the minimum ticket
 143 /// number that can be served by the current array) must use earlier
 144 /// closed arrays instead of the current one. Information about closed
 145 /// slots arrays (array address, capacity, stride, and offset) is
 146 /// maintained in a logarithmic-sized structure. Each entry in that
 147 /// structure never needs to be changed once set. The number of closed
 148 /// arrays is half the value of the seqlock (when unlocked).
 149 ///
 150 /// The acquisition of the seqlock to perform an expansion does not
 151 /// prevent the issuing of new push and pop tickets concurrently. The
 152 /// expansion must set the new ticket offset to a value that couldn't
 153 /// have been issued to an operation that has already gone through a
 154 /// seqlock read-only section (and hence obtained information for
 155 /// older closed arrays).
 156 ///
 157 /// Note that the total queue capacity can temporarily exceed the
 158 /// specified capacity when there are lagging consumers that haven't
 159 /// yet consumed all the elements in closed arrays. Users should not
 160 /// rely on the capacity of dynamic queues for synchronization, e.g.,
 161 /// they should not expect that a thread will definitely block on a
 162 /// call to blockingWrite() when the queue size is known to be equal
 163 /// to its capacity.
 164 ///
 165 /// Note that some writeIfNotFull() and tryWriteUntil() operations may
 166 /// fail even if the size of the queue is less than its maximum
 167 /// capacity and despite the success of expansion, if the operation
 168 /// happens to acquire a ticket that belongs to a closed array. This
 169 /// is a transient condition. Typically, one or two ticket values may
 170 /// be subject to such condition per expansion.
 171 ///
 172 /// The dynamic version is a partial specialization of MPMCQueue with
 173 /// Dynamic == true
 174 template <typename T, template <typename> class Atom>
 175 class MPMCQueue<T,Atom,true> :
 176       public detail::MPMCQueueBase<MPMCQueue<T,Atom,true>> {
 177   friend class detail::MPMCQueueBase<MPMCQueue<T,Atom,true>>;
 178   using Slot = detail::SingleElementQueue<T,Atom>;
 179
 180   struct ClosedArray {
 181     uint64_t offset_ {0};
 182     Slot* slots_ {nullptr};
 183     size_t capacity_ {0};
 184     int stride_ {0};
 185   };
 186
 187  public:
 188
 189   explicit MPMCQueue(size_t queueCapacity)
 190     : detail::MPMCQueueBase<MPMCQueue<T,Atom,true>>(queueCapacity)
 191   {
 192     size_t cap = std::min<size_t>(kDefaultMinDynamicCapacity, queueCapacity);
 193     initQueue(cap, kDefaultExpansionMultiplier);
 194   }
 195
 196   explicit MPMCQueue(size_t queueCapacity,
 197                      size_t minCapacity,
 198                      size_t expansionMultiplier)
 199     : detail::MPMCQueueBase<MPMCQueue<T,Atom,true>>(queueCapacity)
 200   {
 201     minCapacity = std::max<size_t>(1, minCapacity);
 202     size_t cap = std::min<size_t>(minCapacity, queueCapacity);
 203     expansionMultiplier = std::max<size_t>(2, expansionMultiplier);
 204     initQueue(cap, expansionMultiplier);
 205   }
 206
 207   MPMCQueue() noexcept {
 208     dmult_ = 0;
 209     closed_ = nullptr;
 210   }
 211
 212   MPMCQueue(MPMCQueue<T,Atom,true>&& rhs) noexcept {
 213     this->capacity_ = rhs.capacity_;
 214     this->slots_ = rhs.slots_;
 215     this->stride_ = rhs.stride_;
 216     this->dstate_.store(rhs.dstate_.load(std::memory_order_relaxed),
 217                         std::memory_order_relaxed);
 218     this->dcapacity_.store(rhs.dcapacity_.load(std::memory_order_relaxed),
 219                            std::memory_order_relaxed);
 220     this->pushTicket_.store(rhs.pushTicket_.load(std::memory_order_relaxed),
 221                             std::memory_order_relaxed);
 222     this->popTicket_.store(rhs.popTicket_.load(std::memory_order_relaxed),
 223                            std::memory_order_relaxed);
 224     this->pushSpinCutoff_.store(
 225       rhs.pushSpinCutoff_.load(std::memory_order_relaxed),
 226       std::memory_order_relaxed);
 227     this->popSpinCutoff_.store(
 228       rhs.popSpinCutoff_.load(std::memory_order_relaxed),
 229       std::memory_order_relaxed);
 230     dmult_ = rhs.dmult_;
 231     closed_ = rhs.closed_;
 232
 233     rhs.capacity_ = 0;
 234     rhs.slots_ = nullptr;
 235     rhs.stride_ = 0;
 236     rhs.dstate_.store(0, std::memory_order_relaxed);
 237     rhs.dcapacity_.store(0, std::memory_order_relaxed);
 238     rhs.pushTicket_.store(0, std::memory_order_relaxed);
 239     rhs.popTicket_.store(0, std::memory_order_relaxed);
 240     rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
 241     rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
 242     rhs.dmult_ = 0;
 243     rhs.closed_ = nullptr;
 244   }
 245
 246   MPMCQueue<T,Atom, true> const& operator= (MPMCQueue<T,Atom, true>&& rhs) {
 247     if (this != &rhs) {
 248       this->~MPMCQueue();
 249       new (this) MPMCQueue(std::move(rhs));
 250     }
 251     return *this;
 252   }
 253
 254   ~MPMCQueue() {
 255     if (closed_ != nullptr) {
 256       for (int i = getNumClosed(this->dstate_.load()) - 1; i >= 0; --i) {
 257         delete[] closed_[i].slots_;
 258       }
 259       delete[] closed_;
 260     }
 261   }
 262
 263   size_t allocatedCapacity() const noexcept {
 264     return this->dcapacity_.load(std::memory_order_relaxed);
 265   }
 266
 267   template <typename ...Args>
 268   void blockingWrite(Args&&... args) noexcept {
 269     uint64_t ticket = this->pushTicket_++;
 270     Slot* slots;
 271     size_t cap;
 272     int stride;
 273     uint64_t state;
 274     uint64_t offset;
 275     do {
 276       if (!trySeqlockReadSection(state, slots, cap, stride)) {
 277         asm_volatile_pause();
 278         continue;
 279       }
 280       if (maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride)) {
 281         // There was an expansion after this ticket was issued.
 282         break;
 283       }
 284       if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
 285               this->turn(ticket - offset, cap))) {
 286         // A slot is ready. No need to expand.
 287         break;
 288       } else if (
 289           this->popTicket_.load(std::memory_order_relaxed) + cap > ticket) {
 290         // May block, but a pop is in progress. No need to expand.
 291         // Get seqlock read section info again in case an expansion
 292         // occurred with an equal or higher ticket.
 293         continue;
 294       } else {
 295         // May block. See if we can expand.
 296         if (tryExpand(state, cap)) {
 297           // This or another thread started an expansion. Get updated info.
 298           continue;
 299         } else {
 300           // Can't expand.
 301           break;
 302         }
 303       }
 304     } while (true);
 305     this->enqueueWithTicketBase(ticket-offset, slots, cap, stride,
 306                                 std::forward<Args>(args)...);
 307   }
 308
 309   void blockingReadWithTicket(uint64_t& ticket, T& elem) noexcept {
 310     ticket = this->popTicket_++;
 311     Slot* slots;
 312     size_t cap;
 313     int stride;
 314     uint64_t state;
 315     uint64_t offset;
 316     while (!trySeqlockReadSection(state, slots, cap, stride)) {
 317       asm_volatile_pause();
 318     }
 319     // If there was an expansion after the corresponding push ticket
 320     // was issued, adjust accordingly
 321     maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
 322     this->dequeueWithTicketBase(ticket-offset, slots, cap, stride, elem);
 323   }
 324
 325  private:
 326   enum {
 327     kSeqlockBits = 6,
 328     kDefaultMinDynamicCapacity = 10,
 329     kDefaultExpansionMultiplier = 10,
 330   };
 331
 332   size_t dmult_;
 333
 334   //  Info about closed slots arrays for use by lagging operations
 335   ClosedArray* closed_;
 336
 337   void initQueue(const size_t cap, const size_t mult) {
 338     this->stride_ = this->computeStride(cap);
 339     this->slots_ = new Slot[cap + 2 * this->kSlotPadding];
 340     this->dstate_.store(0);
 341     this->dcapacity_.store(cap);
 342     dmult_ = mult;
 343     size_t maxClosed = 0;
 344     for (size_t expanded = cap;
 345          expanded < this->capacity_;
 346          expanded *= mult) {
 347       ++maxClosed;
 348     }
 349     closed_ = (maxClosed > 0) ? new ClosedArray[maxClosed] : nullptr;
 350   }
 351
 352   bool tryObtainReadyPushTicket(
 353       uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
 354   ) noexcept {
 355     uint64_t state;
 356     do {
 357       ticket = this->pushTicket_.load(std::memory_order_acquire); // A
 358       if (!trySeqlockReadSection(state, slots, cap, stride)) {
 359         asm_volatile_pause();
 360         continue;
 361       }
 362
 363       // If there was an expansion with offset greater than this ticket,
 364       // adjust accordingly
 365       uint64_t offset;
 366       maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
 367
 368       if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
 369               this->turn(ticket - offset, cap))) {
 370         // A slot is ready.
 371         if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
 372           // Adjust ticket
 373           ticket -= offset;
 374           return true;
 375         } else {
 376           continue;
 377         }
 378       } else {
 379         if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B
 380           // Try again. Ticket changed.
 381           continue;
 382         }
 383         // Likely to block.
 384         // Try to expand unless the ticket is for a closed array
 385         if (offset == getOffset(state)) {
 386           if (tryExpand(state, cap)) {
 387             // This or another thread started an expansion. Get up-to-date info.
 388             continue;
 389           }
 390         }
 391         return false;
 392       }
 393     } while (true);
 394   }
 395
 396   bool tryObtainPromisedPushTicket(
 397     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
 398   ) noexcept {
 399     uint64_t state;
 400     do {
 401       ticket = this->pushTicket_.load(std::memory_order_acquire);
 402       auto numPops = this->popTicket_.load(std::memory_order_acquire);
 403       if (!trySeqlockReadSection(state, slots, cap, stride)) {
 404         asm_volatile_pause();
 405         continue;
 406       }
 407
 408       const auto curCap = cap;
 409       // If there was an expansion with offset greater than this ticket,
 410       // adjust accordingly
 411       uint64_t offset;
 412       maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
 413
 414       int64_t n = ticket - numPops;
 415
 416       if (n >= static_cast<ssize_t>(cap)) {
 417         if ((cap == curCap) && tryExpand(state, cap)) {
 418           // This or another thread started an expansion. Start over.
 419           continue;
 420         }
 421         // Can't expand.
 422         ticket -= offset;
 423         return false;
 424       }
 425
 426       if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
 427         // Adjust ticket
 428         ticket -= offset;
 429         return true;
 430       }
 431     } while (true);
 432   }
 433
 434   bool tryObtainReadyPopTicket(
 435     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
 436   ) noexcept {
 437     uint64_t state;
 438     do {
 439       ticket = this->popTicket_.load(std::memory_order_relaxed);
 440       if (!trySeqlockReadSection(state, slots, cap, stride)) {
 441         asm_volatile_pause();
 442         continue;
 443       }
 444
 445       // If there was an expansion after the corresponding push ticket
 446       // was issued, adjust accordingly
 447       uint64_t offset;
 448       maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
 449
 450       if (slots[this->idx((ticket - offset), cap, stride)].mayDequeue(
 451               this->turn(ticket - offset, cap))) {
 452         if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
 453           // Adjust ticket
 454           ticket -= offset;
 455           return true;
 456         }
 457       } else {
 458         return false;
 459       }
 460     } while (true);
 461   }
 462
 463   bool tryObtainPromisedPopTicket(
 464     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
 465   ) noexcept {
 466     uint64_t state;
 467     do {
 468       ticket = this->popTicket_.load(std::memory_order_acquire);
 469       auto numPushes = this->pushTicket_.load(std::memory_order_acquire);
 470       if (!trySeqlockReadSection(state, slots, cap, stride)) {
 471         asm_volatile_pause();
 472         continue;
 473       }
 474
 475       uint64_t offset;
 476       // If there was an expansion after the corresponding push
 477       // ticket was issued, adjust accordingly
 478       maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
 479
 480       if (ticket >= numPushes) {
 481         ticket -= offset;
 482         return false;
 483       }
 484       if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
 485         ticket -= offset;
 486         return true;
 487       }
 488     } while (true);
 489   }
 490
 491   /// Enqueues an element with a specific ticket number
 492   template <typename ...Args>
 493   void enqueueWithTicket(const uint64_t ticket, Args&&... args) noexcept {
 494     Slot* slots;
 495     size_t cap;
 496     int stride;
 497     uint64_t state;
 498     uint64_t offset;
 499
 500     while (!trySeqlockReadSection(state, slots, cap, stride)) {
 501     }
 502
 503     // If there was an expansion after this ticket was issued, adjust
 504     // accordingly
 505     maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
 506
 507     this->enqueueWithTicketBase(ticket-offset, slots, cap, stride,
 508                                 std::forward<Args>(args)...);
 509   }
 510
 511   uint64_t getOffset(const uint64_t state) const noexcept {
 512     return state >> kSeqlockBits;
 513   }
 514
 515   int getNumClosed(const uint64_t state) const noexcept {
 516     return (state & ((1 << kSeqlockBits) - 1)) >> 1;
 517   }
 518
 519   /// Try to expand the queue. Returns true if this expansion was
 520   /// successful or a concurent expansion is in progress. Returns
 521   /// false if the queue has reached its maximum capacity or
 522   /// allocation has failed.
 523   bool tryExpand(const uint64_t state, const size_t cap) noexcept {
 524     if (cap == this->capacity_) {
 525       return false;
 526     }
 527     // Acquire seqlock
 528     uint64_t oldval = state;
 529     assert((state & 1) == 0);
 530     if (this->dstate_.compare_exchange_strong(oldval, state + 1)) {
 531       assert(cap == this->dcapacity_.load());
 532       uint64_t ticket =
 533           1 + std::max(this->pushTicket_.load(), this->popTicket_.load());
 534       size_t newCapacity = std::min(dmult_ * cap, this->capacity_);
 535       Slot* newSlots =
 536           new (std::nothrow) Slot[newCapacity + 2 * this->kSlotPadding];
 537       if (newSlots == nullptr) {
 538         // Expansion failed. Restore the seqlock
 539         this->dstate_.store(state);
 540         return false;
 541       }
 542       // Successful expansion
 543       // calculate the current ticket offset
 544       uint64_t offset = getOffset(state);
 545       // calculate index in closed array
 546       int index = getNumClosed(state);
 547       assert((index << 1) < (1 << kSeqlockBits));
 548       // fill the info for the closed slots array
 549       closed_[index].offset_ = offset;
 550       closed_[index].slots_ = this->dslots_.load();
 551       closed_[index].capacity_ = cap;
 552       closed_[index].stride_ = this->dstride_.load();
 553       // update the new slots array info
 554       this->dslots_.store(newSlots);
 555       this->dcapacity_.store(newCapacity);
 556       this->dstride_.store(this->computeStride(newCapacity));
 557       // Release the seqlock and record the new ticket offset
 558       this->dstate_.store((ticket << kSeqlockBits) + (2 * (index + 1)));
 559       return true;
 560     } else { // failed to acquire seqlock
 561       // Someone acaquired the seqlock. Go back to the caller and get
 562       // up-to-date info.
 563       return true;
 564     }
 565   }
 566
 567   /// Seqlock read-only section
 568   bool trySeqlockReadSection(
 569     uint64_t& state, Slot*& slots, size_t& cap, int& stride
 570   ) noexcept {
 571     state = this->dstate_.load(std::memory_order_acquire);
 572     if (state & 1) {
 573       // Locked.
 574       return false;
 575     }
 576     // Start read-only section.
 577     slots = this->dslots_.load(std::memory_order_relaxed);
 578     cap = this->dcapacity_.load(std::memory_order_relaxed);
 579     stride = this->dstride_.load(std::memory_order_relaxed);
 580     // End of read-only section. Validate seqlock.
 581     std::atomic_thread_fence(std::memory_order_acquire);
 582     return (state == this->dstate_.load(std::memory_order_relaxed));
 583   }
 584
 585   /// If there was an expansion after ticket was issued, update local variables
 586   /// of the lagging operation using the most recent closed array with
 587   /// offset <= ticket and return true. Otherwise, return false;
 588   bool maybeUpdateFromClosed(
 589       const uint64_t state,
 590       const uint64_t ticket,
 591       uint64_t& offset,
 592       Slot*& slots,
 593       size_t& cap,
 594       int& stride) noexcept {
 595     offset = getOffset(state);
 596     if (ticket >= offset) {
 597       return false;
 598     }
 599     for (int i = getNumClosed(state) - 1; i >= 0; --i) {
 600       offset = closed_[i].offset_;
 601       if (offset <= ticket) {
 602         slots = closed_[i].slots_;
 603         cap = closed_[i].capacity_;
 604         stride = closed_[i].stride_;
 605         return true;
 606       }
 607     }
 608     // A closed array with offset <= ticket should have been found
 609     assert(false);
 610     return false;
 611   }
 612 };
 613
 614 namespace detail {
 615
 616 /// CRTP specialization of MPMCQueueBase
 617 template <
 618     template <typename T, template <typename> class Atom, bool Dynamic>
 619     class Derived,
 620     typename T,
 621     template <typename> class Atom,
 622     bool Dynamic>
 623 class MPMCQueueBase<Derived<T, Atom, Dynamic>> : boost::noncopyable {
 624
 625 // Note: Using CRTP static casts in several functions of this base
 626 // template instead of making called functions virtual or duplicating
 627 // the code of calling functions in the derived partially specialized
 628 // template
 629
 630   static_assert(std::is_nothrow_constructible<T,T&&>::value ||
 631                 folly::IsRelocatable<T>::value,
 632       "T must be relocatable or have a noexcept move constructor");
 633
 634  public:
 635   typedef T value_type;
 636
 637   using Slot = detail::SingleElementQueue<T,Atom>;
 638
 639   explicit MPMCQueueBase(size_t queueCapacity)
 640     : capacity_(queueCapacity)
 641     , pushTicket_(0)
 642     , popTicket_(0)
 643     , pushSpinCutoff_(0)
 644     , popSpinCutoff_(0)
 645   {
 646     if (queueCapacity == 0) {
 647       throw std::invalid_argument(
 648         "MPMCQueue with explicit capacity 0 is impossible"
 649         // Stride computation in derived classes would sigfpe if capacity is 0
 650       );
 651     }
 652
 653     // ideally this would be a static assert, but g++ doesn't allow it
 654     assert(
 655         alignof(MPMCQueue<T, Atom>) >= hardware_destructive_interference_size);
 656     assert(
 657         static_cast<uint8_t*>(static_cast<void*>(&popTicket_)) -
 658             static_cast<uint8_t*>(static_cast<void*>(&pushTicket_)) >=
 659         static_cast<ptrdiff_t>(hardware_destructive_interference_size));
 660   }
 661
 662   /// A default-constructed queue is useful because a usable (non-zero
 663   /// capacity) queue can be moved onto it or swapped with it
 664   MPMCQueueBase() noexcept
 665     : capacity_(0)
 666     , slots_(nullptr)
 667     , stride_(0)
 668     , dstate_(0)
 669     , dcapacity_(0)
 670     , pushTicket_(0)
 671     , popTicket_(0)
 672     , pushSpinCutoff_(0)
 673     , popSpinCutoff_(0)
 674   {}
 675
 676   /// IMPORTANT: The move constructor is here to make it easier to perform
 677   /// the initialization phase, it is not safe to use when there are any
 678   /// concurrent accesses (this is not checked).
 679   MPMCQueueBase(MPMCQueueBase<Derived<T,Atom,Dynamic>>&& rhs) noexcept
 680     : capacity_(rhs.capacity_)
 681     , slots_(rhs.slots_)
 682     , stride_(rhs.stride_)
 683     , dstate_(rhs.dstate_.load(std::memory_order_relaxed))
 684     , dcapacity_(rhs.dcapacity_.load(std::memory_order_relaxed))
 685     , pushTicket_(rhs.pushTicket_.load(std::memory_order_relaxed))
 686     , popTicket_(rhs.popTicket_.load(std::memory_order_relaxed))
 687     , pushSpinCutoff_(rhs.pushSpinCutoff_.load(std::memory_order_relaxed))
 688     , popSpinCutoff_(rhs.popSpinCutoff_.load(std::memory_order_relaxed))
 689   {
 690     // relaxed ops are okay for the previous reads, since rhs queue can't
 691     // be in concurrent use
 692
 693     // zero out rhs
 694     rhs.capacity_ = 0;
 695     rhs.slots_ = nullptr;
 696     rhs.stride_ = 0;
 697     rhs.dstate_.store(0, std::memory_order_relaxed);
 698     rhs.dcapacity_.store(0, std::memory_order_relaxed);
 699     rhs.pushTicket_.store(0, std::memory_order_relaxed);
 700     rhs.popTicket_.store(0, std::memory_order_relaxed);
 701     rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
 702     rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
 703   }
 704
 705   /// IMPORTANT: The move operator is here to make it easier to perform
 706   /// the initialization phase, it is not safe to use when there are any
 707   /// concurrent accesses (this is not checked).
 708   MPMCQueueBase<Derived<T,Atom,Dynamic>> const& operator=
 709     (MPMCQueueBase<Derived<T,Atom,Dynamic>>&& rhs) {
 710     if (this != &rhs) {
 711       this->~MPMCQueueBase();
 712       new (this) MPMCQueueBase(std::move(rhs));
 713     }
 714     return *this;
 715   }
 716
 717   /// MPMCQueue can only be safely destroyed when there are no
 718   /// pending enqueuers or dequeuers (this is not checked).
 719   ~MPMCQueueBase() {
 720     delete[] slots_;
 721   }
 722
 723   /// Returns the number of writes (including threads that are blocked waiting
 724   /// to write) minus the number of reads (including threads that are blocked
 725   /// waiting to read). So effectively, it becomes:
 726   /// elements in queue + pending(calls to write) - pending(calls to read).
 727   /// If nothing is pending, then the method returns the actual number of
 728   /// elements in the queue.
 729   /// The returned value can be negative if there are no writers and the queue
 730   /// is empty, but there is one reader that is blocked waiting to read (in
 731   /// which case, the returned size will be -1).
 732   ssize_t size() const noexcept {
 733     // since both pushes and pops increase monotonically, we can get a
 734     // consistent snapshot either by bracketing a read of popTicket_ with
 735     // two reads of pushTicket_ that return the same value, or the other
 736     // way around.  We maximize our chances by alternately attempting
 737     // both bracketings.
 738     uint64_t pushes = pushTicket_.load(std::memory_order_acquire); // A
 739     uint64_t pops = popTicket_.load(std::memory_order_acquire); // B
 740     while (true) {
 741       uint64_t nextPushes = pushTicket_.load(std::memory_order_acquire); // C
 742       if (pushes == nextPushes) {
 743         // pushTicket_ didn't change from A (or the previous C) to C,
 744         // so we can linearize at B (or D)
 745         return ssize_t(pushes - pops);
 746       }
 747       pushes = nextPushes;
 748       uint64_t nextPops = popTicket_.load(std::memory_order_acquire); // D
 749       if (pops == nextPops) {
 750         // popTicket_ didn't chance from B (or the previous D), so we
 751         // can linearize at C
 752         return ssize_t(pushes - pops);
 753       }
 754       pops = nextPops;
 755     }
 756   }
 757
 758   /// Returns true if there are no items available for dequeue
 759   bool isEmpty() const noexcept {
 760     return size() <= 0;
 761   }
 762
 763   /// Returns true if there is currently no empty space to enqueue
 764   bool isFull() const noexcept {
 765     // careful with signed -> unsigned promotion, since size can be negative
 766     return size() >= static_cast<ssize_t>(capacity_);
 767   }
 768
 769   /// Returns is a guess at size() for contexts that don't need a precise
 770   /// value, such as stats. More specifically, it returns the number of writes
 771   /// minus the number of reads, but after reading the number of writes, more
 772   /// writers could have came before the number of reads was sampled,
 773   /// and this method doesn't protect against such case.
 774   /// The returned value can be negative.
 775   ssize_t sizeGuess() const noexcept {
 776     return writeCount() - readCount();
 777   }
 778
 779   /// Doesn't change
 780   size_t capacity() const noexcept {
 781     return capacity_;
 782   }
 783
 784   /// Doesn't change for non-dynamic
 785   size_t allocatedCapacity() const noexcept {
 786     return capacity_;
 787   }
 788
 789   /// Returns the total number of calls to blockingWrite or successful
 790   /// calls to write, including those blockingWrite calls that are
 791   /// currently blocking
 792   uint64_t writeCount() const noexcept {
 793     return pushTicket_.load(std::memory_order_acquire);
 794   }
 795
 796   /// Returns the total number of calls to blockingRead or successful
 797   /// calls to read, including those blockingRead calls that are currently
 798   /// blocking
 799   uint64_t readCount() const noexcept {
 800     return popTicket_.load(std::memory_order_acquire);
 801   }
 802
 803   /// Enqueues a T constructed from args, blocking until space is
 804   /// available.  Note that this method signature allows enqueue via
 805   /// move, if args is a T rvalue, via copy, if args is a T lvalue, or
 806   /// via emplacement if args is an initializer list that can be passed
 807   /// to a T constructor.
 808   template <typename ...Args>
 809   void blockingWrite(Args&&... args) noexcept {
 810     enqueueWithTicketBase(pushTicket_++, slots_, capacity_, stride_,
 811                           std::forward<Args>(args)...);
 812   }
 813
 814   /// If an item can be enqueued with no blocking, does so and returns
 815   /// true, otherwise returns false.  This method is similar to
 816   /// writeIfNotFull, but if you don't have a specific need for that
 817   /// method you should use this one.
 818   ///
 819   /// One of the common usages of this method is to enqueue via the
 820   /// move constructor, something like q.write(std::move(x)).  If write
 821   /// returns false because the queue is full then x has not actually been
 822   /// consumed, which looks strange.  To understand why it is actually okay
 823   /// to use x afterward, remember that std::move is just a typecast that
 824   /// provides an rvalue reference that enables use of a move constructor
 825   /// or operator.  std::move doesn't actually move anything.  It could
 826   /// more accurately be called std::rvalue_cast or std::move_permission.
 827   template <typename ...Args>
 828   bool write(Args&&... args) noexcept {
 829     uint64_t ticket;
 830     Slot* slots;
 831     size_t cap;
 832     int stride;
 833     if (static_cast<Derived<T,Atom,Dynamic>*>(this)->
 834         tryObtainReadyPushTicket(ticket, slots, cap, stride)) {
 835       // we have pre-validated that the ticket won't block
 836       enqueueWithTicketBase(ticket, slots, cap, stride,
 837                             std::forward<Args>(args)...);
 838       return true;
 839     } else {
 840       return false;
 841     }
 842   }
 843
 844   template <class Clock, typename... Args>
 845   bool tryWriteUntil(const std::chrono::time_point<Clock>& when,
 846                      Args&&... args) noexcept {
 847     uint64_t ticket;
 848     Slot* slots;
 849     size_t cap;
 850     int stride;
 851     if (tryObtainPromisedPushTicketUntil(ticket, slots, cap, stride, when)) {
 852         // we have pre-validated that the ticket won't block, or rather that
 853         // it won't block longer than it takes another thread to dequeue an
 854         // element from the slot it identifies.
 855       enqueueWithTicketBase(ticket, slots, cap, stride,
 856                             std::forward<Args>(args)...);
 857       return true;
 858     } else {
 859       return false;
 860     }
 861   }
 862
 863   /// If the queue is not full, enqueues and returns true, otherwise
 864   /// returns false.  Unlike write this method can be blocked by another
 865   /// thread, specifically a read that has linearized (been assigned
 866   /// a ticket) but not yet completed.  If you don't really need this
 867   /// function you should probably use write.
 868   ///
 869   /// MPMCQueue isn't lock-free, so just because a read operation has
 870   /// linearized (and isFull is false) doesn't mean that space has been
 871   /// made available for another write.  In this situation write will
 872   /// return false, but writeIfNotFull will wait for the dequeue to finish.
 873   /// This method is required if you are composing queues and managing
 874   /// your own wakeup, because it guarantees that after every successful
 875   /// write a readIfNotEmpty will succeed.
 876   template <typename ...Args>
 877   bool writeIfNotFull(Args&&... args) noexcept {
 878     uint64_t ticket;
 879     Slot* slots;
 880     size_t cap;
 881     int stride;
 882     if (static_cast<Derived<T,Atom,Dynamic>*>(this)->
 883         tryObtainPromisedPushTicket(ticket, slots, cap, stride)) {
 884       // some other thread is already dequeuing the slot into which we
 885       // are going to enqueue, but we might have to wait for them to finish
 886       enqueueWithTicketBase(ticket, slots, cap, stride,
 887                             std::forward<Args>(args)...);
 888       return true;
 889     } else {
 890       return false;
 891     }
 892   }
 893
 894   /// Moves a dequeued element onto elem, blocking until an element
 895   /// is available
 896   void blockingRead(T& elem) noexcept {
 897     uint64_t ticket;
 898     static_cast<Derived<T,Atom,Dynamic>*>(this)->
 899       blockingReadWithTicket(ticket, elem);
 900   }
 901
 902   /// Same as blockingRead() but also records the ticket nunmer
 903   void blockingReadWithTicket(uint64_t& ticket, T& elem) noexcept {
 904     assert(capacity_ != 0);
 905     ticket = popTicket_++;
 906     dequeueWithTicketBase(ticket, slots_, capacity_, stride_, elem);
 907   }
 908
 909   /// If an item can be dequeued with no blocking, does so and returns
 910   /// true, otherwise returns false.
 911   bool read(T& elem) noexcept {
 912     uint64_t ticket;
 913     return readAndGetTicket(ticket, elem);
 914   }
 915
 916   /// Same as read() but also records the ticket nunmer
 917   bool readAndGetTicket(uint64_t& ticket, T& elem) noexcept {
 918     Slot* slots;
 919     size_t cap;
 920     int stride;
 921     if (static_cast<Derived<T,Atom,Dynamic>*>(this)->
 922         tryObtainReadyPopTicket(ticket, slots, cap, stride)) {
 923       // the ticket has been pre-validated to not block
 924       dequeueWithTicketBase(ticket, slots, cap, stride, elem);
 925       return true;
 926     } else {
 927       return false;
 928     }
 929   }
 930
 931   template <class Clock, typename... Args>
 932   bool tryReadUntil(
 933       const std::chrono::time_point<Clock>& when,
 934       T& elem) noexcept {
 935     uint64_t ticket;
 936     Slot* slots;
 937     size_t cap;
 938     int stride;
 939     if (tryObtainPromisedPopTicketUntil(ticket, slots, cap, stride, when)) {
 940       // we have pre-validated that the ticket won't block, or rather that
 941       // it won't block longer than it takes another thread to enqueue an
 942       // element on the slot it identifies.
 943       dequeueWithTicketBase(ticket, slots, cap, stride, elem);
 944       return true;
 945     } else {
 946       return false;
 947     }
 948   }
 949
 950   /// If the queue is not empty, dequeues and returns true, otherwise
 951   /// returns false.  If the matching write is still in progress then this
 952   /// method may block waiting for it.  If you don't rely on being able
 953   /// to dequeue (such as by counting completed write) then you should
 954   /// prefer read.
 955   bool readIfNotEmpty(T& elem) noexcept {
 956     uint64_t ticket;
 957     Slot* slots;
 958     size_t cap;
 959     int stride;
 960     if (static_cast<Derived<T,Atom,Dynamic>*>(this)->
 961         tryObtainPromisedPopTicket(ticket, slots, cap, stride)) {
 962       // the matching enqueue already has a ticket, but might not be done
 963       dequeueWithTicketBase(ticket, slots, cap, stride, elem);
 964       return true;
 965     } else {
 966       return false;
 967     }
 968   }
 969
 970  protected:
 971   enum {
 972     /// Once every kAdaptationFreq we will spin longer, to try to estimate
 973     /// the proper spin backoff
 974     kAdaptationFreq = 128,
 975
 976     /// To avoid false sharing in slots_ with neighboring memory
 977     /// allocations, we pad it with this many SingleElementQueue-s at
 978     /// each end
 979     kSlotPadding =
 980         (hardware_destructive_interference_size - 1) / sizeof(Slot) + 1
 981   };
 982
 983   /// The maximum number of items in the queue at once
 984   size_t FOLLY_ALIGN_TO_AVOID_FALSE_SHARING capacity_;
 985
 986   /// Anonymous union for use when Dynamic = false and true, respectively
 987   union {
 988     /// An array of capacity_ SingleElementQueue-s, each of which holds
 989     /// either 0 or 1 item.  We over-allocate by 2 * kSlotPadding and don't
 990     /// touch the slots at either end, to avoid false sharing
 991     Slot* slots_;
 992     /// Current dynamic slots array of dcapacity_ SingleElementQueue-s
 993     Atom<Slot*> dslots_;
 994   };
 995
 996   /// Anonymous union for use when Dynamic = false and true, respectively
 997   union {
 998     /// The number of slots_ indices that we advance for each ticket, to
 999     /// avoid false sharing.  Ideally slots_[i] and slots_[i + stride_]
1000     /// aren't on the same cache line
1001     int stride_;
1002     /// Current stride
1003     Atom<int> dstride_;
1004   };
1005
1006   /// The following two memebers are used by dynamic MPMCQueue.
1007   /// Ideally they should be in MPMCQueue<T,Atom,true>, but we get
1008   /// better cache locality if they are in the same cache line as
1009   /// dslots_ and dstride_.
1010   ///
1011   /// Dynamic state. A packed seqlock and ticket offset
1012   Atom<uint64_t> dstate_;
1013   /// Dynamic capacity
1014   Atom<size_t> dcapacity_;
1015
1016   /// Enqueuers get tickets from here
1017   Atom<uint64_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING pushTicket_;
1018
1019   /// Dequeuers get tickets from here
1020   Atom<uint64_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING popTicket_;
1021
1022   /// This is how many times we will spin before using FUTEX_WAIT when
1023   /// the queue is full on enqueue, adaptively computed by occasionally
1024   /// spinning for longer and smoothing with an exponential moving average
1025   Atom<uint32_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING pushSpinCutoff_;
1026
1027   /// The adaptive spin cutoff when the queue is empty on dequeue
1028   Atom<uint32_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING popSpinCutoff_;
1029
1030   /// Alignment doesn't prevent false sharing at the end of the struct,
1031   /// so fill out the last cache line
1032   char pad_[hardware_destructive_interference_size - sizeof(Atom<uint32_t>)];
1033
1034   /// We assign tickets in increasing order, but we don't want to
1035   /// access neighboring elements of slots_ because that will lead to
1036   /// false sharing (multiple cores accessing the same cache line even
1037   /// though they aren't accessing the same bytes in that cache line).
1038   /// To avoid this we advance by stride slots per ticket.
1039   ///
1040   /// We need gcd(capacity, stride) to be 1 so that we will use all
1041   /// of the slots.  We ensure this by only considering prime strides,
1042   /// which either have no common divisors with capacity or else have
1043   /// a zero remainder after dividing by capacity.  That is sufficient
1044   /// to guarantee correctness, but we also want to actually spread the
1045   /// accesses away from each other to avoid false sharing (consider a
1046   /// stride of 7 with a capacity of 8).  To that end we try a few taking
1047   /// care to observe that advancing by -1 is as bad as advancing by 1
1048   /// when in comes to false sharing.
1049   ///
1050   /// The simple way to avoid false sharing would be to pad each
1051   /// SingleElementQueue, but since we have capacity_ of them that could
1052   /// waste a lot of space.
1053   static int computeStride(size_t capacity) noexcept {
1054     static const int smallPrimes[] = { 2, 3, 5, 7, 11, 13, 17, 19, 23 };
1055
1056     int bestStride = 1;
1057     size_t bestSep = 1;
1058     for (int stride : smallPrimes) {
1059       if ((stride % capacity) == 0 || (capacity % stride) == 0) {
1060         continue;
1061       }
1062       size_t sep = stride % capacity;
1063       sep = std::min(sep, capacity - sep);
1064       if (sep > bestSep) {
1065         bestStride = stride;
1066         bestSep = sep;
1067       }
1068     }
1069     return bestStride;
1070   }
1071
1072   /// Returns the index into slots_ that should be used when enqueuing or
1073   /// dequeuing with the specified ticket
1074   size_t idx(uint64_t ticket, size_t cap, int stride) noexcept {
1075     return ((ticket * stride) % cap) + kSlotPadding;
1076   }
1077
1078   /// Maps an enqueue or dequeue ticket to the turn should be used at the
1079   /// corresponding SingleElementQueue
1080   uint32_t turn(uint64_t ticket, size_t cap) noexcept {
1081     assert(cap != 0);
1082     return uint32_t(ticket / cap);
1083   }
1084
1085   /// Tries to obtain a push ticket for which SingleElementQueue::enqueue
1086   /// won't block.  Returns true on immediate success, false on immediate
1087   /// failure.
1088   bool tryObtainReadyPushTicket(
1089     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
1090   ) noexcept {
1091     ticket = pushTicket_.load(std::memory_order_acquire); // A
1092     slots = slots_;
1093     cap = capacity_;
1094     stride = stride_;
1095     while (true) {
1096       if (!slots[idx(ticket, cap, stride)]
1097           .mayEnqueue(turn(ticket, cap))) {
1098         // if we call enqueue(ticket, ...) on the SingleElementQueue
1099         // right now it would block, but this might no longer be the next
1100         // ticket.  We can increase the chance of tryEnqueue success under
1101         // contention (without blocking) by rechecking the ticket dispenser
1102         auto prev = ticket;
1103         ticket = pushTicket_.load(std::memory_order_acquire); // B
1104         if (prev == ticket) {
1105           // mayEnqueue was bracketed by two reads (A or prev B or prev
1106           // failing CAS to B), so we are definitely unable to enqueue
1107           return false;
1108         }
1109       } else {
1110         // we will bracket the mayEnqueue check with a read (A or prev B
1111         // or prev failing CAS) and the following CAS.  If the CAS fails
1112         // it will effect a load of pushTicket_
1113         if (pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
1114           return true;
1115         }
1116       }
1117     }
1118   }
1119
1120   /// Tries until when to obtain a push ticket for which
1121   /// SingleElementQueue::enqueue  won't block.  Returns true on success, false
1122   /// on failure.
1123   /// ticket is filled on success AND failure.
1124   template <class Clock>
1125   bool tryObtainPromisedPushTicketUntil(
1126     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride,
1127     const std::chrono::time_point<Clock>& when
1128   ) noexcept {
1129     bool deadlineReached = false;
1130     while (!deadlineReached) {
1131       if (static_cast<Derived<T,Atom,Dynamic>*>(this)->
1132           tryObtainPromisedPushTicket(ticket, slots, cap, stride)) {
1133         return true;
1134       }
1135       // ticket is a blocking ticket until the preceding ticket has been
1136       // processed: wait until this ticket's turn arrives. We have not reserved
1137       // this ticket so we will have to re-attempt to get a non-blocking ticket
1138       // if we wake up before we time-out.
1139       deadlineReached = !slots[idx(ticket, cap, stride)]
1140         .tryWaitForEnqueueTurnUntil(turn(ticket, cap), pushSpinCutoff_,
1141                                     (ticket % kAdaptationFreq) == 0, when);
1142     }
1143     return false;
1144   }
1145
1146   /// Tries to obtain a push ticket which can be satisfied if all
1147   /// in-progress pops complete.  This function does not block, but
1148   /// blocking may be required when using the returned ticket if some
1149   /// other thread's pop is still in progress (ticket has been granted but
1150   /// pop has not yet completed).
1151   bool tryObtainPromisedPushTicket(
1152     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
1153   ) noexcept {
1154     auto numPushes = pushTicket_.load(std::memory_order_acquire); // A
1155     slots = slots_;
1156     cap = capacity_;
1157     stride = stride_;
1158     while (true) {
1159       ticket = numPushes;
1160       const auto numPops = popTicket_.load(std::memory_order_acquire); // B
1161       // n will be negative if pops are pending
1162       const int64_t n = int64_t(numPushes - numPops);
1163       if (n >= static_cast<ssize_t>(capacity_)) {
1164         // Full, linearize at B.  We don't need to recheck the read we
1165         // performed at A, because if numPushes was stale at B then the
1166         // real numPushes value is even worse
1167         return false;
1168       }
1169       if (pushTicket_.compare_exchange_strong(numPushes, numPushes + 1)) {
1170         return true;
1171       }
1172     }
1173   }
1174
1175   /// Tries to obtain a pop ticket for which SingleElementQueue::dequeue
1176   /// won't block.  Returns true on immediate success, false on immediate
1177   /// failure.
1178   bool tryObtainReadyPopTicket(
1179     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
1180   ) noexcept {
1181     ticket = popTicket_.load(std::memory_order_acquire);
1182     slots = slots_;
1183     cap = capacity_;
1184     stride = stride_;
1185     while (true) {
1186       if (!slots[idx(ticket, cap, stride)]
1187           .mayDequeue(turn(ticket, cap))) {
1188         auto prev = ticket;
1189         ticket = popTicket_.load(std::memory_order_acquire);
1190         if (prev == ticket) {
1191           return false;
1192         }
1193       } else {
1194         if (popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
1195           return true;
1196         }
1197       }
1198     }
1199   }
1200
1201   /// Tries until when to obtain a pop ticket for which
1202   /// SingleElementQueue::dequeue won't block.  Returns true on success, false
1203   /// on failure.
1204   /// ticket is filled on success AND failure.
1205   template <class Clock>
1206   bool tryObtainPromisedPopTicketUntil(
1207       uint64_t& ticket,
1208       Slot*& slots,
1209       size_t& cap,
1210       int& stride,
1211       const std::chrono::time_point<Clock>& when) noexcept {
1212     bool deadlineReached = false;
1213     while (!deadlineReached) {
1214       if (static_cast<Derived<T, Atom, Dynamic>*>(this)
1215               ->tryObtainPromisedPopTicket(ticket, slots, cap, stride)) {
1216         return true;
1217       }
1218       // ticket is a blocking ticket until the preceding ticket has been
1219       // processed: wait until this ticket's turn arrives. We have not reserved
1220       // this ticket so we will have to re-attempt to get a non-blocking ticket
1221       // if we wake up before we time-out.
1222       deadlineReached =
1223           !slots[idx(ticket, cap, stride)].tryWaitForDequeueTurnUntil(
1224               turn(ticket, cap),
1225               pushSpinCutoff_,
1226               (ticket % kAdaptationFreq) == 0,
1227               when);
1228     }
1229     return false;
1230   }
1231
1232   /// Similar to tryObtainReadyPopTicket, but returns a pop ticket whose
1233   /// corresponding push ticket has already been handed out, rather than
1234   /// returning one whose corresponding push ticket has already been
1235   /// completed.  This means that there is a possibility that the caller
1236   /// will block when using the ticket, but it allows the user to rely on
1237   /// the fact that if enqueue has succeeded, tryObtainPromisedPopTicket
1238   /// will return true.  The "try" part of this is that we won't have
1239   /// to block waiting for someone to call enqueue, although we might
1240   /// have to block waiting for them to finish executing code inside the
1241   /// MPMCQueue itself.
1242   bool tryObtainPromisedPopTicket(
1243     uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
1244   ) noexcept {
1245     auto numPops = popTicket_.load(std::memory_order_acquire); // A
1246     slots = slots_;
1247     cap = capacity_;
1248     stride = stride_;
1249     while (true) {
1250       ticket = numPops;
1251       const auto numPushes = pushTicket_.load(std::memory_order_acquire); // B
1252       if (numPops >= numPushes) {
1253         // Empty, or empty with pending pops.  Linearize at B.  We don't
1254         // need to recheck the read we performed at A, because if numPops
1255         // is stale then the fresh value is larger and the >= is still true
1256         return false;
1257       }
1258       if (popTicket_.compare_exchange_strong(numPops, numPops + 1)) {
1259         return true;
1260       }
1261     }
1262   }
1263
1264   // Given a ticket, constructs an enqueued item using args
1265   template <typename ...Args>
1266   void enqueueWithTicketBase(
1267     uint64_t ticket, Slot* slots, size_t cap, int stride, Args&&... args
1268   ) noexcept {
1269     slots[idx(ticket, cap, stride)]
1270       .enqueue(turn(ticket, cap),
1271                pushSpinCutoff_,
1272                (ticket % kAdaptationFreq) == 0,
1273                std::forward<Args>(args)...);
1274   }
1275
1276   // To support tracking ticket numbers in MPMCPipelineStageImpl
1277   template <typename ...Args>
1278   void enqueueWithTicket(uint64_t ticket, Args&&... args) noexcept {
1279     enqueueWithTicketBase(ticket, slots_, capacity_, stride_,
1280                           std::forward<Args>(args)...);
1281   }
1282
1283   // Given a ticket, dequeues the corresponding element
1284   void dequeueWithTicketBase(
1285     uint64_t ticket, Slot* slots, size_t cap, int stride, T& elem
1286   ) noexcept {
1287     assert(cap != 0);
1288     slots[idx(ticket, cap, stride)]
1289       .dequeue(turn(ticket, cap),
1290                popSpinCutoff_,
1291                (ticket % kAdaptationFreq) == 0,
1292                elem);
1293   }
1294 };
1295
1296 /// SingleElementQueue implements a blocking queue that holds at most one
1297 /// item, and that requires its users to assign incrementing identifiers
1298 /// (turns) to each enqueue and dequeue operation.  Note that the turns
1299 /// used by SingleElementQueue are doubled inside the TurnSequencer
1300 template <typename T, template <typename> class Atom>
1301 struct SingleElementQueue {
1302
1303   ~SingleElementQueue() noexcept {
1304     if ((sequencer_.uncompletedTurnLSB() & 1) == 1) {
1305       // we are pending a dequeue, so we have a constructed item
1306       destroyContents();
1307     }
1308   }
1309
1310   /// enqueue using in-place noexcept construction
1311   template <
1312       typename... Args,
1313       typename = typename std::enable_if<
1314           std::is_nothrow_constructible<T, Args...>::value>::type>
1315   void enqueue(const uint32_t turn,
1316                Atom<uint32_t>& spinCutoff,
1317                const bool updateSpinCutoff,
1318                Args&&... args) noexcept {
1319     sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
1320     new (&contents_) T(std::forward<Args>(args)...);
1321     sequencer_.completeTurn(turn * 2);
1322   }
1323
1324   /// enqueue using move construction, either real (if
1325   /// is_nothrow_move_constructible) or simulated using relocation and
1326   /// default construction (if IsRelocatable and is_nothrow_constructible)
1327   template <
1328       typename = typename std::enable_if<
1329           (folly::IsRelocatable<T>::value &&
1330            std::is_nothrow_constructible<T>::value) ||
1331           std::is_nothrow_constructible<T, T&&>::value>::type>
1332   void enqueue(
1333       const uint32_t turn,
1334       Atom<uint32_t>& spinCutoff,
1335       const bool updateSpinCutoff,
1336       T&& goner) noexcept {
1337     enqueueImpl(
1338         turn,
1339         spinCutoff,
1340         updateSpinCutoff,
1341         std::move(goner),
1342         typename std::conditional<std::is_nothrow_constructible<T,T&&>::value,
1343                                   ImplByMove, ImplByRelocation>::type());
1344   }
1345
1346   /// Waits until either:
1347   /// 1: the dequeue turn preceding the given enqueue turn has arrived
1348   /// 2: the given deadline has arrived
1349   /// Case 1 returns true, case 2 returns false.
1350   template <class Clock>
1351   bool tryWaitForEnqueueTurnUntil(
1352       const uint32_t turn,
1353       Atom<uint32_t>& spinCutoff,
1354       const bool updateSpinCutoff,
1355       const std::chrono::time_point<Clock>& when) noexcept {
1356     return sequencer_.tryWaitForTurn(
1357                turn * 2, spinCutoff, updateSpinCutoff, &when) !=
1358         TurnSequencer<Atom>::TryWaitResult::TIMEDOUT;
1359   }
1360
1361   bool mayEnqueue(const uint32_t turn) const noexcept {
1362     return sequencer_.isTurn(turn * 2);
1363   }
1364
1365   void dequeue(uint32_t turn,
1366                Atom<uint32_t>& spinCutoff,
1367                const bool updateSpinCutoff,
1368                T& elem) noexcept {
1369     dequeueImpl(turn,
1370                 spinCutoff,
1371                 updateSpinCutoff,
1372                 elem,
1373                 typename std::conditional<folly::IsRelocatable<T>::value,
1374                                           ImplByRelocation,
1375                                           ImplByMove>::type());
1376   }
1377
1378   /// Waits until either:
1379   /// 1: the enqueue turn preceding the given dequeue turn has arrived
1380   /// 2: the given deadline has arrived
1381   /// Case 1 returns true, case 2 returns false.
1382   template <class Clock>
1383   bool tryWaitForDequeueTurnUntil(
1384       const uint32_t turn,
1385       Atom<uint32_t>& spinCutoff,
1386       const bool updateSpinCutoff,
1387       const std::chrono::time_point<Clock>& when) noexcept {
1388     return sequencer_.tryWaitForTurn(
1389                turn * 2 + 1, spinCutoff, updateSpinCutoff, &when) !=
1390         TurnSequencer<Atom>::TryWaitResult::TIMEDOUT;
1391   }
1392
1393   bool mayDequeue(const uint32_t turn) const noexcept {
1394     return sequencer_.isTurn(turn * 2 + 1);
1395   }
1396
1397  private:
1398   /// Storage for a T constructed with placement new
1399   typename std::aligned_storage<sizeof(T),alignof(T)>::type contents_;
1400
1401   /// Even turns are pushes, odd turns are pops
1402   TurnSequencer<Atom> sequencer_;
1403
1404   T* ptr() noexcept {
1405     return static_cast<T*>(static_cast<void*>(&contents_));
1406   }
1407
1408   void destroyContents() noexcept {
1409     try {
1410       ptr()->~T();
1411     } catch (...) {
1412       // g++ doesn't seem to have std::is_nothrow_destructible yet
1413     }
1414 #ifndef NDEBUG
1415     memset(&contents_, 'Q', sizeof(T));
1416 #endif
1417   }
1418
1419   /// Tag classes for dispatching to enqueue/dequeue implementation.
1420   struct ImplByRelocation {};
1421   struct ImplByMove {};
1422
1423   /// enqueue using nothrow move construction.
1424   void enqueueImpl(const uint32_t turn,
1425                    Atom<uint32_t>& spinCutoff,
1426                    const bool updateSpinCutoff,
1427                    T&& goner,
1428                    ImplByMove) noexcept {
1429     sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
1430     new (&contents_) T(std::move(goner));
1431     sequencer_.completeTurn(turn * 2);
1432   }
1433
1434   /// enqueue by simulating nothrow move with relocation, followed by
1435   /// default construction to a noexcept relocation.
1436   void enqueueImpl(const uint32_t turn,
1437                    Atom<uint32_t>& spinCutoff,
1438                    const bool updateSpinCutoff,
1439                    T&& goner,
1440                    ImplByRelocation) noexcept {
1441     sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
1442     memcpy(&contents_, &goner, sizeof(T));
1443     sequencer_.completeTurn(turn * 2);
1444     new (&goner) T();
1445   }
1446
1447   /// dequeue by destructing followed by relocation.  This version is preferred,
1448   /// because as much work as possible can be done before waiting.
1449   void dequeueImpl(uint32_t turn,
1450                    Atom<uint32_t>& spinCutoff,
1451                    const bool updateSpinCutoff,
1452                    T& elem,
1453                    ImplByRelocation) noexcept {
1454     try {
1455       elem.~T();
1456     } catch (...) {
1457       // unlikely, but if we don't complete our turn the queue will die
1458     }
1459     sequencer_.waitForTurn(turn * 2 + 1, spinCutoff, updateSpinCutoff);
1460     memcpy(&elem, &contents_, sizeof(T));
1461     sequencer_.completeTurn(turn * 2 + 1);
1462   }
1463
1464   /// dequeue by nothrow move assignment.
1465   void dequeueImpl(uint32_t turn,
1466                    Atom<uint32_t>& spinCutoff,
1467                    const bool updateSpinCutoff,
1468                    T& elem,
1469                    ImplByMove) noexcept {
1470     sequencer_.waitForTurn(turn * 2 + 1, spinCutoff, updateSpinCutoff);
1471     elem = std::move(*ptr());
1472     destroyContents();
1473     sequencer_.completeTurn(turn * 2 + 1);
1474   }
1475 };
1476
1477 } // namespace detail
1478
1479 } // namespace folly