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