Job::SPMCDeque< T > Class Template Reference

#include <job_queue.hpp>

Public Types
using	size_type = std::int64_t
using	atomic_size_type = std::atomic< size_type >

Public Member Functions
	SPMCDeque ()=default
	~SPMCDeque ()=default
void	Initialize (AtomicT *const memory_backing, const size_type capacity) noexcept
SPMCDequeStatus	Push (const T &value)
SPMCDequeStatus	Pop (T *const out_value)
SPMCDequeStatus	Steal (T *const out_value)

Private Types
using	AtomicT = std::atomic< T >

Private Member Functions
AtomicT *	ElementAt (const size_type index) const noexcept

Private Attributes
atomic_size_type	m_ProducerIndex
atomic_size_type	m_ConsumerIndex
unsigned char	m_Padding0 [k_FalseSharingPadSize - sizeof(m_ProducerIndex) - sizeof(m_ConsumerIndex)]
AtomicT *	m_Data
size_type	m_Capacity
size_type	m_CapacityMask

Detailed Description

template<typename T>
class Job::SPMCDeque< T >

Definition at line 240 of file job_queue.hpp.

Member Typedef Documentation

AtomicT

template<typename T >

using Job::SPMCDeque< T >::AtomicT = std::atomic<T>

private

Definition at line 243 of file job_queue.hpp.

size_type

template<typename T >

using Job::SPMCDeque< T >::size_type = std::int64_t

Definition at line 248 of file job_queue.hpp.

atomic_size_type

template<typename T >

using Job::SPMCDeque< T >::atomic_size_type = std::atomic<size_type>

Definition at line 249 of file job_queue.hpp.

Constructor & Destructor Documentation

SPMCDeque()

template<typename T >

Job::SPMCDeque< T >::SPMCDeque ( )

default

~SPMCDeque()

template<typename T >

Job::SPMCDeque< T >::~SPMCDeque ( )

default

Member Function Documentation

Initialize()

template<typename T >

void Job::SPMCDeque< T >::Initialize ( AtomicT *const  memory_backing, const size_type  capacity  )

inlinenoexcept

Definition at line 266 of file job_queue.hpp.

    {
      m_ProducerIndex = 0;
      m_ConsumerIndex = 0;
      m_Data          = memory_backing;
      m_Capacity      = capacity;
      m_CapacityMask  = capacity - 1;
 
      JobAssert((m_Capacity & m_CapacityMask) == 0, "Capacity must be a power of 2.");
    }

References JobAssert, Job::SPMCDeque< T >::m_Capacity, Job::SPMCDeque< T >::m_CapacityMask, Job::SPMCDeque< T >::m_ConsumerIndex, Job::SPMCDeque< T >::m_Data, and Job::SPMCDeque< T >::m_ProducerIndex.

Push()

template<typename T >

SPMCDequeStatus Job::SPMCDeque< T >::Push ( const T &  value )

inline

Definition at line 279 of file job_queue.hpp.

    {
      const size_type write_index = m_ProducerIndex.load(std::memory_order_relaxed);
      const size_type read_index  = m_ConsumerIndex.load(std::memory_order_acquire);
      const size_type size        = write_index - read_index;
 
      if (size > m_CapacityMask)
      {
        return SPMCDequeStatus::FAILED_SIZE;
      }
 
      ElementAt(write_index)->store(value, std::memory_order_relaxed);
 
      m_ProducerIndex.store(write_index + 1, std::memory_order_release);
 
      return SPMCDequeStatus::SUCCESS;
    }

References Job::SPMCDeque< T >::ElementAt(), Job::FAILED_SIZE, Job::SPMCDeque< T >::m_CapacityMask, Job::SPMCDeque< T >::m_ConsumerIndex, Job::SPMCDeque< T >::m_ProducerIndex, and Job::SUCCESS.

Pop()

template<typename T >

SPMCDequeStatus Job::SPMCDeque< T >::Pop ( T *const  out_value )

inline

Definition at line 297 of file job_queue.hpp.

    {
      const size_type producer_index = m_ProducerIndex.load(std::memory_order_relaxed) - 1;
 
      // Reserve the slot at the producer end.
      m_ProducerIndex.store(producer_index, std::memory_order_relaxed);
 
      // The above store needs to happen before this next read
      // to have consistent view of the buffer.
      //
      // `m_ProducerIndex` can only be written to by this thread
      // so first reserve a slot then we read what the other threads have to say.
      //
      std::atomic_thread_fence(std::memory_order_seq_cst);
 
      size_type consumer_index = m_ConsumerIndex.load(std::memory_order_relaxed);
 
      if (consumer_index <= producer_index)
      {
        if (consumer_index == producer_index)  // Only one item in queue
        {
          const bool successful_pop = m_ConsumerIndex.compare_exchange_strong(consumer_index, consumer_index + 1, std::memory_order_seq_cst, std::memory_order_relaxed);
 
          if (successful_pop)
          {
            *out_value = ElementAt(producer_index)->load(std::memory_order_relaxed);
          }
 
          m_ProducerIndex.store(producer_index + 1, std::memory_order_relaxed);
          return successful_pop ? SPMCDequeStatus::SUCCESS : SPMCDequeStatus::FAILED_RACE;
        }
 
        *out_value = ElementAt(producer_index)->load(std::memory_order_relaxed);
        return SPMCDequeStatus::SUCCESS;
      }
 
      // Empty Queue, so restore to canonical empty.
      m_ProducerIndex.store(producer_index + 1, std::memory_order_seq_cst);
      return SPMCDequeStatus::FAILED_SIZE;
    }

References Job::SPMCDeque< T >::ElementAt(), Job::FAILED_RACE, Job::FAILED_SIZE, Job::SPMCDeque< T >::m_ConsumerIndex, Job::SPMCDeque< T >::m_ProducerIndex, and Job::SUCCESS.

Steal()

template<typename T >

SPMCDequeStatus Job::SPMCDeque< T >::Steal ( T *const  out_value )

inline

Definition at line 340 of file job_queue.hpp.

    {
      size_type read_index = m_ConsumerIndex.load(std::memory_order_acquire);
 
      // Must fully read `m_ConsumerIndex` before we read the producer owned `m_ProducerIndex`.
      std::atomic_thread_fence(std::memory_order_seq_cst);
 
      const size_type write_index = m_ProducerIndex.load(std::memory_order_acquire);
 
      // if (next_read_index <= write_index)
      if (read_index < write_index)
      {
        // Must load result before the CAS, since a push can happen concurrently right after the CAS.
        T result = ElementAt(read_index)->load(std::memory_order_relaxed);
 
        // Need strong memory ordering to read the element before the cas.
        if (m_ConsumerIndex.compare_exchange_strong(read_index, read_index + 1, std::memory_order_seq_cst, std::memory_order_relaxed))
        {
          *out_value = std::move(result);
          return SPMCDequeStatus::SUCCESS;
        }
 
        return SPMCDequeStatus::FAILED_RACE;
      }
 
      return SPMCDequeStatus::FAILED_SIZE;
    }

References Job::SPMCDeque< T >::ElementAt(), Job::FAILED_RACE, Job::FAILED_SIZE, Job::SPMCDeque< T >::m_ConsumerIndex, Job::SPMCDeque< T >::m_ProducerIndex, and Job::SUCCESS.

ElementAt()

template<typename T >

AtomicT * Job::SPMCDeque< T >::ElementAt ( const size_type  index ) const

inlineprivatenoexcept

Definition at line 369 of file job_queue.hpp.

    {
      return m_Data + (index & m_CapacityMask);
    }

References Job::SPMCDeque< T >::m_CapacityMask, and Job::SPMCDeque< T >::m_Data.

Referenced by Job::SPMCDeque< T >::Pop(), Job::SPMCDeque< T >::Push(), and Job::SPMCDeque< T >::Steal().

Member Data Documentation

m_ProducerIndex

template<typename T >

atomic_size_type Job::SPMCDeque< T >::m_ProducerIndex

private

Definition at line 252 of file job_queue.hpp.

Referenced by Job::SPMCDeque< T >::Initialize(), Job::SPMCDeque< T >::Pop(), Job::SPMCDeque< T >::Push(), and Job::SPMCDeque< T >::Steal().

m_ConsumerIndex

template<typename T >

atomic_size_type Job::SPMCDeque< T >::m_ConsumerIndex

private

Definition at line 253 of file job_queue.hpp.

Referenced by Job::SPMCDeque< T >::Initialize(), Job::SPMCDeque< T >::Pop(), Job::SPMCDeque< T >::Push(), and Job::SPMCDeque< T >::Steal().

m_Padding0

template<typename T >

unsigned char Job::SPMCDeque< T >::m_Padding0[k_FalseSharingPadSize - sizeof(m_ProducerIndex) - sizeof(m_ConsumerIndex)]

private

Definition at line 254 of file job_queue.hpp.

m_Data

template<typename T >

AtomicT* Job::SPMCDeque< T >::m_Data

private

Definition at line 258 of file job_queue.hpp.

Referenced by Job::SPMCDeque< T >::ElementAt(), and Job::SPMCDeque< T >::Initialize().

m_Capacity

template<typename T >

size_type Job::SPMCDeque< T >::m_Capacity

private

Definition at line 259 of file job_queue.hpp.

Referenced by Job::SPMCDeque< T >::Initialize().

m_CapacityMask

template<typename T >

size_type Job::SPMCDeque< T >::m_CapacityMask

private

Definition at line 260 of file job_queue.hpp.

Referenced by Job::SPMCDeque< T >::ElementAt(), Job::SPMCDeque< T >::Initialize(), and Job::SPMCDeque< T >::Push().

---

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

• include/concurrent/job_queue.hpp