#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 238 of file job_queue.hpp.

Member Typedef Documentation

◆ AtomicT

template<typename T >

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

private

Definition at line 241 of file job_queue.hpp.

◆ size_type

template<typename T >

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

Definition at line 246 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 247 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 264 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 277 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 295 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 338 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 367 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 250 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 251 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 252 of file job_queue.hpp.

◆ m_Data

template<typename T >

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

private

Definition at line 256 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 257 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 258 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

Public Types

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ AtomicT

◆ size_type

◆ atomic_size_type

Constructor & Destructor Documentation

◆ SPMCDeque()

◆ ~SPMCDeque()

Member Function Documentation

◆ Initialize()

◆ Push()

◆ Pop()

◆ Steal()

◆ ElementAt()

Member Data Documentation

◆ m_ProducerIndex

◆ m_ConsumerIndex

◆ m_Padding0

◆ m_Data

◆ m_Capacity

◆ m_CapacityMask