folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > > Class Template Reference

CRTP specialization of MPMCQueueBase. More...

#include <MPMCQueue.h>

Inheritance diagram for folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >:

Public Types
typedef T	value_type
using	Slot = detail::SingleElementQueue< T, Atom >

Public Member Functions
	MPMCQueueBase (size_t queueCapacity)
	MPMCQueueBase () noexcept
	MPMCQueueBase (MPMCQueueBase< Derived< T, Atom, Dynamic >> &&rhs) noexcept
MPMCQueueBase< Derived< T, Atom, Dynamic > > const &	operator= (MPMCQueueBase< Derived< T, Atom, Dynamic >> &&rhs)
	~MPMCQueueBase ()
ssize_t	size () const noexcept
bool	isEmpty () const noexcept
	Returns true if there are no items available for dequeue. More...
bool	isFull () const noexcept
	Returns true if there is currently no empty space to enqueue. More...
ssize_t	sizeGuess () const noexcept
size_t	capacity () const noexcept
	Doesn't change. More...
size_t	allocatedCapacity () const noexcept
	Doesn't change for non-dynamic. More...
uint64_t	writeCount () const noexcept
uint64_t	readCount () const noexcept
template<typename... Args>
void	blockingWrite (Args &&...args) noexcept
template<typename... Args>
bool	write (Args &&...args) noexcept
template<class Clock , typename... Args>
bool	tryWriteUntil (const std::chrono::time_point< Clock > &when, Args &&...args) noexcept
template<typename... Args>
bool	writeIfNotFull (Args &&...args) noexcept
void	blockingRead (T &elem) noexcept
void	blockingReadWithTicket (uint64_t &ticket, T &elem) noexcept
	Same as blockingRead() but also records the ticket nunmer. More...
bool	read (T &elem) noexcept
bool	readAndGetTicket (uint64_t &ticket, T &elem) noexcept
	Same as read() but also records the ticket nunmer. More...
template<class Clock , typename... Args>
bool	tryReadUntil (const std::chrono::time_point< Clock > &when, T &elem) noexcept
bool	readIfNotEmpty (T &elem) noexcept

Protected Types
enum	{ kAdaptationFreq = 128, kSlotPadding }

Protected Member Functions
size_t	idx (uint64_t ticket, size_t cap, int stride) noexcept
uint32_t	turn (uint64_t ticket, size_t cap) noexcept
bool	tryObtainReadyPushTicket (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride) noexcept
template<class Clock >
bool	tryObtainPromisedPushTicketUntil (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride, const std::chrono::time_point< Clock > &when) 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
template<class Clock >
bool	tryObtainPromisedPopTicketUntil (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride, const std::chrono::time_point< Clock > &when) noexcept
bool	tryObtainPromisedPopTicket (uint64_t &ticket, Slot *&slots, size_t &cap, int &stride) noexcept
template<typename... Args>
void	enqueueWithTicketBase (uint64_t ticket, Slot *slots, size_t cap, int stride, Args &&...args) noexcept
template<typename... Args>
void	enqueueWithTicket (uint64_t ticket, Args &&...args) noexcept
void	dequeueWithTicketBase (uint64_t ticket, Slot *slots, size_t cap, int stride, T &elem) noexcept

Static Protected Member Functions
static int	computeStride (size_t capacity) noexcept

Protected Attributes
size_t	capacity_
	The maximum number of items in the queue at once. More...
union folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >:: { ... }
	Anonymous union for use when Dynamic = false and true, respectively. More...
union folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >:: { ... }
	Anonymous union for use when Dynamic = false and true, respectively. More...
Atom< uint64_t >	dstate_
Atom< size_t >	dcapacity_
	Dynamic capacity. More...
Atom< uint64_t >	pushTicket_
	Enqueuers get tickets from here. More...
Atom< uint64_t >	popTicket_
	Dequeuers get tickets from here. More...
Atom< uint32_t >	pushSpinCutoff_
Atom< uint32_t >	popSpinCutoff_
	The adaptive spin cutoff when the queue is empty on dequeue. More...
char	pad_ [hardware_destructive_interference_size-sizeof(Atom< uint32_t >)]

Detailed Description

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T, template< typename > class Atom, bool Dynamic>
class folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >

CRTP specialization of MPMCQueueBase.

Definition at line 643 of file MPMCQueue.h.

Member Typedef Documentation

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

using folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::Slot = detail::SingleElementQueue<T, Atom>

Definition at line 657 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

typedef T folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::value_type

Definition at line 652 of file MPMCQueue.h.

Member Enumeration Documentation

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

anonymous enum

protected

Enumerator
kAdaptationFreq	Once every kAdaptationFreq we will spin longer, to try to estimate the proper spin backoff
kSlotPadding	To avoid false sharing in slots_ with neighboring memory allocations, we pad it with this many SingleElementQueue-s at each end

Definition at line 991 of file MPMCQueue.h.

       {
    kAdaptationFreq = 128,

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

Constructor & Destructor Documentation

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::MPMCQueueBase ( size_t  queueCapacity )

inlineexplicit

Definition at line 659 of file MPMCQueue.h.

References folly::hardware_destructive_interference_size.

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::MPMCQueueBase ( )

inlinenoexcept

A default-constructed queue is useful because a usable (non-zero capacity) queue can be moved onto it or swapped with it

Definition at line 685 of file MPMCQueue.h.

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::MPMCQueueBase ( MPMCQueueBase< Derived< T, Atom, Dynamic >> &&  rhs )

inlinenoexcept

IMPORTANT: The move constructor is here to make it easier to perform the initialization phase, it is not safe to use when there are any concurrent accesses (this is not checked).

Definition at line 699 of file MPMCQueue.h.

References folly::detail::rhs.

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::~MPMCQueueBase ( )

inline

MPMCQueue can only be safely destroyed when there are no pending enqueuers or dequeuers (this is not checked).

Definition at line 738 of file MPMCQueue.h.

                   {
    delete[] slots_;
  }

Member Function Documentation

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

size_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::allocatedCapacity ( ) const

inlinenoexcept

Doesn't change for non-dynamic.

Definition at line 804 of file MPMCQueue.h.

                                            {
    return capacity_;
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

void folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::blockingRead ( T &  elem )

inlinenoexcept

Moves a dequeued element onto elem, blocking until an element is available

Definition at line 916 of file MPMCQueue.h.

References ticket, and uint64_t.

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

void folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::blockingReadWithTicket ( uint64_t &  ticket, T &  elem  )

inlinenoexcept

Same as blockingRead() but also records the ticket nunmer.

Definition at line 923 of file MPMCQueue.h.

References ticket.

                                                                  {
    assert(capacity_ != 0);
    ticket = popTicket_++;
    dequeueWithTicketBase(ticket, slots_, capacity_, stride_, elem);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<typename... Args>

void folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::blockingWrite ( Args &&...  args )

inlinenoexcept

Enqueues a T constructed from args, blocking until space is available. Note that this method signature allows enqueue via move, if args is a T rvalue, via copy, if args is a T lvalue, or via emplacement if args is an initializer list that can be passed to a T constructor.

Definition at line 828 of file MPMCQueue.h.

References testing::Args().

                                              {
    enqueueWithTicketBase(
        pushTicket_++, slots_, capacity_, stride_, std::forward<Args>(args)...);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

size_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::capacity ( ) const

inlinenoexcept

Doesn't change.

Definition at line 799 of file MPMCQueue.h.

                                   {
    return capacity_;
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

static int folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::computeStride ( size_t  capacity )

inlinestaticprotectednoexcept

We assign tickets in increasing order, but we don't want to access neighboring elements of slots_ because that will lead to false sharing (multiple cores accessing the same cache line even though they aren't accessing the same bytes in that cache line). To avoid this we advance by stride slots per ticket.

We need gcd(capacity, stride) to be 1 so that we will use all of the slots. We ensure this by only considering prime strides, which either have no common divisors with capacity or else have a zero remainder after dividing by capacity. That is sufficient to guarantee correctness, but we also want to actually spread the accesses away from each other to avoid false sharing (consider a stride of 7 with a capacity of 8). To that end we try a few taking care to observe that advancing by -1 is as bad as advancing by 1 when in comes to false sharing.

The simple way to avoid false sharing would be to pad each SingleElementQueue, but since we have capacity_ of them that could waste a lot of space.

Definition at line 1074 of file MPMCQueue.h.

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

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

void folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::dequeueWithTicketBase ( uint64_t  ticket, Slot *  slots, size_t  cap, int  stride, T &  elem  )

inlineprotectednoexcept

Definition at line 1319 of file MPMCQueue.h.

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

                        {
    assert(cap != 0);
    slots[idx(ticket, cap, stride)].dequeue(
        turn(ticket, cap),
        popSpinCutoff_,
        (ticket % kAdaptationFreq) == 0,
        elem);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<typename... Args>

void folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::enqueueWithTicket ( uint64_t  ticket, Args &&...  args  )

inlineprotectednoexcept

Definition at line 1313 of file MPMCQueue.h.

                                                                   {
    enqueueWithTicketBase(
        ticket, slots_, capacity_, stride_, std::forward<Args>(args)...);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<typename... Args>

void folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::enqueueWithTicketBase ( uint64_t  ticket, Slot *  slots, size_t  cap, int  stride, Args &&...  args  )

inlineprotectednoexcept

Definition at line 1298 of file MPMCQueue.h.

References testing::Args(), and folly::detail::SingleElementQueue< T, Atom >::enqueue().

                               {
    slots[idx(ticket, cap, stride)].enqueue(
        turn(ticket, cap),
        pushSpinCutoff_,
        (ticket % kAdaptationFreq) == 0,
        std::forward<Args>(args)...);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

size_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::idx ( uint64_t  ticket, size_t  cap, int  stride  )

inlineprotectednoexcept

Returns the index into slots_ that should be used when enqueuing or dequeuing with the specified ticket

Definition at line 1095 of file MPMCQueue.h.

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

                                                               {
    return ((ticket * stride) % cap) + kSlotPadding;
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::isEmpty ( ) const

inlinenoexcept

Returns true if there are no items available for dequeue.

Definition at line 778 of file MPMCQueue.h.

References folly::size().

                                {
    return size() <= 0;
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::isFull ( ) const

inlinenoexcept

Returns true if there is currently no empty space to enqueue.

Definition at line 783 of file MPMCQueue.h.

References folly::size().

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

MPMCQueueBase<Derived<T, Atom, Dynamic> > const& folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::operator= ( MPMCQueueBase< Derived< T, Atom, Dynamic >> &&  rhs )

inline

IMPORTANT: The move operator is here to make it easier to perform the initialization phase, it is not safe to use when there are any concurrent accesses (this is not checked).

Definition at line 727 of file MPMCQueue.h.

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

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::read ( T &  elem )

inlinenoexcept

If an item can be dequeued with no blocking, does so and returns true, otherwise returns false.

Definition at line 931 of file MPMCQueue.h.

References ticket, and uint64_t.

                              {
    uint64_t ticket;
    return readAndGetTicket(ticket, elem);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::readAndGetTicket ( uint64_t &  ticket, T &  elem  )

inlinenoexcept

Same as read() but also records the ticket nunmer.

Definition at line 937 of file MPMCQueue.h.

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

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

uint64_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::readCount ( ) const

inlinenoexcept

Returns the total number of calls to blockingRead or successful calls to read, including those blockingRead calls that are currently blocking

Definition at line 818 of file MPMCQueue.h.

References testing::Args().

                                      {
    return popTicket_.load(std::memory_order_acquire);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::readIfNotEmpty ( T &  elem )

inlinenoexcept

If the queue is not empty, dequeues and returns true, otherwise returns false. If the matching write is still in progress then this method may block waiting for it. If you don't rely on being able to dequeue (such as by counting completed write) then you should prefer read.

Definition at line 975 of file MPMCQueue.h.

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

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

ssize_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::size ( ) const

inlinenoexcept

Returns the number of writes (including threads that are blocked waiting to write) minus the number of reads (including threads that are blocked waiting to read). So effectively, it becomes: elements in queue + pending(calls to write) - pending(calls to read). If nothing is pending, then the method returns the actual number of elements in the queue. The returned value can be negative if there are no writers and the queue is empty, but there is one reader that is blocked waiting to read (in which case, the returned size will be -1).

Definition at line 751 of file MPMCQueue.h.

References uint64_t.

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

ssize_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::sizeGuess ( ) const

inlinenoexcept

Returns is a guess at size() for contexts that don't need a precise value, such as stats. More specifically, it returns the number of writes minus the number of reads, but after reading the number of writes, more writers could have came before the number of reads was sampled, and this method doesn't protect against such case. The returned value can be negative.

Definition at line 794 of file MPMCQueue.h.

                                     {
    return writeCount() - readCount();
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryObtainPromisedPopTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprotectednoexcept

Similar to tryObtainReadyPopTicket, but returns a pop ticket whose corresponding push ticket has already been handed out, rather than returning one whose corresponding push ticket has already been completed. This means that there is a possibility that the caller will block when using the ticket, but it allows the user to rely on the fact that if enqueue has succeeded, tryObtainPromisedPopTicket will return true. The "try" part of this is that we won't have to block waiting for someone to call enqueue, although we might have to block waiting for them to finish executing code inside the MPMCQueue itself.

Definition at line 1272 of file MPMCQueue.h.

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

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<class Clock >

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryObtainPromisedPopTicketUntil ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride, const std::chrono::time_point< Clock > &  when  )

inlineprotectednoexcept

Tries until when to obtain a pop ticket for which SingleElementQueue::dequeue won't block. Returns true on success, false on failure. ticket is filled on success AND failure.

Definition at line 1236 of file MPMCQueue.h.

References folly::gen::stride(), ticket, and folly::when().

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryObtainPromisedPushTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprotectednoexcept

Tries to obtain a push ticket which can be satisfied if all in-progress pops complete. This function does not block, but blocking may be required when using the returned ticket if some other thread's pop is still in progress (ticket has been granted but pop has not yet completed).

Definition at line 1178 of file MPMCQueue.h.

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

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<class Clock >

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryObtainPromisedPushTicketUntil ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride, const std::chrono::time_point< Clock > &  when  )

inlineprotectednoexcept

Tries until when to obtain a push ticket for which SingleElementQueue::enqueue won't block. Returns true on success, false on failure. ticket is filled on success AND failure.

Definition at line 1147 of file MPMCQueue.h.

References folly::gen::stride(), ticket, and folly::when().

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryObtainReadyPopTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprotectednoexcept

Tries to obtain a pop ticket for which SingleElementQueue::dequeue won't block. Returns true on immediate success, false on immediate failure.

Definition at line 1207 of file MPMCQueue.h.

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

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

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryObtainReadyPushTicket ( uint64_t &  ticket, Slot *&  slots, size_t &  cap, int &  stride  )

inlineprotectednoexcept

Tries to obtain a push ticket for which SingleElementQueue::enqueue won't block. Returns true on immediate success, false on immediate failure.

Definition at line 1109 of file MPMCQueue.h.

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

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

template<class Clock , typename... Args>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryReadUntil ( const std::chrono::time_point< Clock > &  when, T &  elem  )

inlinenoexcept

Definition at line 952 of file MPMCQueue.h.

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

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

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<class Clock , typename... Args>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::tryWriteUntil ( const std::chrono::time_point< Clock > &  when, Args &&...  args  )

inlinenoexcept

Definition at line 864 of file MPMCQueue.h.

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

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

uint32_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::turn ( uint64_t  ticket, size_t  cap  )

inlineprotectednoexcept

Maps an enqueue or dequeue ticket to the turn should be used at the corresponding SingleElementQueue

Definition at line 1101 of file MPMCQueue.h.

References ticket, and uint32_t.

                                                      {
    assert(cap != 0);
    return uint32_t(ticket / cap);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<typename... Args>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::write ( Args &&...  args )

inlinenoexcept

If an item can be enqueued with no blocking, does so and returns true, otherwise returns false. This method is similar to writeIfNotFull, but if you don't have a specific need for that method you should use this one.

One of the common usages of this method is to enqueue via the move constructor, something like q.write(std::move(x)). If write returns false because the queue is full then x has not actually been consumed, which looks strange. To understand why it is actually okay to use x afterward, remember that std::move is just a typecast that provides an rvalue reference that enables use of a move constructor or operator. std::move doesn't actually move anything. It could more accurately be called std::rvalue_cast or std::move_permission.

Definition at line 847 of file MPMCQueue.h.

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

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

uint64_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::writeCount ( ) const

inlinenoexcept

Returns the total number of calls to blockingWrite or successful calls to write, including those blockingWrite calls that are currently blocking

Definition at line 811 of file MPMCQueue.h.

                                       {
    return pushTicket_.load(std::memory_order_acquire);
  }

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

template<typename... Args>

bool folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::writeIfNotFull ( Args &&...  args )

inlinenoexcept

If the queue is not full, enqueues and returns true, otherwise returns false. Unlike write this method can be blocked by another thread, specifically a read that has linearized (been assigned a ticket) but not yet completed. If you don't really need this function you should probably use write.

MPMCQueue isn't lock-free, so just because a read operation has linearized (and isFull is false) doesn't mean that space has been made available for another write. In this situation write will return false, but writeIfNotFull will wait for the dequeue to finish. This method is required if you are composing queues and managing your own wakeup, because it guarantees that after every successful write a readIfNotEmpty will succeed.

Definition at line 897 of file MPMCQueue.h.

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

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

Member Data Documentation

union { ... }

Anonymous union for use when Dynamic = false and true, respectively.

union { ... }

Anonymous union for use when Dynamic = false and true, respectively.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

size_t folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::capacity_

protected

The maximum number of items in the queue at once.

Definition at line 1004 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<size_t> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::dcapacity_

protected

Dynamic capacity.

Definition at line 1034 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<Slot*> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::dslots_

Current dynamic slots array of dcapacity_ SingleElementQueue-s.

Definition at line 1013 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<uint64_t> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::dstate_

protected

The following two memebers are used by dynamic MPMCQueue. Ideally they should be in MPMCQueue<T,Atom,true>, but we get better cache locality if they are in the same cache line as dslots_ and dstride_.

Dynamic state. A packed seqlock and ticket offset

Definition at line 1032 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<int> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::dstride_

Current stride.

Definition at line 1023 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

char folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::pad_[hardware_destructive_interference_size-sizeof(Atom< uint32_t >)]

protected

Alignment doesn't prevent false sharing at the end of the struct, so fill out the last cache line

Definition at line 1053 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<uint32_t> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::popSpinCutoff_

protected

The adaptive spin cutoff when the queue is empty on dequeue.

Definition at line 1049 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<uint64_t> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::popTicket_

protected

Dequeuers get tickets from here.

Definition at line 1040 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<uint32_t> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::pushSpinCutoff_

protected

This is how many times we will spin before using FUTEX_WAIT when the queue is full on enqueue, adaptively computed by occasionally spinning for longer and smoothing with an exponential moving average

Definition at line 1046 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Atom<uint64_t> folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::pushTicket_

protected

Enqueuers get tickets from here.

Definition at line 1037 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

Slot* folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::slots_

An array of capacity_ SingleElementQueue-s, each of which holds either 0 or 1 item. We over-allocate by 2 * kSlotPadding and don't touch the slots at either end, to avoid false sharing

Definition at line 1011 of file MPMCQueue.h.

template<template< typename T, template< typename > class Atom, bool Dynamic > class Derived, typename T , template< typename > class Atom, bool Dynamic>

int folly::detail::MPMCQueueBase< Derived< T, Atom, Dynamic > >::stride_

The number of slots_ indices that we advance for each ticket, to avoid false sharing. Ideally slots_[i] and slots_[i + stride_] aren't on the same cache line

Definition at line 1021 of file MPMCQueue.h.

---

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

• proxygen/folly/folly/MPMCQueue.h