folly/MPMCQueue.h

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