folly/MPMCQueue.h

   1 /*
   2  * Copyright 2015 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 #include <unistd.h>
  27
  28 #include <folly/Traits.h>
  29 #include <folly/detail/CacheLocality.h>
  30 #include <folly/detail/TurnSequencer.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 } // namespace detail
  42
  43 /// MPMCQueue<T> is a high-performance bounded concurrent queue that
  44 /// supports multiple producers, multiple consumers, and optional blocking.
  45 /// The queue has a fixed capacity, for which all memory will be allocated
  46 /// up front.  The bulk of the work of enqueuing and dequeuing can be
  47 /// performed in parallel.
  48 ///
  49 /// MPMCQueue is linearizable.  That means that if a call to write(A)
  50 /// returns before a call to write(B) begins, then A will definitely end up
  51 /// in the queue before B, and if a call to read(X) returns before a call
  52 /// to read(Y) is started, that X will be something from earlier in the
  53 /// queue than Y.  This also means that if a read call returns a value, you
  54 /// can be sure that all previous elements of the queue have been assigned
  55 /// a reader (that reader might not yet have returned, but it exists).
  56 ///
  57 /// The underlying implementation uses a ticket dispenser for the head and
  58 /// the tail, spreading accesses across N single-element queues to produce
  59 /// a queue with capacity N.  The ticket dispensers use atomic increment,
  60 /// which is more robust to contention than a CAS loop.  Each of the
  61 /// single-element queues uses its own CAS to serialize access, with an
  62 /// adaptive spin cutoff.  When spinning fails on a single-element queue
  63 /// it uses futex()'s _BITSET operations to reduce unnecessary wakeups
  64 /// even if multiple waiters are present on an individual queue (such as
  65 /// when the MPMCQueue's capacity is smaller than the number of enqueuers
  66 /// or dequeuers).
  67 ///
  68 /// In benchmarks (contained in tao/queues/ConcurrentQueueTests)
  69 /// it handles 1 to 1, 1 to N, N to 1, and N to M thread counts better
  70 /// than any of the alternatives present in fbcode, for both small (~10)
  71 /// and large capacities.  In these benchmarks it is also faster than
  72 /// tbb::concurrent_bounded_queue for all configurations.  When there are
  73 /// many more threads than cores, MPMCQueue is _much_ faster than the tbb
  74 /// queue because it uses futex() to block and unblock waiting threads,
  75 /// rather than spinning with sched_yield.
  76 ///
  77 /// NOEXCEPT INTERACTION: tl;dr; If it compiles you're fine.  Ticket-based
  78 /// queues separate the assignment of queue positions from the actual
  79 /// construction of the in-queue elements, which means that the T
  80 /// constructor used during enqueue must not throw an exception.  This is
  81 /// enforced at compile time using type traits, which requires that T be
  82 /// adorned with accurate noexcept information.  If your type does not
  83 /// use noexcept, you will have to wrap it in something that provides
  84 /// the guarantee.  We provide an alternate safe implementation for types
  85 /// that don't use noexcept but that are marked folly::IsRelocatable
  86 /// and boost::has_nothrow_constructor, which is common for folly types.
  87 /// In particular, if you can declare FOLLY_ASSUME_FBVECTOR_COMPATIBLE
  88 /// then your type can be put in MPMCQueue.
  89 ///
  90 /// If you have a pool of N queue consumers that you want to shut down
  91 /// after the queue has drained, one way is to enqueue N sentinel values
  92 /// to the queue.  If the producer doesn't know how many consumers there
  93 /// are you can enqueue one sentinel and then have each consumer requeue
  94 /// two sentinels after it receives it (by requeuing 2 the shutdown can
  95 /// complete in O(log P) time instead of O(P)).
  96 template<typename T,
  97          template<typename> class Atom = std::atomic>
  98 class MPMCQueue : boost::noncopyable {
  99
 100   static_assert(std::is_nothrow_constructible<T,T&&>::value ||
 101                 folly::IsRelocatable<T>::value,
 102       "T must be relocatable or have a noexcept move constructor");
 103
 104   friend class detail::MPMCPipelineStageImpl<T>;
 105  public:
 106   typedef T value_type;
 107
 108   explicit MPMCQueue(size_t queueCapacity)
 109     : capacity_(queueCapacity)
 110     , pushTicket_(0)
 111     , popTicket_(0)
 112     , pushSpinCutoff_(0)
 113     , popSpinCutoff_(0)
 114   {
 115     if (queueCapacity == 0)
 116       throw std::invalid_argument(
 117         "MPMCQueue with explicit capacity 0 is impossible"
 118       );
 119
 120     // would sigfpe if capacity is 0
 121     stride_ = computeStride(queueCapacity);
 122     slots_ = new detail::SingleElementQueue<T,Atom>[queueCapacity +
 123                                                     2 * kSlotPadding];
 124
 125     // ideally this would be a static assert, but g++ doesn't allow it
 126     assert(alignof(MPMCQueue<T,Atom>)
 127            >= detail::CacheLocality::kFalseSharingRange);
 128     assert(static_cast<uint8_t*>(static_cast<void*>(&popTicket_))
 129            - static_cast<uint8_t*>(static_cast<void*>(&pushTicket_))
 130            >= detail::CacheLocality::kFalseSharingRange);
 131   }
 132
 133   /// A default-constructed queue is useful because a usable (non-zero
 134   /// capacity) queue can be moved onto it or swapped with it
 135   MPMCQueue() noexcept
 136     : capacity_(0)
 137     , slots_(nullptr)
 138     , stride_(0)
 139     , pushTicket_(0)
 140     , popTicket_(0)
 141     , pushSpinCutoff_(0)
 142     , popSpinCutoff_(0)
 143   {}
 144
 145   /// IMPORTANT: The move constructor is here to make it easier to perform
 146   /// the initialization phase, it is not safe to use when there are any
 147   /// concurrent accesses (this is not checked).
 148   MPMCQueue(MPMCQueue<T,Atom>&& rhs) noexcept
 149     : capacity_(rhs.capacity_)
 150     , slots_(rhs.slots_)
 151     , stride_(rhs.stride_)
 152     , pushTicket_(rhs.pushTicket_.load(std::memory_order_relaxed))
 153     , popTicket_(rhs.popTicket_.load(std::memory_order_relaxed))
 154     , pushSpinCutoff_(rhs.pushSpinCutoff_.load(std::memory_order_relaxed))
 155     , popSpinCutoff_(rhs.popSpinCutoff_.load(std::memory_order_relaxed))
 156   {
 157     // relaxed ops are okay for the previous reads, since rhs queue can't
 158     // be in concurrent use
 159
 160     // zero out rhs
 161     rhs.capacity_ = 0;
 162     rhs.slots_ = nullptr;
 163     rhs.stride_ = 0;
 164     rhs.pushTicket_.store(0, std::memory_order_relaxed);
 165     rhs.popTicket_.store(0, std::memory_order_relaxed);
 166     rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
 167     rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
 168   }
 169
 170   /// IMPORTANT: The move operator is here to make it easier to perform
 171   /// the initialization phase, it is not safe to use when there are any
 172   /// concurrent accesses (this is not checked).
 173   MPMCQueue<T,Atom> const& operator= (MPMCQueue<T,Atom>&& rhs) {
 174     if (this != &rhs) {
 175       this->~MPMCQueue();
 176       new (this) MPMCQueue(std::move(rhs));
 177     }
 178     return *this;
 179   }
 180
 181   /// MPMCQueue can only be safely destroyed when there are no
 182   /// pending enqueuers or dequeuers (this is not checked).
 183   ~MPMCQueue() {
 184     delete[] slots_;
 185   }
 186
 187   /// Returns the number of successful reads minus the number of successful
 188   /// writes.  Waiting blockingRead and blockingWrite calls are included,
 189   /// so this value can be negative.
 190   ssize_t size() const noexcept {
 191     // since both pushes and pops increase monotonically, we can get a
 192     // consistent snapshot either by bracketing a read of popTicket_ with
 193     // two reads of pushTicket_ that return the same value, or the other
 194     // way around.  We maximize our chances by alternately attempting
 195     // both bracketings.
 196     uint64_t pushes = pushTicket_.load(std::memory_order_acquire); // A
 197     uint64_t pops = popTicket_.load(std::memory_order_acquire); // B
 198     while (true) {
 199       uint64_t nextPushes = pushTicket_.load(std::memory_order_acquire); // C
 200       if (pushes == nextPushes) {
 201         // pushTicket_ didn't change from A (or the previous C) to C,
 202         // so we can linearize at B (or D)
 203         return pushes - pops;
 204       }
 205       pushes = nextPushes;
 206       uint64_t nextPops = popTicket_.load(std::memory_order_acquire); // D
 207       if (pops == nextPops) {
 208         // popTicket_ didn't chance from B (or the previous D), so we
 209         // can linearize at C
 210         return pushes - pops;
 211       }
 212       pops = nextPops;
 213     }
 214   }
 215
 216   /// Returns true if there are no items available for dequeue
 217   bool isEmpty() const noexcept {
 218     return size() <= 0;
 219   }
 220
 221   /// Returns true if there is currently no empty space to enqueue
 222   bool isFull() const noexcept {
 223     // careful with signed -> unsigned promotion, since size can be negative
 224     return size() >= static_cast<ssize_t>(capacity_);
 225   }
 226
 227   /// Returns is a guess at size() for contexts that don't need a precise
 228   /// value, such as stats.
 229   ssize_t sizeGuess() const noexcept {
 230     return writeCount() - readCount();
 231   }
 232
 233   /// Doesn't change
 234   size_t capacity() const noexcept {
 235     return capacity_;
 236   }
 237
 238   /// Returns the total number of calls to blockingWrite or successful
 239   /// calls to write, including those blockingWrite calls that are
 240   /// currently blocking
 241   uint64_t writeCount() const noexcept {
 242     return pushTicket_.load(std::memory_order_acquire);
 243   }
 244
 245   /// Returns the total number of calls to blockingRead or successful
 246   /// calls to read, including those blockingRead calls that are currently
 247   /// blocking
 248   uint64_t readCount() const noexcept {
 249     return popTicket_.load(std::memory_order_acquire);
 250   }
 251
 252   /// Enqueues a T constructed from args, blocking until space is
 253   /// available.  Note that this method signature allows enqueue via
 254   /// move, if args is a T rvalue, via copy, if args is a T lvalue, or
 255   /// via emplacement if args is an initializer list that can be passed
 256   /// to a T constructor.
 257   template <typename ...Args>
 258   void blockingWrite(Args&&... args) noexcept {
 259     enqueueWithTicket(pushTicket_++, std::forward<Args>(args)...);
 260   }
 261
 262   /// If an item can be enqueued with no blocking, does so and returns
 263   /// true, otherwise returns false.  This method is similar to
 264   /// writeIfNotFull, but if you don't have a specific need for that
 265   /// method you should use this one.
 266   ///
 267   /// One of the common usages of this method is to enqueue via the
 268   /// move constructor, something like q.write(std::move(x)).  If write
 269   /// returns false because the queue is full then x has not actually been
 270   /// consumed, which looks strange.  To understand why it is actually okay
 271   /// to use x afterward, remember that std::move is just a typecast that
 272   /// provides an rvalue reference that enables use of a move constructor
 273   /// or operator.  std::move doesn't actually move anything.  It could
 274   /// more accurately be called std::rvalue_cast or std::move_permission.
 275   template <typename ...Args>
 276   bool write(Args&&... args) noexcept {
 277     uint64_t ticket;
 278     if (tryObtainReadyPushTicket(ticket)) {
 279       // we have pre-validated that the ticket won't block
 280       enqueueWithTicket(ticket, std::forward<Args>(args)...);
 281       return true;
 282     } else {
 283       return false;
 284     }
 285   }
 286
 287   /// If the queue is not full, enqueues and returns true, otherwise
 288   /// returns false.  Unlike write this method can be blocked by another
 289   /// thread, specifically a read that has linearized (been assigned
 290   /// a ticket) but not yet completed.  If you don't really need this
 291   /// function you should probably use write.
 292   ///
 293   /// MPMCQueue isn't lock-free, so just because a read operation has
 294   /// linearized (and isFull is false) doesn't mean that space has been
 295   /// made available for another write.  In this situation write will
 296   /// return false, but writeIfNotFull will wait for the dequeue to finish.
 297   /// This method is required if you are composing queues and managing
 298   /// your own wakeup, because it guarantees that after every successful
 299   /// write a readIfNotEmpty will succeed.
 300   template <typename ...Args>
 301   bool writeIfNotFull(Args&&... args) noexcept {
 302     uint64_t ticket;
 303     if (tryObtainPromisedPushTicket(ticket)) {
 304       // some other thread is already dequeuing the slot into which we
 305       // are going to enqueue, but we might have to wait for them to finish
 306       enqueueWithTicket(ticket, std::forward<Args>(args)...);
 307       return true;
 308     } else {
 309       return false;
 310     }
 311   }
 312
 313   /// Moves a dequeued element onto elem, blocking until an element
 314   /// is available
 315   void blockingRead(T& elem) noexcept {
 316     dequeueWithTicket(popTicket_++, elem);
 317   }
 318
 319   /// If an item can be dequeued with no blocking, does so and returns
 320   /// true, otherwise returns false.
 321   bool read(T& elem) noexcept {
 322     uint64_t ticket;
 323     if (tryObtainReadyPopTicket(ticket)) {
 324       // the ticket has been pre-validated to not block
 325       dequeueWithTicket(ticket, elem);
 326       return true;
 327     } else {
 328       return false;
 329     }
 330   }
 331
 332   /// If the queue is not empty, dequeues and returns true, otherwise
 333   /// returns false.  If the matching write is still in progress then this
 334   /// method may block waiting for it.  If you don't rely on being able
 335   /// to dequeue (such as by counting completed write) then you should
 336   /// prefer read.
 337   bool readIfNotEmpty(T& elem) noexcept {
 338     uint64_t ticket;
 339     if (tryObtainPromisedPopTicket(ticket)) {
 340       // the matching enqueue already has a ticket, but might not be done
 341       dequeueWithTicket(ticket, elem);
 342       return true;
 343     } else {
 344       return false;
 345     }
 346   }
 347
 348  private:
 349   enum {
 350     /// Once every kAdaptationFreq we will spin longer, to try to estimate
 351     /// the proper spin backoff
 352     kAdaptationFreq = 128,
 353
 354     /// To avoid false sharing in slots_ with neighboring memory
 355     /// allocations, we pad it with this many SingleElementQueue-s at
 356     /// each end
 357     kSlotPadding = (detail::CacheLocality::kFalseSharingRange - 1)
 358         / sizeof(detail::SingleElementQueue<T,Atom>) + 1
 359   };
 360
 361   /// The maximum number of items in the queue at once
 362   size_t FOLLY_ALIGN_TO_AVOID_FALSE_SHARING capacity_;
 363
 364   /// An array of capacity_ SingleElementQueue-s, each of which holds
 365   /// either 0 or 1 item.  We over-allocate by 2 * kSlotPadding and don't
 366   /// touch the slots at either end, to avoid false sharing
 367   detail::SingleElementQueue<T,Atom>* slots_;
 368
 369   /// The number of slots_ indices that we advance for each ticket, to
 370   /// avoid false sharing.  Ideally slots_[i] and slots_[i + stride_]
 371   /// aren't on the same cache line
 372   int stride_;
 373
 374   /// Enqueuers get tickets from here
 375   Atom<uint64_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING pushTicket_;
 376
 377   /// Dequeuers get tickets from here
 378   Atom<uint64_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING popTicket_;
 379
 380   /// This is how many times we will spin before using FUTEX_WAIT when
 381   /// the queue is full on enqueue, adaptively computed by occasionally
 382   /// spinning for longer and smoothing with an exponential moving average
 383   Atom<uint32_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING pushSpinCutoff_;
 384
 385   /// The adaptive spin cutoff when the queue is empty on dequeue
 386   Atom<uint32_t> FOLLY_ALIGN_TO_AVOID_FALSE_SHARING popSpinCutoff_;
 387
 388   /// Alignment doesn't prevent false sharing at the end of the struct,
 389   /// so fill out the last cache line
 390   char padding_[detail::CacheLocality::kFalseSharingRange -
 391                 sizeof(Atom<uint32_t>)];
 392
 393
 394   /// We assign tickets in increasing order, but we don't want to
 395   /// access neighboring elements of slots_ because that will lead to
 396   /// false sharing (multiple cores accessing the same cache line even
 397   /// though they aren't accessing the same bytes in that cache line).
 398   /// To avoid this we advance by stride slots per ticket.
 399   ///
 400   /// We need gcd(capacity, stride) to be 1 so that we will use all
 401   /// of the slots.  We ensure this by only considering prime strides,
 402   /// which either have no common divisors with capacity or else have
 403   /// a zero remainder after dividing by capacity.  That is sufficient
 404   /// to guarantee correctness, but we also want to actually spread the
 405   /// accesses away from each other to avoid false sharing (consider a
 406   /// stride of 7 with a capacity of 8).  To that end we try a few taking
 407   /// care to observe that advancing by -1 is as bad as advancing by 1
 408   /// when in comes to false sharing.
 409   ///
 410   /// The simple way to avoid false sharing would be to pad each
 411   /// SingleElementQueue, but since we have capacity_ of them that could
 412   /// waste a lot of space.
 413   static int computeStride(size_t capacity) noexcept {
 414     static const int smallPrimes[] = { 2, 3, 5, 7, 11, 13, 17, 19, 23 };
 415
 416     int bestStride = 1;
 417     size_t bestSep = 1;
 418     for (int stride : smallPrimes) {
 419       if ((stride % capacity) == 0 || (capacity % stride) == 0) {
 420         continue;
 421       }
 422       size_t sep = stride % capacity;
 423       sep = std::min(sep, capacity - sep);
 424       if (sep > bestSep) {
 425         bestStride = stride;
 426         bestSep = sep;
 427       }
 428     }
 429     return bestStride;
 430   }
 431
 432   /// Returns the index into slots_ that should be used when enqueuing or
 433   /// dequeuing with the specified ticket
 434   size_t idx(uint64_t ticket) noexcept {
 435     return ((ticket * stride_) % capacity_) + kSlotPadding;
 436   }
 437
 438   /// Maps an enqueue or dequeue ticket to the turn should be used at the
 439   /// corresponding SingleElementQueue
 440   uint32_t turn(uint64_t ticket) noexcept {
 441     return ticket / capacity_;
 442   }
 443
 444   /// Tries to obtain a push ticket for which SingleElementQueue::enqueue
 445   /// won't block.  Returns true on immediate success, false on immediate
 446   /// failure.
 447   bool tryObtainReadyPushTicket(uint64_t& rv) noexcept {
 448     auto ticket = pushTicket_.load(std::memory_order_acquire); // A
 449     while (true) {
 450       if (!slots_[idx(ticket)].mayEnqueue(turn(ticket))) {
 451         // if we call enqueue(ticket, ...) on the SingleElementQueue
 452         // right now it would block, but this might no longer be the next
 453         // ticket.  We can increase the chance of tryEnqueue success under
 454         // contention (without blocking) by rechecking the ticket dispenser
 455         auto prev = ticket;
 456         ticket = pushTicket_.load(std::memory_order_acquire); // B
 457         if (prev == ticket) {
 458           // mayEnqueue was bracketed by two reads (A or prev B or prev
 459           // failing CAS to B), so we are definitely unable to enqueue
 460           return false;
 461         }
 462       } else {
 463         // we will bracket the mayEnqueue check with a read (A or prev B
 464         // or prev failing CAS) and the following CAS.  If the CAS fails
 465         // it will effect a load of pushTicket_
 466         if (pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
 467           rv = ticket;
 468           return true;
 469         }
 470       }
 471     }
 472   }
 473
 474   /// Tries to obtain a push ticket which can be satisfied if all
 475   /// in-progress pops complete.  This function does not block, but
 476   /// blocking may be required when using the returned ticket if some
 477   /// other thread's pop is still in progress (ticket has been granted but
 478   /// pop has not yet completed).
 479   bool tryObtainPromisedPushTicket(uint64_t& rv) noexcept {
 480     auto numPushes = pushTicket_.load(std::memory_order_acquire); // A
 481     while (true) {
 482       auto numPops = popTicket_.load(std::memory_order_acquire); // B
 483       // n will be negative if pops are pending
 484       int64_t n = numPushes - numPops;
 485       if (n >= static_cast<ssize_t>(capacity_)) {
 486         // Full, linearize at B.  We don't need to recheck the read we
 487         // performed at A, because if numPushes was stale at B then the
 488         // real numPushes value is even worse
 489         return false;
 490       }
 491       if (pushTicket_.compare_exchange_strong(numPushes, numPushes + 1)) {
 492         rv = numPushes;
 493         return true;
 494       }
 495     }
 496   }
 497
 498   /// Tries to obtain a pop ticket for which SingleElementQueue::dequeue
 499   /// won't block.  Returns true on immediate success, false on immediate
 500   /// failure.
 501   bool tryObtainReadyPopTicket(uint64_t& rv) noexcept {
 502     auto ticket = popTicket_.load(std::memory_order_acquire);
 503     while (true) {
 504       if (!slots_[idx(ticket)].mayDequeue(turn(ticket))) {
 505         auto prev = ticket;
 506         ticket = popTicket_.load(std::memory_order_acquire);
 507         if (prev == ticket) {
 508           return false;
 509         }
 510       } else {
 511         if (popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
 512           rv = ticket;
 513           return true;
 514         }
 515       }
 516     }
 517   }
 518
 519   /// Similar to tryObtainReadyPopTicket, but returns a pop ticket whose
 520   /// corresponding push ticket has already been handed out, rather than
 521   /// returning one whose corresponding push ticket has already been
 522   /// completed.  This means that there is a possibility that the caller
 523   /// will block when using the ticket, but it allows the user to rely on
 524   /// the fact that if enqueue has succeeded, tryObtainPromisedPopTicket
 525   /// will return true.  The "try" part of this is that we won't have
 526   /// to block waiting for someone to call enqueue, although we might
 527   /// have to block waiting for them to finish executing code inside the
 528   /// MPMCQueue itself.
 529   bool tryObtainPromisedPopTicket(uint64_t& rv) noexcept {
 530     auto numPops = popTicket_.load(std::memory_order_acquire); // A
 531     while (true) {
 532       auto numPushes = pushTicket_.load(std::memory_order_acquire); // B
 533       if (numPops >= numPushes) {
 534         // Empty, or empty with pending pops.  Linearize at B.  We don't
 535         // need to recheck the read we performed at A, because if numPops
 536         // is stale then the fresh value is larger and the >= is still true
 537         return false;
 538       }
 539       if (popTicket_.compare_exchange_strong(numPops, numPops + 1)) {
 540         rv = numPops;
 541         return true;
 542       }
 543     }
 544   }
 545
 546   // Given a ticket, constructs an enqueued item using args
 547   template <typename ...Args>
 548   void enqueueWithTicket(uint64_t ticket, Args&&... args) noexcept {
 549     slots_[idx(ticket)].enqueue(turn(ticket),
 550                                 pushSpinCutoff_,
 551                                 (ticket % kAdaptationFreq) == 0,
 552                                 std::forward<Args>(args)...);
 553   }
 554
 555   // Given a ticket, dequeues the corresponding element
 556   void dequeueWithTicket(uint64_t ticket, T& elem) noexcept {
 557     slots_[idx(ticket)].dequeue(turn(ticket),
 558                                 popSpinCutoff_,
 559                                 (ticket % kAdaptationFreq) == 0,
 560                                 elem);
 561   }
 562 };
 563
 564
 565 namespace detail {
 566
 567 /// SingleElementQueue implements a blocking queue that holds at most one
 568 /// item, and that requires its users to assign incrementing identifiers
 569 /// (turns) to each enqueue and dequeue operation.  Note that the turns
 570 /// used by SingleElementQueue are doubled inside the TurnSequencer
 571 template <typename T, template <typename> class Atom>
 572 struct SingleElementQueue {
 573
 574   ~SingleElementQueue() noexcept {
 575     if ((sequencer_.uncompletedTurnLSB() & 1) == 1) {
 576       // we are pending a dequeue, so we have a constructed item
 577       destroyContents();
 578     }
 579   }
 580
 581   /// enqueue using in-place noexcept construction
 582   template <typename ...Args,
 583             typename = typename std::enable_if<
 584               std::is_nothrow_constructible<T,Args...>::value>::type>
 585   void enqueue(const uint32_t turn,
 586                Atom<uint32_t>& spinCutoff,
 587                const bool updateSpinCutoff,
 588                Args&&... args) noexcept {
 589     sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
 590     new (&contents_) T(std::forward<Args>(args)...);
 591     sequencer_.completeTurn(turn * 2);
 592   }
 593
 594   /// enqueue using move construction, either real (if
 595   /// is_nothrow_move_constructible) or simulated using relocation and
 596   /// default construction (if IsRelocatable and has_nothrow_constructor)
 597   template <typename = typename std::enable_if<
 598                 (folly::IsRelocatable<T>::value &&
 599                  boost::has_nothrow_constructor<T>::value) ||
 600                 std::is_nothrow_constructible<T,T&&>::value>::type>
 601   void enqueue(const uint32_t turn,
 602                Atom<uint32_t>& spinCutoff,
 603                const bool updateSpinCutoff,
 604                T&& goner) noexcept {
 605     enqueueImpl(
 606         turn,
 607         spinCutoff,
 608         updateSpinCutoff,
 609         std::move(goner),
 610         typename std::conditional<std::is_nothrow_constructible<T,T&&>::value,
 611                                   ImplByMove, ImplByRelocation>::type());
 612   }
 613
 614   bool mayEnqueue(const uint32_t turn) const noexcept {
 615     return sequencer_.isTurn(turn * 2);
 616   }
 617
 618   void dequeue(uint32_t turn,
 619                Atom<uint32_t>& spinCutoff,
 620                const bool updateSpinCutoff,
 621                T& elem) noexcept {
 622     dequeueImpl(turn,
 623                 spinCutoff,
 624                 updateSpinCutoff,
 625                 elem,
 626                 typename std::conditional<folly::IsRelocatable<T>::value,
 627                                           ImplByRelocation,
 628                                           ImplByMove>::type());
 629   }
 630
 631   bool mayDequeue(const uint32_t turn) const noexcept {
 632     return sequencer_.isTurn(turn * 2 + 1);
 633   }
 634
 635  private:
 636   /// Storage for a T constructed with placement new
 637   typename std::aligned_storage<sizeof(T),alignof(T)>::type contents_;
 638
 639   /// Even turns are pushes, odd turns are pops
 640   TurnSequencer<Atom> sequencer_;
 641
 642   T* ptr() noexcept {
 643     return static_cast<T*>(static_cast<void*>(&contents_));
 644   }
 645
 646   void destroyContents() noexcept {
 647     try {
 648       ptr()->~T();
 649     } catch (...) {
 650       // g++ doesn't seem to have std::is_nothrow_destructible yet
 651     }
 652 #ifndef NDEBUG
 653     memset(&contents_, 'Q', sizeof(T));
 654 #endif
 655   }
 656
 657   /// Tag classes for dispatching to enqueue/dequeue implementation.
 658   struct ImplByRelocation {};
 659   struct ImplByMove {};
 660
 661   /// enqueue using nothrow move construction.
 662   void enqueueImpl(const uint32_t turn,
 663                    Atom<uint32_t>& spinCutoff,
 664                    const bool updateSpinCutoff,
 665                    T&& goner,
 666                    ImplByMove) noexcept {
 667     sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
 668     new (&contents_) T(std::move(goner));
 669     sequencer_.completeTurn(turn * 2);
 670   }
 671
 672   /// enqueue by simulating nothrow move with relocation, followed by
 673   /// default construction to a noexcept relocation.
 674   void enqueueImpl(const uint32_t turn,
 675                    Atom<uint32_t>& spinCutoff,
 676                    const bool updateSpinCutoff,
 677                    T&& goner,
 678                    ImplByRelocation) noexcept {
 679     sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
 680     memcpy(&contents_, &goner, sizeof(T));
 681     sequencer_.completeTurn(turn * 2);
 682     new (&goner) T();
 683   }
 684
 685   /// dequeue by destructing followed by relocation.  This version is preferred,
 686   /// because as much work as possible can be done before waiting.
 687   void dequeueImpl(uint32_t turn,
 688                    Atom<uint32_t>& spinCutoff,
 689                    const bool updateSpinCutoff,
 690                    T& elem,
 691                    ImplByRelocation) noexcept {
 692     try {
 693       elem.~T();
 694     } catch (...) {
 695       // unlikely, but if we don't complete our turn the queue will die
 696     }
 697     sequencer_.waitForTurn(turn * 2 + 1, spinCutoff, updateSpinCutoff);
 698     memcpy(&elem, &contents_, sizeof(T));
 699     sequencer_.completeTurn(turn * 2 + 1);
 700   }
 701
 702   /// dequeue by nothrow move assignment.
 703   void dequeueImpl(uint32_t turn,
 704                    Atom<uint32_t>& spinCutoff,
 705                    const bool updateSpinCutoff,
 706                    T& elem,
 707                    ImplByMove) noexcept {
 708     sequencer_.waitForTurn(turn * 2 + 1, spinCutoff, updateSpinCutoff);
 709     elem = std::move(*ptr());
 710     destroyContents();
 711     sequencer_.completeTurn(turn * 2 + 1);
 712   }
 713 };
 714
 715 } // namespace detail
 716
 717 } // namespace folly