#include <job_queue.hpp>

Classes
struct	IndexRange

Public Types
using	size_type = std::size_t

using	atomic_size_type = std::atomic< size_type >

using	value_type = unsigned char

Public Member Functions
	MPMCQueue ()=default

	~MPMCQueue ()=default

void	Initialize (value_type *const memory_backing, const size_type capacity) noexcept

bool	PushExact (const value_type *elements, const size_type num_elements)

size_type	PushUpTo (const value_type *elements, const size_type num_elements)

bool	PopExact (value_type *out_elements, const size_type num_elements)

size_type	PopUpTo (value_type *out_elements, const size_type num_elements)

Private Member Functions
template<bool allOrNothing>
size_type	PushImpl (const value_type *elements, const size_type num_elements)

template<bool allOrNothing>
size_type	PopImpl (value_type *out_elements, const size_type num_elements)

template<bool allOrNothing>
bool	RequestWriteRange (IndexRange *out_range, const size_type num_items)

template<bool allOrNothing>
bool	RequestPopRange (IndexRange *out_range, const size_type num_items)

size_type	WriteElements (const value_type *const elements, const IndexRange range)

size_type	ReadElements (value_type *const out_elements, const IndexRange range) const

void	Commit (atomic_size_type *commit, const IndexRange range) const

size_type	Distance (const size_type a, const size_type b) const

Private Attributes
atomic_size_type	m_ProducerPending

atomic_size_type	m_ProducerCommited

unsigned char	m_Padding0 [k_FalseSharingPadSize - sizeof(atomic_size_type) *2]

atomic_size_type	m_ConsumerPending

atomic_size_type	m_ConsumerCommited

unsigned char	m_Padding1 [k_FalseSharingPadSize - sizeof(atomic_size_type) *2]

value_type *	m_Queue

size_type	m_Capacity

unsigned char	m_Padding2 [k_FalseSharingPadSize - sizeof(m_Queue) - sizeof(m_Capacity)]

Detailed Description

Definition at line 378 of file job_queue.hpp.

Class Documentation

◆ Job::MPMCQueue::IndexRange

struct Job::MPMCQueue::IndexRange

Definition at line 386 of file job_queue.hpp.

Class Members
size_type	start
size_type	end

Member Typedef Documentation

◆ size_type

using Job::MPMCQueue::size_type = std::size_t

Definition at line 381 of file job_queue.hpp.

◆ atomic_size_type

using Job::MPMCQueue::atomic_size_type = std::atomic<size_type>

Definition at line 382 of file job_queue.hpp.

◆ value_type

using Job::MPMCQueue::value_type = unsigned char

Definition at line 383 of file job_queue.hpp.

Constructor & Destructor Documentation

◆ MPMCQueue()

Job::MPMCQueue::MPMCQueue ( )

default

◆ ~MPMCQueue()

Job::MPMCQueue::~MPMCQueue ( )

default

Member Function Documentation

◆ Initialize()

void Job::MPMCQueue::Initialize	(	value_type *const	memory_backing,
		const size_type	capacity
	)

inlinenoexcept

Definition at line 408 of file job_queue.hpp.

    {
      m_ProducerPending.store(0, std::memory_order_relaxed);
      m_ProducerCommited.store(0, std::memory_order_relaxed);
      m_ConsumerPending.store(0, std::memory_order_relaxed);
      m_ConsumerCommited.store(0, std::memory_order_relaxed);
      m_Queue    = memory_backing;
      m_Capacity = capacity;
    }

References m_Capacity, m_ConsumerCommited, m_ConsumerPending, m_ProducerCommited, m_ProducerPending, and m_Queue.

◆ PushExact()

bool Job::MPMCQueue::PushExact	(	const value_type *	elements,
		const size_type	num_elements
	)

inline

Definition at line 420 of file job_queue.hpp.

    {
      return PushImpl<true>(elements, num_elements) != 0u;
    }

◆ PushUpTo()

size_type Job::MPMCQueue::PushUpTo	(	const value_type *	elements,
		const size_type	num_elements
	)

inline

Definition at line 425 of file job_queue.hpp.

    {
      return PushImpl<false>(elements, num_elements);
    }

◆ PopExact()

bool Job::MPMCQueue::PopExact	(	value_type *	out_elements,
		const size_type	num_elements
	)

inline

Definition at line 430 of file job_queue.hpp.

    {
      return PopImpl<true>(out_elements, num_elements) != 0u;
    }

◆ PopUpTo()

size_type Job::MPMCQueue::PopUpTo	(	value_type *	out_elements,
		const size_type	num_elements
	)

inline

Definition at line 435 of file job_queue.hpp.

    {
      return PopImpl<false>(out_elements, num_elements);
    }

◆ PushImpl()

template<bool allOrNothing>

size_type Job::MPMCQueue::PushImpl	(	const value_type *	elements,
		const size_type	num_elements
	)

inlineprivate

Definition at line 442 of file job_queue.hpp.

    {
      IndexRange range;
      if (RequestWriteRange<allOrNothing>(&range, num_elements))
      {
        const size_type written_elements = WriteElements(elements, range);
        Commit(&m_ProducerCommited, range);
        return written_elements;
      }
 
      return 0u;
    }

References Commit(), m_ProducerCommited, and WriteElements().

◆ PopImpl()

template<bool allOrNothing>

size_type Job::MPMCQueue::PopImpl	(	value_type *	out_elements,
		const size_type	num_elements
	)

inlineprivate

Definition at line 456 of file job_queue.hpp.

    {
      IndexRange range;
      if (RequestPopRange<allOrNothing>(&range, num_elements))
      {
        const size_type read_elements = ReadElements(out_elements, range);
        Commit(&m_ConsumerCommited, range);
        return read_elements;
      }
 
      return 0u;
    }

References Commit(), m_ConsumerCommited, and ReadElements().

◆ RequestWriteRange()

template<bool allOrNothing>

bool Job::MPMCQueue::RequestWriteRange	(	IndexRange *	out_range,
		const size_type	num_items
	)

inlineprivate

Definition at line 470 of file job_queue.hpp.

    {
      size_type old_head, new_head;
 
      old_head = m_ProducerPending.load(std::memory_order_relaxed);
      do
      {
        const size_type tail = m_ConsumerCommited.load(std::memory_order_acquire);
 
        size_type capacity_left = Distance(old_head, tail);
        if constexpr (allOrNothing)
        {
          if (capacity_left < num_items)
          {
            capacity_left = 0;
          }
        }
 
        if (capacity_left == 0)
        {
          return false;
        }
 
        const size_type num_element_to_write = capacity_left < num_items ? capacity_left : num_items;
 
        new_head = old_head + num_element_to_write;
 
      } while (!m_ProducerPending.compare_exchange_weak(old_head, new_head, std::memory_order_relaxed, std::memory_order_relaxed));
 
      *out_range = {old_head, new_head};
      return true;
    }

References Distance(), m_ConsumerCommited, and m_ProducerPending.

◆ RequestPopRange()

template<bool allOrNothing>

bool Job::MPMCQueue::RequestPopRange	(	IndexRange *	out_range,
		const size_type	num_items
	)

inlineprivate

Definition at line 504 of file job_queue.hpp.

    {
      size_type old_tail, new_tail;
 
      old_tail = m_ConsumerPending.load(std::memory_order_relaxed);
      do
      {
        const size_type head     = m_ProducerCommited.load(std::memory_order_acquire);
        const size_type distance = Distance(head, old_tail);
 
        size_t capacity_left = (m_Capacity - distance);
        if constexpr (allOrNothing)
        {
          if (capacity_left < num_items)
          {
            capacity_left = 0;
          }
        }
 
        if (!capacity_left)
        {
          return false;
        }
 
        const size_type num_element_to_read = capacity_left < num_items ? capacity_left : num_items;
 
        new_tail = old_tail + num_element_to_read;
 
      } while (!m_ConsumerPending.compare_exchange_weak(old_tail, new_tail, std::memory_order_relaxed, std::memory_order_relaxed));
 
      *out_range = {old_tail, new_tail};
      return true;
    }

References Distance(), m_Capacity, m_ConsumerPending, and m_ProducerCommited.

◆ WriteElements()

size_type Job::MPMCQueue::WriteElements	(	const value_type *const	elements,
		const IndexRange	range
	)

inlineprivate

Definition at line 538 of file job_queue.hpp.

    {
      const size_type real_start             = range.start % m_Capacity;
      const size_type write_size             = Distance(real_start, range.end % m_Capacity);
      const size_type capacity_before_split  = m_Capacity - real_start;
      const size_type num_items_before_split = write_size < capacity_before_split ? write_size : capacity_before_split;
      const size_type num_items_after_split  = write_size - num_items_before_split;
 
      std::copy_n(elements + 0u, num_items_before_split, m_Queue + real_start);
      std::copy_n(elements + num_items_before_split, num_items_after_split, m_Queue + 0u);
 
      return write_size;
    }

References Distance(), Job::MPMCQueue::IndexRange::end, m_Capacity, m_Queue, and Job::MPMCQueue::IndexRange::start.

Referenced by PushImpl().

◆ ReadElements()

size_type Job::MPMCQueue::ReadElements	(	value_type *const	out_elements,
		const IndexRange	range
	)		const

inlineprivate

Definition at line 552 of file job_queue.hpp.

    {
      const size_type real_start             = range.start % m_Capacity;
      const size_type read_size              = Distance(real_start, range.end % m_Capacity);
      const size_type capacity_before_split  = m_Capacity - real_start;
      const size_type num_items_before_split = read_size < capacity_before_split ? read_size : capacity_before_split;
      const size_type num_items_after_split  = read_size - num_items_before_split;
 
      std::copy_n(std::make_move_iterator(m_Queue + real_start), num_items_before_split, out_elements + 0u);
      std::copy_n(std::make_move_iterator(m_Queue + 0u), num_items_after_split, out_elements + num_items_before_split);
 
      return read_size;
    }

References Distance(), Job::MPMCQueue::IndexRange::end, m_Capacity, m_Queue, and Job::MPMCQueue::IndexRange::start.

Referenced by PopImpl().

◆ Commit()

void Job::MPMCQueue::Commit	(	atomic_size_type *	commit,
		const IndexRange	range
	)		const

inlineprivate

Definition at line 566 of file job_queue.hpp.

    {
      size_type start_copy;
      while (!commit->compare_exchange_weak(
       start_copy = range.start,
       range.end,
       std::memory_order_release,
       std::memory_order_relaxed))
      {
        PauseProcessor();
      }
    }

References Job::MPMCQueue::IndexRange::end, Job::PauseProcessor(), and Job::MPMCQueue::IndexRange::start.

Referenced by PopImpl(), and PushImpl().

◆ Distance()

size_type Job::MPMCQueue::Distance	(	const size_type	a,
		const size_type	b
	)		const

inlineprivate

Definition at line 579 of file job_queue.hpp.

    {
      return (b > a) ? (b - a) : m_Capacity - a + b;
    }

References m_Capacity.

Referenced by ReadElements(), RequestPopRange(), RequestWriteRange(), and WriteElements().

Member Data Documentation

◆ m_ProducerPending

atomic_size_type Job::MPMCQueue::m_ProducerPending

private

Definition at line 393 of file job_queue.hpp.

Referenced by Initialize(), and RequestWriteRange().

◆ m_ProducerCommited

atomic_size_type Job::MPMCQueue::m_ProducerCommited

private

Definition at line 394 of file job_queue.hpp.

Referenced by Initialize(), PushImpl(), and RequestPopRange().

◆ m_Padding0

unsigned char Job::MPMCQueue::m_Padding0[k_FalseSharingPadSize - sizeof(atomic_size_type) *2]

private

Definition at line 395 of file job_queue.hpp.

◆ m_ConsumerPending

atomic_size_type Job::MPMCQueue::m_ConsumerPending

private

Definition at line 396 of file job_queue.hpp.

Referenced by Initialize(), and RequestPopRange().

◆ m_ConsumerCommited

atomic_size_type Job::MPMCQueue::m_ConsumerCommited

private

Definition at line 397 of file job_queue.hpp.

Referenced by Initialize(), PopImpl(), and RequestWriteRange().

◆ m_Padding1

unsigned char Job::MPMCQueue::m_Padding1[k_FalseSharingPadSize - sizeof(atomic_size_type) *2]

private

Definition at line 398 of file job_queue.hpp.

◆ m_Queue

value_type* Job::MPMCQueue::m_Queue

private

Definition at line 399 of file job_queue.hpp.

Referenced by Initialize(), ReadElements(), and WriteElements().

◆ m_Capacity

size_type Job::MPMCQueue::m_Capacity

private

Definition at line 400 of file job_queue.hpp.

Referenced by Distance(), Initialize(), ReadElements(), RequestPopRange(), and WriteElements().

◆ m_Padding2

unsigned char Job::MPMCQueue::m_Padding2[k_FalseSharingPadSize - sizeof(m_Queue) - sizeof(m_Capacity)]

private

Definition at line 401 of file job_queue.hpp.

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

include/concurrent/job_queue.hpp

Classes

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Class Documentation

◆ Job::MPMCQueue::IndexRange

Member Typedef Documentation

◆ size_type

◆ atomic_size_type

◆ value_type

Constructor & Destructor Documentation

◆ MPMCQueue()

◆ ~MPMCQueue()

Member Function Documentation

◆ Initialize()

◆ PushExact()

◆ PushUpTo()

◆ PopExact()

◆ PopUpTo()

◆ PushImpl()

◆ PopImpl()

◆ RequestWriteRange()

◆ RequestPopRange()

◆ WriteElements()

◆ ReadElements()

◆ Commit()

◆ Distance()

Member Data Documentation

◆ m_ProducerPending

◆ m_ProducerCommited

◆ m_Padding0

◆ m_ConsumerPending

◆ m_ConsumerCommited

◆ m_Padding1

◆ m_Queue

◆ m_Capacity

◆ m_Padding2