MPMCQueue.h

Go to the documentation of this file.

/*
 * Copyright 2013-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include <algorithm>
#include <atomic>
#include <cassert>
#include <cstring>
#include <limits>
#include <type_traits>

#include <boost/noncopyable.hpp>

#include <folly/Traits.h>
#include <folly/concurrency/CacheLocality.h>
#include <folly/detail/TurnSequencer.h>
#include <folly/portability/Unistd.h>

namespace folly {

namespace detail {

template <typename T, template <typename> class Atom>
struct SingleElementQueue;

template <typename T>
class MPMCPipelineStageImpl;

template <typename>
class MPMCQueueBase;

} // namespace detail

template <
    typename T,
    template <typename> class Atom = std::atomic,
    bool Dynamic = false>
class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T, Atom, Dynamic>> {
  friend class detail::MPMCPipelineStageImpl<T>;
  using Slot = detail::SingleElementQueue<T, Atom>;

 public:
  explicit MPMCQueue(size_t queueCapacity)
      : detail::MPMCQueueBase<MPMCQueue<T, Atom, Dynamic>>(queueCapacity) {
    this->stride_ = this->computeStride(queueCapacity);
    this->slots_ = new Slot[queueCapacity + 2 * this->kSlotPadding];
  }

  MPMCQueue() noexcept {}
};

template <typename T, template <typename> class Atom>
class MPMCQueue<T, Atom, true>
    : public detail::MPMCQueueBase<MPMCQueue<T, Atom, true>> {
  friend class detail::MPMCQueueBase<MPMCQueue<T, Atom, true>>;
  using Slot = detail::SingleElementQueue<T, Atom>;

  struct ClosedArray {
    uint64_t offset_{0};
    Slot* slots_{nullptr};
    size_t capacity_{0};
    int stride_{0};
  };

 public:
  explicit MPMCQueue(size_t queueCapacity)
      : detail::MPMCQueueBase<MPMCQueue<T, Atom, true>>(queueCapacity) {
    size_t cap = std::min<size_t>(kDefaultMinDynamicCapacity, queueCapacity);
    initQueue(cap, kDefaultExpansionMultiplier);
  }

  explicit MPMCQueue(
      size_t queueCapacity,
      size_t minCapacity,
      size_t expansionMultiplier)
      : detail::MPMCQueueBase<MPMCQueue<T, Atom, true>>(queueCapacity) {
    minCapacity = std::max<size_t>(1, minCapacity);
    size_t cap = std::min<size_t>(minCapacity, queueCapacity);
    expansionMultiplier = std::max<size_t>(2, expansionMultiplier);
    initQueue(cap, expansionMultiplier);
  }

  MPMCQueue() noexcept {
    dmult_ = 0;
    closed_ = nullptr;
  }

  MPMCQueue(MPMCQueue<T, Atom, true>&& rhs) noexcept {
    this->capacity_ = rhs.capacity_;
    new (&this->dslots_)
        Atom<Slot*>(rhs.dslots_.load(std::memory_order_relaxed));
    new (&this->dstride_)
        Atom<int>(rhs.dstride_.load(std::memory_order_relaxed));
    this->dstate_.store(
        rhs.dstate_.load(std::memory_order_relaxed), std::memory_order_relaxed);
    this->dcapacity_.store(
        rhs.dcapacity_.load(std::memory_order_relaxed),
        std::memory_order_relaxed);
    this->pushTicket_.store(
        rhs.pushTicket_.load(std::memory_order_relaxed),
        std::memory_order_relaxed);
    this->popTicket_.store(
        rhs.popTicket_.load(std::memory_order_relaxed),
        std::memory_order_relaxed);
    this->pushSpinCutoff_.store(
        rhs.pushSpinCutoff_.load(std::memory_order_relaxed),
        std::memory_order_relaxed);
    this->popSpinCutoff_.store(
        rhs.popSpinCutoff_.load(std::memory_order_relaxed),
        std::memory_order_relaxed);
    dmult_ = rhs.dmult_;
    closed_ = rhs.closed_;

    rhs.capacity_ = 0;
    rhs.dslots_.store(nullptr, std::memory_order_relaxed);
    rhs.dstride_.store(0, std::memory_order_relaxed);
    rhs.dstate_.store(0, std::memory_order_relaxed);
    rhs.dcapacity_.store(0, std::memory_order_relaxed);
    rhs.pushTicket_.store(0, std::memory_order_relaxed);
    rhs.popTicket_.store(0, std::memory_order_relaxed);
    rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
    rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
    rhs.dmult_ = 0;
    rhs.closed_ = nullptr;
  }

  MPMCQueue<T, Atom, true> const& operator=(MPMCQueue<T, Atom, true>&& rhs) {
    if (this != &rhs) {
      this->~MPMCQueue();
      new (this) MPMCQueue(std::move(rhs));
    }
    return *this;
  }

  ~MPMCQueue() {
    if (closed_ != nullptr) {
      for (int i = getNumClosed(this->dstate_.load()) - 1; i >= 0; --i) {
        delete[] closed_[i].slots_;
      }
      delete[] closed_;
    }
    using AtomInt = Atom<int>;
    this->dstride_.~AtomInt();
    using AtomSlot = Atom<Slot*>;
    // Sort of a hack to get ~MPMCQueueBase to free dslots_
    auto slots = this->dslots_.load();
    this->dslots_.~AtomSlot();
    this->slots_ = slots;
  }

  size_t allocatedCapacity() const noexcept {
    return this->dcapacity_.load(std::memory_order_relaxed);
  }

  template <typename... Args>
  void blockingWrite(Args&&... args) noexcept {
    uint64_t ticket = this->pushTicket_++;
    Slot* slots;
    size_t cap;
    int stride;
    uint64_t state;
    uint64_t offset;
    do {
      if (!trySeqlockReadSection(state, slots, cap, stride)) {
        asm_volatile_pause();
        continue;
      }
      if (maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride)) {
        // There was an expansion after this ticket was issued.
        break;
      }
      if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
              this->turn(ticket - offset, cap))) {
        // A slot is ready. No need to expand.
        break;
      } else if (
          this->popTicket_.load(std::memory_order_relaxed) + cap > ticket) {
        // May block, but a pop is in progress. No need to expand.
        // Get seqlock read section info again in case an expansion
        // occurred with an equal or higher ticket.
        continue;
      } else {
        // May block. See if we can expand.
        if (tryExpand(state, cap)) {
          // This or another thread started an expansion. Get updated info.
          continue;
        } else {
          // Can't expand.
          break;
        }
      }
    } while (true);
    this->enqueueWithTicketBase(
        ticket - offset, slots, cap, stride, std::forward<Args>(args)...);
  }

  void blockingReadWithTicket(uint64_t& ticket, T& elem) noexcept {
    ticket = this->popTicket_++;
    Slot* slots;
    size_t cap;
    int stride;
    uint64_t state;
    uint64_t offset;
    while (!trySeqlockReadSection(state, slots, cap, stride)) {
      asm_volatile_pause();
    }
    // If there was an expansion after the corresponding push ticket
    // was issued, adjust accordingly
    maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
    this->dequeueWithTicketBase(ticket - offset, slots, cap, stride, elem);
  }

 private:
  enum {
    kSeqlockBits = 6,
    kDefaultMinDynamicCapacity = 10,
    kDefaultExpansionMultiplier = 10,
  };

  size_t dmult_;

  //  Info about closed slots arrays for use by lagging operations
  ClosedArray* closed_;

  void initQueue(const size_t cap, const size_t mult) {
    new (&this->dstride_) Atom<int>(this->computeStride(cap));
    Slot* slots = new Slot[cap + 2 * this->kSlotPadding];
    new (&this->dslots_) Atom<Slot*>(slots);
    this->dstate_.store(0);
    this->dcapacity_.store(cap);
    dmult_ = mult;
    size_t maxClosed = 0;
    for (size_t expanded = cap; expanded < this->capacity_; expanded *= mult) {
      ++maxClosed;
    }
    closed_ = (maxClosed > 0) ? new ClosedArray[maxClosed] : nullptr;
  }

  bool tryObtainReadyPushTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    uint64_t state;
    do {
      ticket = this->pushTicket_.load(std::memory_order_acquire); // A
      if (!trySeqlockReadSection(state, slots, cap, stride)) {
        asm_volatile_pause();
        continue;
      }

      // If there was an expansion with offset greater than this ticket,
      // adjust accordingly
      uint64_t offset;
      maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);

      if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
              this->turn(ticket - offset, cap))) {
        // A slot is ready.
        if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
          // Adjust ticket
          ticket -= offset;
          return true;
        } else {
          continue;
        }
      } else {
        if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B
          // Try again. Ticket changed.
          continue;
        }
        // Likely to block.
        // Try to expand unless the ticket is for a closed array
        if (offset == getOffset(state)) {
          if (tryExpand(state, cap)) {
            // This or another thread started an expansion. Get up-to-date info.
            continue;
          }
        }
        return false;
      }
    } while (true);
  }

  bool tryObtainPromisedPushTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    uint64_t state;
    do {
      ticket = this->pushTicket_.load(std::memory_order_acquire);
      auto numPops = this->popTicket_.load(std::memory_order_acquire);
      if (!trySeqlockReadSection(state, slots, cap, stride)) {
        asm_volatile_pause();
        continue;
      }

      const auto curCap = cap;
      // If there was an expansion with offset greater than this ticket,
      // adjust accordingly
      uint64_t offset;
      maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);

      int64_t n = ticket - numPops;

      if (n >= static_cast<ssize_t>(cap)) {
        if ((cap == curCap) && tryExpand(state, cap)) {
          // This or another thread started an expansion. Start over.
          continue;
        }
        // Can't expand.
        ticket -= offset;
        return false;
      }

      if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
        // Adjust ticket
        ticket -= offset;
        return true;
      }
    } while (true);
  }

  bool tryObtainReadyPopTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    uint64_t state;
    do {
      ticket = this->popTicket_.load(std::memory_order_relaxed);
      if (!trySeqlockReadSection(state, slots, cap, stride)) {
        asm_volatile_pause();
        continue;
      }

      // If there was an expansion after the corresponding push ticket
      // was issued, adjust accordingly
      uint64_t offset;
      maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);

      if (slots[this->idx((ticket - offset), cap, stride)].mayDequeue(
              this->turn(ticket - offset, cap))) {
        if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
          // Adjust ticket
          ticket -= offset;
          return true;
        }
      } else {
        return false;
      }
    } while (true);
  }

  bool tryObtainPromisedPopTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    uint64_t state;
    do {
      ticket = this->popTicket_.load(std::memory_order_acquire);
      auto numPushes = this->pushTicket_.load(std::memory_order_acquire);
      if (!trySeqlockReadSection(state, slots, cap, stride)) {
        asm_volatile_pause();
        continue;
      }

      uint64_t offset;
      // If there was an expansion after the corresponding push
      // ticket was issued, adjust accordingly
      maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);

      if (ticket >= numPushes) {
        ticket -= offset;
        return false;
      }
      if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
        ticket -= offset;
        return true;
      }
    } while (true);
  }

  template <typename... Args>
  void enqueueWithTicket(const uint64_t ticket, Args&&... args) noexcept {
    Slot* slots;
    size_t cap;
    int stride;
    uint64_t state;
    uint64_t offset;

    while (!trySeqlockReadSection(state, slots, cap, stride)) {
    }

    // If there was an expansion after this ticket was issued, adjust
    // accordingly
    maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);

    this->enqueueWithTicketBase(
        ticket - offset, slots, cap, stride, std::forward<Args>(args)...);
  }

  uint64_t getOffset(const uint64_t state) const noexcept {
    return state >> kSeqlockBits;
  }

  int getNumClosed(const uint64_t state) const noexcept {
    return (state & ((1 << kSeqlockBits) - 1)) >> 1;
  }

  bool tryExpand(const uint64_t state, const size_t cap) noexcept {
    if (cap == this->capacity_) {
      return false;
    }
    // Acquire seqlock
    uint64_t oldval = state;
    assert((state & 1) == 0);
    if (this->dstate_.compare_exchange_strong(oldval, state + 1)) {
      assert(cap == this->dcapacity_.load());
      uint64_t ticket =
          1 + std::max(this->pushTicket_.load(), this->popTicket_.load());
      size_t newCapacity = std::min(dmult_ * cap, this->capacity_);
      Slot* newSlots =
          new (std::nothrow) Slot[newCapacity + 2 * this->kSlotPadding];
      if (newSlots == nullptr) {
        // Expansion failed. Restore the seqlock
        this->dstate_.store(state);
        return false;
      }
      // Successful expansion
      // calculate the current ticket offset
      uint64_t offset = getOffset(state);
      // calculate index in closed array
      int index = getNumClosed(state);
      assert((index << 1) < (1 << kSeqlockBits));
      // fill the info for the closed slots array
      closed_[index].offset_ = offset;
      closed_[index].slots_ = this->dslots_.load();
      closed_[index].capacity_ = cap;
      closed_[index].stride_ = this->dstride_.load();
      // update the new slots array info
      this->dslots_.store(newSlots);
      this->dcapacity_.store(newCapacity);
      this->dstride_.store(this->computeStride(newCapacity));
      // Release the seqlock and record the new ticket offset
      this->dstate_.store((ticket << kSeqlockBits) + (2 * (index + 1)));
      return true;
    } else { // failed to acquire seqlock
      // Someone acaquired the seqlock. Go back to the caller and get
      // up-to-date info.
      return true;
    }
  }

  bool trySeqlockReadSection(
      uint64_t& state,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    state = this->dstate_.load(std::memory_order_acquire);
    if (state & 1) {
      // Locked.
      return false;
    }
    // Start read-only section.
    slots = this->dslots_.load(std::memory_order_relaxed);
    cap = this->dcapacity_.load(std::memory_order_relaxed);
    stride = this->dstride_.load(std::memory_order_relaxed);
    // End of read-only section. Validate seqlock.
    std::atomic_thread_fence(std::memory_order_acquire);
    return (state == this->dstate_.load(std::memory_order_relaxed));
  }

  bool maybeUpdateFromClosed(
      const uint64_t state,
      const uint64_t ticket,
      uint64_t& offset,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    offset = getOffset(state);
    if (ticket >= offset) {
      return false;
    }
    for (int i = getNumClosed(state) - 1; i >= 0; --i) {
      offset = closed_[i].offset_;
      if (offset <= ticket) {
        slots = closed_[i].slots_;
        cap = closed_[i].capacity_;
        stride = closed_[i].stride_;
        return true;
      }
    }
    // A closed array with offset <= ticket should have been found
    assert(false);
    return false;
  }
};

namespace detail {

template <
    template <typename T, template <typename> class Atom, bool Dynamic>
    class Derived,
    typename T,
    template <typename> class Atom,
    bool Dynamic>
class MPMCQueueBase<Derived<T, Atom, Dynamic>> : boost::noncopyable {
  // Note: Using CRTP static casts in several functions of this base
  // template instead of making called functions virtual or duplicating
  // the code of calling functions in the derived partially specialized
  // template

  static_assert(
      std::is_nothrow_constructible<T, T&&>::value ||
          folly::IsRelocatable<T>::value,
      "T must be relocatable or have a noexcept move constructor");

 public:
  typedef T value_type;

  using Slot = detail::SingleElementQueue<T, Atom>;

  explicit MPMCQueueBase(size_t queueCapacity)
      : capacity_(queueCapacity),
        dstate_(0),
        dcapacity_(0),
        pushTicket_(0),
        popTicket_(0),
        pushSpinCutoff_(0),
        popSpinCutoff_(0) {
    if (queueCapacity == 0) {
      throw std::invalid_argument(
          "MPMCQueue with explicit capacity 0 is impossible"
          // Stride computation in derived classes would sigfpe if capacity is 0
      );
    }

    // ideally this would be a static assert, but g++ doesn't allow it
    assert(
        alignof(MPMCQueue<T, Atom>) >= hardware_destructive_interference_size);
    assert(
        static_cast<uint8_t*>(static_cast<void*>(&popTicket_)) -
            static_cast<uint8_t*>(static_cast<void*>(&pushTicket_)) >=
        static_cast<ptrdiff_t>(hardware_destructive_interference_size));
  }

  MPMCQueueBase() noexcept
      : capacity_(0),
        slots_(nullptr),
        stride_(0),
        dstate_(0),
        dcapacity_(0),
        pushTicket_(0),
        popTicket_(0),
        pushSpinCutoff_(0),
        popSpinCutoff_(0) {}

  MPMCQueueBase(MPMCQueueBase<Derived<T, Atom, Dynamic>>&& rhs) noexcept
      : capacity_(rhs.capacity_),
        slots_(rhs.slots_),
        stride_(rhs.stride_),
        dstate_(rhs.dstate_.load(std::memory_order_relaxed)),
        dcapacity_(rhs.dcapacity_.load(std::memory_order_relaxed)),
        pushTicket_(rhs.pushTicket_.load(std::memory_order_relaxed)),
        popTicket_(rhs.popTicket_.load(std::memory_order_relaxed)),
        pushSpinCutoff_(rhs.pushSpinCutoff_.load(std::memory_order_relaxed)),
        popSpinCutoff_(rhs.popSpinCutoff_.load(std::memory_order_relaxed)) {
    // relaxed ops are okay for the previous reads, since rhs queue can't
    // be in concurrent use

    // zero out rhs
    rhs.capacity_ = 0;
    rhs.slots_ = nullptr;
    rhs.stride_ = 0;
    rhs.dstate_.store(0, std::memory_order_relaxed);
    rhs.dcapacity_.store(0, std::memory_order_relaxed);
    rhs.pushTicket_.store(0, std::memory_order_relaxed);
    rhs.popTicket_.store(0, std::memory_order_relaxed);
    rhs.pushSpinCutoff_.store(0, std::memory_order_relaxed);
    rhs.popSpinCutoff_.store(0, std::memory_order_relaxed);
  }

  MPMCQueueBase<Derived<T, Atom, Dynamic>> const& operator=(
      MPMCQueueBase<Derived<T, Atom, Dynamic>>&& rhs) {
    if (this != &rhs) {
      this->~MPMCQueueBase();
      new (this) MPMCQueueBase(std::move(rhs));
    }
    return *this;
  }

  ~MPMCQueueBase() {
    delete[] slots_;
  }

  ssize_t size() const noexcept {
    // since both pushes and pops increase monotonically, we can get a
    // consistent snapshot either by bracketing a read of popTicket_ with
    // two reads of pushTicket_ that return the same value, or the other
    // way around.  We maximize our chances by alternately attempting
    // both bracketings.
    uint64_t pushes = pushTicket_.load(std::memory_order_acquire); // A
    uint64_t pops = popTicket_.load(std::memory_order_acquire); // B
    while (true) {
      uint64_t nextPushes = pushTicket_.load(std::memory_order_acquire); // C
      if (pushes == nextPushes) {
        // pushTicket_ didn't change from A (or the previous C) to C,
        // so we can linearize at B (or D)
        return ssize_t(pushes - pops);
      }
      pushes = nextPushes;
      uint64_t nextPops = popTicket_.load(std::memory_order_acquire); // D
      if (pops == nextPops) {
        // popTicket_ didn't chance from B (or the previous D), so we
        // can linearize at C
        return ssize_t(pushes - pops);
      }
      pops = nextPops;
    }
  }

  bool isEmpty() const noexcept {
    return size() <= 0;
  }

  bool isFull() const noexcept {
    // careful with signed -> unsigned promotion, since size can be negative
    return size() >= static_cast<ssize_t>(capacity_);
  }

  ssize_t sizeGuess() const noexcept {
    return writeCount() - readCount();
  }

  size_t capacity() const noexcept {
    return capacity_;
  }

  size_t allocatedCapacity() const noexcept {
    return capacity_;
  }

  uint64_t writeCount() const noexcept {
    return pushTicket_.load(std::memory_order_acquire);
  }

  uint64_t readCount() const noexcept {
    return popTicket_.load(std::memory_order_acquire);
  }

  template <typename... Args>
  void blockingWrite(Args&&... args) noexcept {
    enqueueWithTicketBase(
        pushTicket_++, slots_, capacity_, stride_, std::forward<Args>(args)...);
  }

  template <typename... Args>
  bool write(Args&&... args) noexcept {
    uint64_t ticket;
    Slot* slots;
    size_t cap;
    int stride;
    if (static_cast<Derived<T, Atom, Dynamic>*>(this)->tryObtainReadyPushTicket(
            ticket, slots, cap, stride)) {
      // we have pre-validated that the ticket won't block
      enqueueWithTicketBase(
          ticket, slots, cap, stride, std::forward<Args>(args)...);
      return true;
    } else {
      return false;
    }
  }

  template <class Clock, typename... Args>
  bool tryWriteUntil(
      const std::chrono::time_point<Clock>& when,
      Args&&... args) noexcept {
    uint64_t ticket;
    Slot* slots;
    size_t cap;
    int stride;
    if (tryObtainPromisedPushTicketUntil(ticket, slots, cap, stride, when)) {
      // we have pre-validated that the ticket won't block, or rather that
      // it won't block longer than it takes another thread to dequeue an
      // element from the slot it identifies.
      enqueueWithTicketBase(
          ticket, slots, cap, stride, std::forward<Args>(args)...);
      return true;
    } else {
      return false;
    }
  }

  template <typename... Args>
  bool writeIfNotFull(Args&&... args) noexcept {
    uint64_t ticket;
    Slot* slots;
    size_t cap;
    int stride;
    if (static_cast<Derived<T, Atom, Dynamic>*>(this)
            ->tryObtainPromisedPushTicket(ticket, slots, cap, stride)) {
      // some other thread is already dequeuing the slot into which we
      // are going to enqueue, but we might have to wait for them to finish
      enqueueWithTicketBase(
          ticket, slots, cap, stride, std::forward<Args>(args)...);
      return true;
    } else {
      return false;
    }
  }

  void blockingRead(T& elem) noexcept {
    uint64_t ticket;
    static_cast<Derived<T, Atom, Dynamic>*>(this)->blockingReadWithTicket(
        ticket, elem);
  }

  void blockingReadWithTicket(uint64_t& ticket, T& elem) noexcept {
    assert(capacity_ != 0);
    ticket = popTicket_++;
    dequeueWithTicketBase(ticket, slots_, capacity_, stride_, elem);
  }

  bool read(T& elem) noexcept {
    uint64_t ticket;
    return readAndGetTicket(ticket, elem);
  }

  bool readAndGetTicket(uint64_t& ticket, T& elem) noexcept {
    Slot* slots;
    size_t cap;
    int stride;
    if (static_cast<Derived<T, Atom, Dynamic>*>(this)->tryObtainReadyPopTicket(
            ticket, slots, cap, stride)) {
      // the ticket has been pre-validated to not block
      dequeueWithTicketBase(ticket, slots, cap, stride, elem);
      return true;
    } else {
      return false;
    }
  }

  template <class Clock, typename... Args>
  bool tryReadUntil(
      const std::chrono::time_point<Clock>& when,
      T& elem) noexcept {
    uint64_t ticket;
    Slot* slots;
    size_t cap;
    int stride;
    if (tryObtainPromisedPopTicketUntil(ticket, slots, cap, stride, when)) {
      // we have pre-validated that the ticket won't block, or rather that
      // it won't block longer than it takes another thread to enqueue an
      // element on the slot it identifies.
      dequeueWithTicketBase(ticket, slots, cap, stride, elem);
      return true;
    } else {
      return false;
    }
  }

  bool readIfNotEmpty(T& elem) noexcept {
    uint64_t ticket;
    Slot* slots;
    size_t cap;
    int stride;
    if (static_cast<Derived<T, Atom, Dynamic>*>(this)
            ->tryObtainPromisedPopTicket(ticket, slots, cap, stride)) {
      // the matching enqueue already has a ticket, but might not be done
      dequeueWithTicketBase(ticket, slots, cap, stride, elem);
      return true;
    } else {
      return false;
    }
  }

 protected:
  enum {
    kAdaptationFreq = 128,

    kSlotPadding =
        (hardware_destructive_interference_size - 1) / sizeof(Slot) + 1
  };

  alignas(hardware_destructive_interference_size) size_t capacity_;

  union {
    Slot* slots_;
    Atom<Slot*> dslots_;
  };

  union {
    int stride_;
    Atom<int> dstride_;
  };

  Atom<uint64_t> dstate_;
  Atom<size_t> dcapacity_;

  alignas(hardware_destructive_interference_size) Atom<uint64_t> pushTicket_;

  alignas(hardware_destructive_interference_size) Atom<uint64_t> popTicket_;

  alignas(
      hardware_destructive_interference_size) Atom<uint32_t> pushSpinCutoff_;

  alignas(hardware_destructive_interference_size) Atom<uint32_t> popSpinCutoff_;

  char pad_[hardware_destructive_interference_size - sizeof(Atom<uint32_t>)];

  static int computeStride(size_t capacity) noexcept {
    static const int smallPrimes[] = {2, 3, 5, 7, 11, 13, 17, 19, 23};

    int bestStride = 1;
    size_t bestSep = 1;
    for (int stride : smallPrimes) {
      if ((stride % capacity) == 0 || (capacity % stride) == 0) {
        continue;
      }
      size_t sep = stride % capacity;
      sep = std::min(sep, capacity - sep);
      if (sep > bestSep) {
        bestStride = stride;
        bestSep = sep;
      }
    }
    return bestStride;
  }

  size_t idx(uint64_t ticket, size_t cap, int stride) noexcept {
    return ((ticket * stride) % cap) + kSlotPadding;
  }

  uint32_t turn(uint64_t ticket, size_t cap) noexcept {
    assert(cap != 0);
    return uint32_t(ticket / cap);
  }

  bool tryObtainReadyPushTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    ticket = pushTicket_.load(std::memory_order_acquire); // A
    slots = slots_;
    cap = capacity_;
    stride = stride_;
    while (true) {
      if (!slots[idx(ticket, cap, stride)].mayEnqueue(turn(ticket, cap))) {
        // if we call enqueue(ticket, ...) on the SingleElementQueue
        // right now it would block, but this might no longer be the next
        // ticket.  We can increase the chance of tryEnqueue success under
        // contention (without blocking) by rechecking the ticket dispenser
        auto prev = ticket;
        ticket = pushTicket_.load(std::memory_order_acquire); // B
        if (prev == ticket) {
          // mayEnqueue was bracketed by two reads (A or prev B or prev
          // failing CAS to B), so we are definitely unable to enqueue
          return false;
        }
      } else {
        // we will bracket the mayEnqueue check with a read (A or prev B
        // or prev failing CAS) and the following CAS.  If the CAS fails
        // it will effect a load of pushTicket_
        if (pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
          return true;
        }
      }
    }
  }

  template <class Clock>
  bool tryObtainPromisedPushTicketUntil(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride,
      const std::chrono::time_point<Clock>& when) noexcept {
    bool deadlineReached = false;
    while (!deadlineReached) {
      if (static_cast<Derived<T, Atom, Dynamic>*>(this)
              ->tryObtainPromisedPushTicket(ticket, slots, cap, stride)) {
        return true;
      }
      // ticket is a blocking ticket until the preceding ticket has been
      // processed: wait until this ticket's turn arrives. We have not reserved
      // this ticket so we will have to re-attempt to get a non-blocking ticket
      // if we wake up before we time-out.
      deadlineReached =
          !slots[idx(ticket, cap, stride)].tryWaitForEnqueueTurnUntil(
              turn(ticket, cap),
              pushSpinCutoff_,
              (ticket % kAdaptationFreq) == 0,
              when);
    }
    return false;
  }

  bool tryObtainPromisedPushTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    auto numPushes = pushTicket_.load(std::memory_order_acquire); // A
    slots = slots_;
    cap = capacity_;
    stride = stride_;
    while (true) {
      ticket = numPushes;
      const auto numPops = popTicket_.load(std::memory_order_acquire); // B
      // n will be negative if pops are pending
      const int64_t n = int64_t(numPushes - numPops);
      if (n >= static_cast<ssize_t>(capacity_)) {
        // Full, linearize at B.  We don't need to recheck the read we
        // performed at A, because if numPushes was stale at B then the
        // real numPushes value is even worse
        return false;
      }
      if (pushTicket_.compare_exchange_strong(numPushes, numPushes + 1)) {
        return true;
      }
    }
  }

  bool tryObtainReadyPopTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    ticket = popTicket_.load(std::memory_order_acquire);
    slots = slots_;
    cap = capacity_;
    stride = stride_;
    while (true) {
      if (!slots[idx(ticket, cap, stride)].mayDequeue(turn(ticket, cap))) {
        auto prev = ticket;
        ticket = popTicket_.load(std::memory_order_acquire);
        if (prev == ticket) {
          return false;
        }
      } else {
        if (popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
          return true;
        }
      }
    }
  }

  template <class Clock>
  bool tryObtainPromisedPopTicketUntil(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride,
      const std::chrono::time_point<Clock>& when) noexcept {
    bool deadlineReached = false;
    while (!deadlineReached) {
      if (static_cast<Derived<T, Atom, Dynamic>*>(this)
              ->tryObtainPromisedPopTicket(ticket, slots, cap, stride)) {
        return true;
      }
      // ticket is a blocking ticket until the preceding ticket has been
      // processed: wait until this ticket's turn arrives. We have not reserved
      // this ticket so we will have to re-attempt to get a non-blocking ticket
      // if we wake up before we time-out.
      deadlineReached =
          !slots[idx(ticket, cap, stride)].tryWaitForDequeueTurnUntil(
              turn(ticket, cap),
              pushSpinCutoff_,
              (ticket % kAdaptationFreq) == 0,
              when);
    }
    return false;
  }

  bool tryObtainPromisedPopTicket(
      uint64_t& ticket,
      Slot*& slots,
      size_t& cap,
      int& stride) noexcept {
    auto numPops = popTicket_.load(std::memory_order_acquire); // A
    slots = slots_;
    cap = capacity_;
    stride = stride_;
    while (true) {
      ticket = numPops;
      const auto numPushes = pushTicket_.load(std::memory_order_acquire); // B
      if (numPops >= numPushes) {
        // Empty, or empty with pending pops.  Linearize at B.  We don't
        // need to recheck the read we performed at A, because if numPops
        // is stale then the fresh value is larger and the >= is still true
        return false;
      }
      if (popTicket_.compare_exchange_strong(numPops, numPops + 1)) {
        return true;
      }
    }
  }

  // Given a ticket, constructs an enqueued item using args
  template <typename... Args>
  void enqueueWithTicketBase(
      uint64_t ticket,
      Slot* slots,
      size_t cap,
      int stride,
      Args&&... args) noexcept {
    slots[idx(ticket, cap, stride)].enqueue(
        turn(ticket, cap),
        pushSpinCutoff_,
        (ticket % kAdaptationFreq) == 0,
        std::forward<Args>(args)...);
  }

  // To support tracking ticket numbers in MPMCPipelineStageImpl
  template <typename... Args>
  void enqueueWithTicket(uint64_t ticket, Args&&... args) noexcept {
    enqueueWithTicketBase(
        ticket, slots_, capacity_, stride_, std::forward<Args>(args)...);
  }

  // Given a ticket, dequeues the corresponding element
  void dequeueWithTicketBase(
      uint64_t ticket,
      Slot* slots,
      size_t cap,
      int stride,
      T& elem) noexcept {
    assert(cap != 0);
    slots[idx(ticket, cap, stride)].dequeue(
        turn(ticket, cap),
        popSpinCutoff_,
        (ticket % kAdaptationFreq) == 0,
        elem);
  }
};

template <typename T, template <typename> class Atom>
struct SingleElementQueue {
  ~SingleElementQueue() noexcept {
    if ((sequencer_.uncompletedTurnLSB() & 1) == 1) {
      // we are pending a dequeue, so we have a constructed item
      destroyContents();
    }
  }

  template <
      typename... Args,
      typename = typename std::enable_if<
          std::is_nothrow_constructible<T, Args...>::value>::type>
  void enqueue(
      const uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      Args&&... args) noexcept {
    sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
    new (&contents_) T(std::forward<Args>(args)...);
    sequencer_.completeTurn(turn * 2);
  }

  template <
      typename = typename std::enable_if<
          (folly::IsRelocatable<T>::value &&
           std::is_nothrow_constructible<T>::value) ||
          std::is_nothrow_constructible<T, T&&>::value>::type>
  void enqueue(
      const uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      T&& goner) noexcept {
    enqueueImpl(
        turn,
        spinCutoff,
        updateSpinCutoff,
        std::move(goner),
        typename std::conditional<
            std::is_nothrow_constructible<T, T&&>::value,
            ImplByMove,
            ImplByRelocation>::type());
  }

  template <class Clock>
  bool tryWaitForEnqueueTurnUntil(
      const uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      const std::chrono::time_point<Clock>& when) noexcept {
    return sequencer_.tryWaitForTurn(
               turn * 2, spinCutoff, updateSpinCutoff, &when) !=
        TurnSequencer<Atom>::TryWaitResult::TIMEDOUT;
  }

  bool mayEnqueue(const uint32_t turn) const noexcept {
    return sequencer_.isTurn(turn * 2);
  }

  void dequeue(
      uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      T& elem) noexcept {
    dequeueImpl(
        turn,
        spinCutoff,
        updateSpinCutoff,
        elem,
        typename std::conditional<
            folly::IsRelocatable<T>::value,
            ImplByRelocation,
            ImplByMove>::type());
  }

  template <class Clock>
  bool tryWaitForDequeueTurnUntil(
      const uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      const std::chrono::time_point<Clock>& when) noexcept {
    return sequencer_.tryWaitForTurn(
               turn * 2 + 1, spinCutoff, updateSpinCutoff, &when) !=
        TurnSequencer<Atom>::TryWaitResult::TIMEDOUT;
  }

  bool mayDequeue(const uint32_t turn) const noexcept {
    return sequencer_.isTurn(turn * 2 + 1);
  }

 private:
  typename std::aligned_storage<sizeof(T), alignof(T)>::type contents_;

  TurnSequencer<Atom> sequencer_;

  T* ptr() noexcept {
    return static_cast<T*>(static_cast<void*>(&contents_));
  }

  void destroyContents() noexcept {
    try {
      ptr()->~T();
    } catch (...) {
      // g++ doesn't seem to have std::is_nothrow_destructible yet
    }
#ifndef NDEBUG
    memset(&contents_, 'Q', sizeof(T));
#endif
  }

  struct ImplByRelocation {};
  struct ImplByMove {};

  void enqueueImpl(
      const uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      T&& goner,
      ImplByMove) noexcept {
    sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
    new (&contents_) T(std::move(goner));
    sequencer_.completeTurn(turn * 2);
  }

  void enqueueImpl(
      const uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      T&& goner,
      ImplByRelocation) noexcept {
    sequencer_.waitForTurn(turn * 2, spinCutoff, updateSpinCutoff);
    memcpy(&contents_, &goner, sizeof(T));
    sequencer_.completeTurn(turn * 2);
    new (&goner) T();
  }

  void dequeueImpl(
      uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      T& elem,
      ImplByRelocation) noexcept {
    try {
      elem.~T();
    } catch (...) {
      // unlikely, but if we don't complete our turn the queue will die
    }
    sequencer_.waitForTurn(turn * 2 + 1, spinCutoff, updateSpinCutoff);
    memcpy(&elem, &contents_, sizeof(T));
    sequencer_.completeTurn(turn * 2 + 1);
  }

  void dequeueImpl(
      uint32_t turn,
      Atom<uint32_t>& spinCutoff,
      const bool updateSpinCutoff,
      T& elem,
      ImplByMove) noexcept {
    sequencer_.waitForTurn(turn * 2 + 1, spinCutoff, updateSpinCutoff);
    elem = std::move(*ptr());
    destroyContents();
    sequencer_.completeTurn(turn * 2 + 1);
  }
};

} // namespace detail

} // namespace folly