folly::MPMCQueue< T, Atom, true > Class Template Reference

#include <MPMCQueue.h>

Inheritance diagram for folly::MPMCQueue< T, Atom, true >:

Classes
struct	ClosedArray

Public Member Functions
	MPMCQueue (size_t queueCapacity)
	MPMCQueue (size_t queueCapacity, size_t minCapacity, size_t expansionMultiplier)
	MPMCQueue () noexcept
	MPMCQueue (MPMCQueue< T, Atom, true > &&rhs) noexcept
MPMCQueue< T, Atom, true > const &	operator= (MPMCQueue< T, Atom, true > &&rhs)
	~MPMCQueue ()
size_t	allocatedCapacity () const noexcept
template<typename... Args>
void	blockingWrite (Args &&...args) noexcept
void	blockingReadWithTicket (uint64_t &ticket, T &elem) noexcept

Private Types
enum	{ kSeqlockBits = 6, kDefaultMinDynamicCapacity = 10, kDefaultExpansionMultiplier = 10 }
using	Slot = detail::SingleElementQueue< T, Atom >

Private Member Functions
void	initQueue (const size_t cap, const size_t mult)
bool	tryObtainReadyPushTicket (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride) noexcept
bool	tryObtainPromisedPushTicket (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride) noexcept
bool	tryObtainReadyPopTicket (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride) noexcept
bool	tryObtainPromisedPopTicket (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride) noexcept
template<typename... Args>
void	enqueueWithTicket (const uint64_t ticket, Args &&...args) noexcept
	Enqueues an element with a specific ticket number. More...
uint64_t	getOffset (const uint64_t state) const noexcept
int	getNumClosed (const uint64_t state) const noexcept
bool	tryExpand (const uint64_t state, const size_t cap) noexcept
bool	trySeqlockReadSection (uint64_t &state, Slot *&slots, size_t &cap, int &stride) noexcept
	Seqlock read-only section. More...
bool	maybeUpdateFromClosed (const uint64_t state, const uint64_t ticket, uint64_t &offset, Slot *&slots, size_t &cap, int &stride) noexcept

Private Attributes
size_t	dmult_
ClosedArray *	closed_

Friends
class	detail::MPMCQueueBase< MPMCQueue< T, Atom, true > >

Detailed Description

template<typename T, template< typename > class Atom>
class folly::MPMCQueue< T, Atom, true >

The dynamic version of MPMCQueue allows dynamic expansion of queue capacity, such that a queue may start with a smaller capacity than specified and expand only if needed. Users may optionally specify the initial capacity and the expansion multiplier.

The design uses a seqlock to enforce mutual exclusion among expansion attempts. Regular operations read up-to-date queue information (slots array, capacity, stride) inside read-only seqlock sections, which are unimpeded when no expansion is in progress.

An expansion computes a new capacity, allocates a new slots array, and updates stride. No information needs to be copied from the current slots array to the new one. When this happens, new slots will not have sequence numbers that match ticket numbers. The expansion needs to compute a ticket offset such that operations that use new arrays can adjust the calculations of slot indexes and sequence numbers that take into account that the new slots start with sequence numbers of zero. The current ticket offset is packed with the seqlock in an atomic 64-bit integer. The initial offset is zero.

Lagging write and read operations with tickets lower than the ticket offset of the current slots array (i.e., the minimum ticket number that can be served by the current array) must use earlier closed arrays instead of the current one. Information about closed slots arrays (array address, capacity, stride, and offset) is maintained in a logarithmic-sized structure. Each entry in that structure never needs to be changed once set. The number of closed arrays is half the value of the seqlock (when unlocked).

The acquisition of the seqlock to perform an expansion does not prevent the issuing of new push and pop tickets concurrently. The expansion must set the new ticket offset to a value that couldn't have been issued to an operation that has already gone through a seqlock read-only section (and hence obtained information for older closed arrays).

Note that the total queue capacity can temporarily exceed the specified capacity when there are lagging consumers that haven't yet consumed all the elements in closed arrays. Users should not rely on the capacity of dynamic queues for synchronization, e.g., they should not expect that a thread will definitely block on a call to blockingWrite() when the queue size is known to be equal to its capacity.

Note that some writeIfNotFull() and tryWriteUntil() operations may fail even if the size of the queue is less than its maximum capacity and despite the success of expansion, if the operation happens to acquire a ticket that belongs to a closed array. This is a transient condition. Typically, one or two ticket values may be subject to such condition per expansion.

The dynamic version is a partial specialization of MPMCQueue with Dynamic == true

Definition at line 176 of file MPMCQueue.h.

Member Typedef Documentation

template<typename T , template< typename > class Atom>

using folly::MPMCQueue< T, Atom, true >::Slot = detail::SingleElementQueue<T, Atom>

private

Definition at line 179 of file MPMCQueue.h.

Member Enumeration Documentation

template<typename T , template< typename > class Atom>

anonymous enum

private

Enumerator
kSeqlockBits
kDefaultMinDynamicCapacity
kDefaultExpansionMultiplier

Definition at line 337 of file MPMCQueue.h.

       {
    kSeqlockBits = 6,
    kDefaultMinDynamicCapacity = 10,
    kDefaultExpansionMultiplier = 10,
  };

Constructor & Destructor Documentation

template<typename T , template< typename > class Atom>

folly::MPMCQueue< T, Atom, true >::MPMCQueue ( size_t  queueCapacity )

inlineexplicit

Definition at line 189 of file MPMCQueue.h.

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

template<typename T , template< typename > class Atom>

folly::MPMCQueue< T, Atom, true >::MPMCQueue ( size_t  queueCapacity, size_t  minCapacity, size_t  expansionMultiplier  )

inlineexplicit

Definition at line 195 of file MPMCQueue.h.

      : 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);
  }

template<typename T , template< typename > class Atom>

folly::MPMCQueue< T, Atom, true >::MPMCQueue ( )

inlinenoexcept

Definition at line 206 of file MPMCQueue.h.

                       {
    dmult_ = 0;
    closed_ = nullptr;
  }

template<typename T , template< typename > class Atom>

folly::MPMCQueue< T, Atom, true >::MPMCQueue ( MPMCQueue< T, Atom, true > &&  rhs )

inlinenoexcept

Definition at line 211 of file MPMCQueue.h.

References folly::detail::rhs.

                                                     {
    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;
  }

template<typename T , template< typename > class Atom>

folly::MPMCQueue< T, Atom, true >::~MPMCQueue ( )

inline

Definition at line 258 of file MPMCQueue.h.

References i.

               {
    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;
  }

Member Function Documentation

template<typename T , template< typename > class Atom>

size_t folly::MPMCQueue< T, Atom, true >::allocatedCapacity ( ) const

inlinenoexcept

Definition at line 274 of file MPMCQueue.h.

References testing::Args().

                                            {
    return this->dcapacity_.load(std::memory_order_relaxed);
  }

template<typename T , template< typename > class Atom>

void folly::MPMCQueue< T, Atom, true >::blockingReadWithTicket ( uint64_t &  ticket, T &  elem  )

inlinenoexcept

Definition at line 320 of file MPMCQueue.h.

References folly::asm_volatile_pause(), folly::gen::stride(), ticket, and uint64_t.

                                                                  {
    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);
  }

template<typename T , template< typename > class Atom>

template<typename... Args>

void folly::MPMCQueue< T, Atom, true >::blockingWrite ( Args &&...  args )

inlinenoexcept

Definition at line 279 of file MPMCQueue.h.

References folly::asm_volatile_pause(), folly::gen::stride(), ticket, and uint64_t.

                                              {
    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)...);
  }

template<typename T , template< typename > class Atom>

template<typename... Args>

void folly::MPMCQueue< T, Atom, true >::enqueueWithTicket ( const uint64_t  ticket, Args &&...  args  )

inlineprivatenoexcept

Enqueues an element with a specific ticket number.

Definition at line 511 of file MPMCQueue.h.

References folly::gen::stride(), and uint64_t.

                                                                         {
    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)...);
  }

template<typename T , template< typename > class Atom>

int folly::MPMCQueue< T, Atom, true >::getNumClosed ( const uint64_t  state ) const

inlineprivatenoexcept

Definition at line 533 of file MPMCQueue.h.

                                                        {
    return (state & ((1 << kSeqlockBits) - 1)) >> 1;
  }

template<typename T , template< typename > class Atom>

uint64_t folly::MPMCQueue< T, Atom, true >::getOffset ( const uint64_t  state ) const

inlineprivatenoexcept

Definition at line 529 of file MPMCQueue.h.

                                                          {
    return state >> kSeqlockBits;
  }

template<typename T , template< typename > class Atom>

void folly::MPMCQueue< T, Atom, true >::initQueue ( const size_t  cap, const size_t  mult  )

inlineprivate

Definition at line 348 of file MPMCQueue.h.

                                                      {
    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;
  }

template<typename T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::maybeUpdateFromClosed ( const uint64_t  state, const uint64_t  ticket, uint64_t &  offset, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprivatenoexcept

If there was an expansion after ticket was issued, update local variables of the lagging operation using the most recent closed array with offset <= ticket and return true. Otherwise, return false;

Definition at line 608 of file MPMCQueue.h.

References Atom, i, folly::gen::stride(), and ticket.

                            {
    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;
  }

template<typename T , template< typename > class Atom>

MPMCQueue<T, Atom, true> const& folly::MPMCQueue< T, Atom, true >::operator= ( MPMCQueue< T, Atom, true > &&  rhs )

inline

Definition at line 250 of file MPMCQueue.h.

References folly::gen::move, and folly::detail::rhs.

                                                                            {
    if (this != &rhs) {
      this->~MPMCQueue();
      new (this) MPMCQueue(std::move(rhs));
    }
    return *this;
  }

template<typename T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::tryExpand ( const uint64_t  state, const size_t  cap  )

inlineprivatenoexcept

Try to expand the queue. Returns true if this expansion was successful or a concurent expansion is in progress. Returns false if the queue has reached its maximum capacity or allocation has failed.

Definition at line 541 of file MPMCQueue.h.

References max, min, ticket, and uint64_t.

                                                                  {
    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;
    }
  }

template<typename T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::tryObtainPromisedPopTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprivatenoexcept

Definition at line 479 of file MPMCQueue.h.

References testing::Args(), folly::asm_volatile_pause(), folly::gen::stride(), ticket, and uint64_t.

                            {
    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 T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::tryObtainPromisedPushTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprivatenoexcept

Definition at line 408 of file MPMCQueue.h.

References folly::asm_volatile_pause(), int64_t, folly::gen::stride(), ticket, and uint64_t.

                            {
    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);
  }

template<typename T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::tryObtainReadyPopTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprivatenoexcept

Definition at line 448 of file MPMCQueue.h.

References folly::asm_volatile_pause(), folly::gen::stride(), ticket, and uint64_t.

                            {
    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);
  }

template<typename T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::tryObtainReadyPushTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprivatenoexcept

Definition at line 362 of file MPMCQueue.h.

References folly::asm_volatile_pause(), folly::gen::stride(), ticket, and uint64_t.

                            {
    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);
  }

template<typename T , template< typename > class Atom>

bool folly::MPMCQueue< T, Atom, true >::trySeqlockReadSection ( uint64_t &  state, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprivatenoexcept

Seqlock read-only section.

Definition at line 586 of file MPMCQueue.h.

References folly::gen::stride().

                            {
    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));
  }

Friends And Related Function Documentation

template<typename T , template< typename > class Atom>

friend class detail::MPMCQueueBase< MPMCQueue< T, Atom, true > >

friend

Definition at line 178 of file MPMCQueue.h.

Member Data Documentation

template<typename T , template< typename > class Atom>

ClosedArray* folly::MPMCQueue< T, Atom, true >::closed_

private

Definition at line 346 of file MPMCQueue.h.

template<typename T , template< typename > class Atom>

size_t folly::MPMCQueue< T, Atom, true >::dmult_

private

Definition at line 343 of file MPMCQueue.h.

---

The documentation for this class was generated from the following file:

• proxygen/folly/folly/MPMCQueue.h