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