mysql::concurrency::Sync_bounded_queue< T, capacity_tp > Class Template Reference

Bounded concurrent queue supporting multiple producers and consumers. More...

#include <sync_bounded_queue.h>

Public Types
using	Element_type = T
using	Allocator = mysql::allocators::Allocator< T >
using	Memory_resource = mysql::allocators::Memory_resource
using	Mutex_type = std::mutex
using	Cv_type = std::condition_variable

Public Member Functions
	Sync_bounded_queue (Memory_resource={})
	Construct the queue. More...
template<typename P >
bool	enqueue (Element_type &&element, P &&stop_predicate)
	Enqueue (push) an element into the queue. More...
bool	enqueue (Element_type &&element)
	Overload without stop_predicate, using default that always returns false. More...
template<typename P >
std::pair< Element_type, bool >	dequeue (P &&stop_predicate)
	Dequeue an element from the queue. More...
bool	empty ()
	Check if the queue is empty. More...
void	notify_all ()
	Signal shutdown: Notifies all blocked consumers. More...
virtual	~Sync_bounded_queue ()=default
	Destructor Before deleting the queue, the caller must ensure that no thread is or will be using the queue, for instance, by first calling notify_all and then joining all consumer threads. More...
std::size_t	size () const
	Estimates the queue size based on m_head and m_tail positions. More...
void	reset ()
	Reset the queue: Clears all slots (sets validity=false, elements to default), resets m_head and m_tail to 0. More...

Static Public Attributes
static constexpr uint64_t	m_capacity = capacity_tp

Private Member Functions
std::size_t	get_next_index (std::atomic< std::size_t > &current)
	Atomically increments the given position (m_head for producers, m_tail for consumers) and returns the new index modulo capacity. More...

Private Attributes
std::array< detail::Padded_slot< Element_type, Mutex_type, Cv_type >, capacity_tp >	m_slots
	Padded slots containing elements and synchronization primitives. More...
std::atomic< std::size_t >	m_head {0}
	Current head idx, when at the end, wraps to 0. More...
std::atomic< std::size_t >	m_tail {0}
	Current tail idx, when at the end, wraps to 0. More...

Detailed Description

template<typename T, uint64_t capacity_tp>
requires Is_power_of_two<capacity_tp>
class mysql::concurrency::Sync_bounded_queue< T, capacity_tp >

Bounded concurrent queue supporting multiple producers and consumers.

Designed to minimize contention using lock-free index allocation followed by per-slot locking for safe element access and notification.

—Implementation: The queue uses a fixed-size circular buffer implemented as a std::array of Padded_slot structures. Each slot contains:

• The element (T)
• A bool validity flag indicating if the slot holds a valid element
• A mutex (validity_mt) protecting the validity and element
• Two condition variables: producer_cv (wait for empty slot) and consumer_cv (wait for valid element)

Producers:

• Lock-free: Atomically fetch_add(1) on m_head to get a unique index (m_head is the next producer position).
• Lock slot mutex, wait on producer_cv until !validity (empty slot).
• Assign element to slot, set validity = true.
• Unlock and notify one waiting consumer on consumer_cv.

Consumers:

• Lock-free: Atomically fetch_add(1) on m_tail to get a unique index (m_tail is the next consumer position).
• Lock slot mutex, wait on consumer_cv until validity (valid element).
• If stop_predicate() is true, return empty pair with false.
• Otherwise, move element, set validity = false.
• Unlock and notify one waiting producer on producer_cv.

Indices wrap around using % m_capacity (capacity is power of 2). The queue has fixed capacity; enqueue blocks if full (no resizing). Contention is low if capacity >= num_producers + num_consumers. Index allocation is lock-free, but slot access uses per-slot mutex/CV.

Template Parameters

T	Type of element stored in the queue.
capacity_tp	Fixed capacity of the queue (must be power of 2).

Member Typedef Documentation

Allocator

template<typename T , uint64_t capacity_tp>

using mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::Allocator = mysql::allocators::Allocator<T>

Cv_type

template<typename T , uint64_t capacity_tp>

using mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::Cv_type = std::condition_variable

Element_type

template<typename T , uint64_t capacity_tp>

using mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::Element_type = T

Memory_resource

template<typename T , uint64_t capacity_tp>

using mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::Memory_resource = mysql::allocators::Memory_resource

Mutex_type

template<typename T , uint64_t capacity_tp>

using mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::Mutex_type = std::mutex

Constructor & Destructor Documentation

Sync_bounded_queue()

template<typename T , uint64_t capacity_tp>

mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::Sync_bounded_queue

(

Memory_resource

= {}

)

Construct the queue.

The memory_resource parameter is reserved for future use and currently ignored.

~Sync_bounded_queue()

template<typename T , uint64_t capacity_tp>

virtual mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::~Sync_bounded_queue ( )

virtualdefault

Destructor Before deleting the queue, the caller must ensure that no thread is or will be using the queue, for instance, by first calling notify_all and then joining all consumer threads.

Member Function Documentation

dequeue()

template<typename T , uint64_t capacity_tp>

template<typename P >

std::pair< Element_type, bool > mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::dequeue

(

P &&

stop_predicate

)

Dequeue an element from the queue.

Called by consumers.

Parameters

stop_predicate

Predicate checked after wakeup; if true and still cannot dequeue, stops waiting and returns {default T, false}.

Returns

Pair of {consumed element, true} if successful, or {default T, false} if stopped by predicate.

• Lock-free: Advances m_tail via atomic fetch_add to get unique index.
• Locks the slot mutex and waits on consumer_cv until the slot is valid (validity == true).
• If stop_predicate() is true, returns without consuming.
• Otherwise, moves the element and sets validity = false.
• Notifies one waiting producer on producer_cv. The operation is lock-free for index allocation but uses per-slot locking for safe access. Blocks if empty until an element is available or stopped.

empty()

template<typename T , uint64_t capacity_tp>

bool mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::empty

(

)

Check if the queue is empty.

Returns

True when queue is empty, false otherwise.

enqueue() [1/2]

template<typename T , uint64_t capacity_tp>

bool mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::enqueue

(

Element_type &&

element

)

Overload without stop_predicate, using default that always returns false.

Parameters

element

Element to enqueue

enqueue() [2/2]

template<typename T , uint64_t capacity_tp>

template<typename P >

bool mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::enqueue	(	Element_type &&	element,
		P &&	stop_predicate
	)

Enqueue (push) an element into the queue.

Blocks the calling producer if the queue is full (no space available).

• Lock-free: Advances m_head via atomic fetch_add to get unique index.
• Locks the slot mutex and waits on producer_cv until the slot is empty (!validity).
• Assigns the element and sets validity = true.
• Notifies one waiting consumer on consumer_cv. The operation is lock-free for index allocation but uses per-slot locking for safe access. Blocks only if full; choose capacity >= num_producers + num_consumers to minimize blocking.

  Parameters

  element	Element to enqueue
  stop_predicate	Predicate checked after wakeup; if true and still cannot enqueue, stops waiting and returns {default T, false}.

get_next_index()

template<typename T , uint64_t capacity_tp>

std::size_t mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::get_next_index ( std::atomic< std::size_t > &  current )

private

Atomically increments the given position (m_head for producers, m_tail for consumers) and returns the new index modulo capacity.

Used internally for lock-free index allocation.

Parameters

[in,out]

current

Atomic position to advance (m_head or m_tail).

Returns

The granted index after increment (wrapped).

notify_all()

template<typename T , uint64_t capacity_tp>

void mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::notify_all

(

)

Signal shutdown: Notifies all blocked consumers.

Allows waiting consumers to wake and check stop_predicate to exit gracefully. Producer stop must be handled by the structure user, before calling "notify_all".

reset()

template<typename T , uint64_t capacity_tp>

void mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::reset

(

)

Reset the queue: Clears all slots (sets validity=false, elements to default), resets m_head and m_tail to 0.

size()

template<typename T , uint64_t capacity_tp>

std::size_t mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::size

(

)

const

Estimates the queue size based on m_head and m_tail positions.

Due to concurrent modifications, this is an approximation. The actual size may be bigger than the returned value; the difference is at most min(N,M) where N and M are the number of concurrent enqueue/dequeue operations.

Returns

Estimated number of elements (unconsumed valid slots).

Member Data Documentation

m_capacity

template<typename T , uint64_t capacity_tp>

constexpr uint64_t mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::m_capacity = capacity_tp

staticconstexpr

m_head

template<typename T , uint64_t capacity_tp>

std::atomic<std::size_t> mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::m_head {0}

private

Current head idx, when at the end, wraps to 0.

m_slots

template<typename T , uint64_t capacity_tp>

std::array<detail::Padded_slot<Element_type, Mutex_type, Cv_type>, capacity_tp> mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::m_slots

private

Padded slots containing elements and synchronization primitives.

m_tail

template<typename T , uint64_t capacity_tp>

std::atomic<std::size_t> mysql::concurrency::Sync_bounded_queue< T, capacity_tp >::m_tail {0}

private

Current tail idx, when at the end, wraps to 0.

---

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

• libs/mysql/concurrency/sync_bounded_queue.h