无锁队列实现 / Lock-free queue implementation More...

#include <lockfree_queue.hpp>

Public Member Functions
	LockFreeQueue (size_t length)
	构造函数 / Constructor

	~LockFreeQueue ()
	析构函数 / Destructor

Data *	operator[] (uint32_t index)
	获取指定索引的数据指针 / Retrieves the data pointer at a specified index

template<typename ElementData = Data>
ErrorCode	Push (ElementData &&item)
	向队列中推入数据 / Pushes data into the queue

template<typename ElementData = Data>
ErrorCode	Pop (ElementData &item)
	从队列中弹出数据 / Pops data from the queue

ErrorCode	Pop (Data &item)
	从队列中移除头部元素，并获取该元素的数据 (Remove the front element from the queue and retrieve its data).

ErrorCode	Pop ()
	从队列中弹出数据（不返回数据） / Pops data from the queue (without returning data)

ErrorCode	Peek (Data &item)
	获取队列头部数据但不弹出 / Retrieves the front data of the queue without popping

ErrorCode	PushBatch (const Data *data, size_t size)
	批量推入数据 / Pushes multiple elements into the queue

template<typename Writer >
ErrorCode	PushWithWriter (size_t size, Writer &&writer)
	通过写入器回调写入固定长度数据（单生产者） / Push fixed-size data via writer callback (single producer)

template<typename Reader >
ErrorCode	PopWithReader (size_t size, Reader &&reader)
	通过读取器回调弹出固定长度数据（单消费者） / Pop fixed-size data via reader callback (single consumer)

ErrorCode	PopBatch (Data *data, size_t size)
	批量弹出数据 / Pops multiple elements from the queue

ErrorCode	PeekBatch (Data *data, size_t size)
	批量查看队列中的数据（不移除） / Peeks multiple elements from the queue without removing them

void	Reset ()
	重置队列 / Resets the queue

size_t	Size () const
	获取当前队列中的元素数量 / Returns the number of elements currently in the queue

size_t	EmptySize ()
	计算队列剩余可用空间 / Calculates the remaining available space in the queue

size_t	MaxSize () const
	获取队列的最大容量 / Returns the maximum capacity of the queue

Private Member Functions
constexpr size_t	AlignUp (size_t size, size_t align)

uint32_t	Increment (uint32_t index) const

Private Attributes
std::atomic< uint32_t >	head_

std::atomic< uint32_t >	tail_

const size_t	LENGTH

Data *	queue_handle_

Detailed Description

template<typename Data>
class LibXR::LockFreeQueue< Data >

无锁队列实现 / Lock-free queue implementation

该类实现了单生产者多消费者无锁队列（SPMC Lock-Free Queue），支持多线程环境下的高效入队和出队操作，适用于需要高并发性能的场景，如实时系统和多线程数据处理。 This class implements a single-producer, multiple-consumer lock-free queue (SPMC Lock-Free Queue) that supports high-efficiency enqueue and dequeue operations in a multi-threaded environment. It is suitable for scenarios requiring high concurrency, such as real-time systems and multi-threaded data processing.

Template Parameters

Data	队列存储的数据类型 / The type of data stored in the queue.

Definition at line 29 of file lockfree_queue.hpp.

Constructor & Destructor Documentation

◆ LockFreeQueue()

template<typename Data >

LibXR::LockFreeQueue< Data >::LockFreeQueue ( size_t length )

inline

构造函数 / Constructor

Parameters

length 队列的最大容量 / Maximum capacity of the queue

创建一个指定大小的无锁队列，并初始化相关变量。 Creates a lock-free queue with the specified size and initializes relevant variables.

Note: 包含动态内存分配。 Contains dynamic memory allocation.

Definition at line 47 of file lockfree_queue.hpp.

      : head_(0),
        tail_(0),
        LENGTH(AlignUp(length, LibXR::ALIGN_SIZE) - 1),
        queue_handle_(new Data[LENGTH + 1])
  {
  }

◆ ~LockFreeQueue()

template<typename Data >

LibXR::LockFreeQueue< Data >::~LockFreeQueue ( )

inline

析构函数 / Destructor

释放队列所占用的内存。 Releases the memory occupied by the queue.

Definition at line 61 of file lockfree_queue.hpp.

61{ delete[] queue_handle_; }

Member Function Documentation

◆ AlignUp()

template<typename Data >

size_t LibXR::LockFreeQueue< Data >::AlignUp	(	size_t	size,
		size_t	align )

inlineconstexprprivate

Definition at line 31 of file lockfree_queue.hpp.

  {
    return (size / align + 1) * align;
  }

◆ EmptySize()

template<typename Data >

size_t LibXR::LockFreeQueue< Data >::EmptySize ( )

inline

计算队列剩余可用空间 / Calculates the remaining available space in the queue

Definition at line 509 of file lockfree_queue.hpp.

509{ return LENGTH - Size(); }

LibXR::LockFreeQueue::Size

size_t Size() const

获取当前队列中的元素数量 / Returns the number of elements currently in the queue

Definition lockfree_queue.hpp:498

◆ Increment()

template<typename Data >

uint32_t LibXR::LockFreeQueue< Data >::Increment ( uint32_t index ) const

inlineprivate

Definition at line 522 of file lockfree_queue.hpp.

522{ return (index + 1) % (LENGTH + 1); }

◆ MaxSize()

template<typename Data >

size_t LibXR::LockFreeQueue< Data >::MaxSize ( ) const

inline

获取队列的最大容量 / Returns the maximum capacity of the queue

Definition at line 514 of file lockfree_queue.hpp.

514{ return LENGTH; }

◆ operator[]()

template<typename Data >

Data * LibXR::LockFreeQueue< Data >::operator[] ( uint32_t index )

inline

获取指定索引的数据指针 / Retrieves the data pointer at a specified index

Parameters

index 数据索引 / Data index

Returns: 指向该索引数据的指针 / Pointer to the data at the given index

Definition at line 68 of file lockfree_queue.hpp.

68{ return &queue_handle_[static_cast<size_t>(index)]; }

◆ Peek()

template<typename Data >

ErrorCode LibXR::LockFreeQueue< Data >::Peek ( Data & item )

inline

获取队列头部数据但不弹出 / Retrieves the front data of the queue without popping

Parameters

item	用于存储获取的数据 / Variable to store the retrieved data

Returns: 操作结果，成功返回 ErrorCode::OK，队列为空返回 ErrorCode::EMPTY / Operation result: returns ErrorCode::OK on success, ErrorCode::EMPTY if the queue is empty

Definition at line 198 of file lockfree_queue.hpp.

  {
    while (true)
    {
      auto current_head = head_.load(std::memory_order_relaxed);
      if (current_head == tail_.load(std::memory_order_acquire))
      {
        return ErrorCode::EMPTY;
      }
 
      item = queue_handle_[current_head];
 
      if (head_.load(std::memory_order_acquire) == current_head)
      {
        return ErrorCode::OK;
      }
    }
  }

◆ PeekBatch()

template<typename Data >

ErrorCode LibXR::LockFreeQueue< Data >::PeekBatch	(	Data *	data,
		size_t	size )

inline

批量查看队列中的数据（不移除） / Peeks multiple elements from the queue without removing them

Parameters

data	数据存储数组指针 / Pointer to the array to store peeked data
size	要查看的元素个数 / Number of elements to peek

Returns: 操作结果，成功返回 ErrorCode::OK，队列中数据不足返回 ErrorCode::EMPTY / Operation result: returns ErrorCode::OK on success, ErrorCode::EMPTY if not enough data is available

Definition at line 436 of file lockfree_queue.hpp.

  {
    if (size == 0)
    {
      return ErrorCode::OK;
    }
 
    const size_t CAPACITY = LENGTH + 1;
 
    while (true)
    {
      auto current_head = head_.load(std::memory_order_relaxed);
      auto current_tail = tail_.load(std::memory_order_acquire);
 
      size_t available = (current_tail >= current_head)
                             ? (current_tail - current_head)
                             : (CAPACITY - current_head + current_tail);
 
      if (available < size)
      {
        return ErrorCode::EMPTY;
      }
 
      if (data != nullptr)
      {
        size_t first_chunk = LibXR::min(size, CAPACITY - current_head);
        LibXR::Memory::FastCopy(
            reinterpret_cast<void*>(data),
            reinterpret_cast<const void*>(queue_handle_ + current_head),
            first_chunk * sizeof(Data));
 
        if (size > first_chunk)
        {
          LibXR::Memory::FastCopy(reinterpret_cast<void*>(data + first_chunk),
                                  reinterpret_cast<const void*>(queue_handle_),
                                  (size - first_chunk) * sizeof(Data));
        }
      }
 
      if (head_.load(std::memory_order_acquire) == current_head)
      {
        return ErrorCode::OK;
      }
    }
  }

◆ Pop() [1/3]

template<typename Data >

ErrorCode LibXR::LockFreeQueue< Data >::Pop ( )

inline

从队列中弹出数据（不返回数据） / Pops data from the queue (without returning data)

Returns: 操作结果，成功返回 ErrorCode::OK，队列为空返回 ErrorCode::EMPTY / Operation result: returns ErrorCode::OK on success, ErrorCode::EMPTY if the queue is empty

Definition at line 170 of file lockfree_queue.hpp.

  {
    auto current_head = head_.load(std::memory_order_relaxed);
 
    while (true)
    {
      if (current_head == tail_.load(std::memory_order_acquire))
      {
        return ErrorCode::EMPTY;
      }
 
      if (head_.compare_exchange_weak(current_head, Increment(current_head),
                                      std::memory_order_acq_rel,
                                      std::memory_order_relaxed))
      {
        return ErrorCode::OK;
      }
    }
  }

◆ Pop() [2/3]

template<typename Data >

ErrorCode LibXR::LockFreeQueue< Data >::Pop ( Data & item )

inline

从队列中移除头部元素，并获取该元素的数据 (Remove the front element from the queue and retrieve its data).

This function atomically updates the head_ index using compare_exchange_weak to ensure thread safety. If the queue is empty, it returns ErrorCode::EMPTY. Otherwise, it updates the head pointer, retrieves the element, and returns ErrorCode::OK. 该函数使用 compare_exchange_weak 原子地更新 head_ 索引，以确保线程安全。如果队列为空，则返回 ErrorCode::EMPTY，否则更新头指针，获取元素数据，并返回 ErrorCode::OK。

Parameters

item	用于存储弹出元素的引用 (Reference to store the popped element).

Returns

操作结果 (Operation result):

ErrorCode::OK 表示成功移除并获取元素 (ErrorCode::OK if an element was successfully removed and retrieved).
ErrorCode::EMPTY 表示队列为空 (ErrorCode::EMPTY if the queue is empty).

Definition at line 141 of file lockfree_queue.hpp.

  {
    auto current_head = head_.load(std::memory_order_relaxed);
 
    while (true)
    {
      if (current_head == tail_.load(std::memory_order_acquire))
      {
        return ErrorCode::EMPTY;
      }
 
      item = queue_handle_[current_head];
 
      if (head_.compare_exchange_weak(current_head, Increment(current_head),
                                      std::memory_order_acq_rel,
                                      std::memory_order_relaxed))
      {
        return ErrorCode::OK;
      }
    }
  }

◆ Pop() [3/3]

template<typename Data >

template<typename ElementData = Data>

ErrorCode LibXR::LockFreeQueue< Data >::Pop ( ElementData & item )

inline

从队列中弹出数据 / Pops data from the queue

Parameters

item	用于存储弹出数据的变量 / Variable to store the popped data

Returns: 操作结果，成功返回 ErrorCode::OK，队列为空返回 ErrorCode::EMPTY / Operation result: returns ErrorCode::OK on success, ErrorCode::EMPTY if the queue is empty

Definition at line 101 of file lockfree_queue.hpp.

  {
    auto current_head = head_.load(std::memory_order_relaxed);
 
    while (true)
    {
      if (current_head == tail_.load(std::memory_order_acquire))
      {
        return ErrorCode::EMPTY;
      }
 
      item = queue_handle_[current_head];
 
      if (head_.compare_exchange_weak(current_head, Increment(current_head),
                                      std::memory_order_acq_rel,
                                      std::memory_order_relaxed))
      {
        return ErrorCode::OK;
      }
    }
  }

◆ PopBatch()

template<typename Data >

ErrorCode LibXR::LockFreeQueue< Data >::PopBatch	(	Data *	data,
		size_t	size )

inline

批量弹出数据 / Pops multiple elements from the queue

Parameters

data	数据存储数组指针 / Pointer to the array to store popped data
size	需要弹出的数据个数 / Number of elements to pop

Returns: 操作结果，成功返回 ErrorCode::OK，队列为空返回 ErrorCode::EMPTY / Operation result: returns ErrorCode::OK on success, ErrorCode::EMPTY if the queue is empty

Definition at line 383 of file lockfree_queue.hpp.

  {
    size_t capacity = LENGTH + 1;
 
    while (true)
    {
      auto current_head = head_.load(std::memory_order_relaxed);
      auto current_tail = tail_.load(std::memory_order_acquire);
 
      size_t available = (current_tail >= current_head)
                             ? (current_tail - current_head)
                             : (capacity - current_head + current_tail);
 
      if (available < size)
      {
        return ErrorCode::EMPTY;
      }
 
      if (data != nullptr)
      {
        size_t first_chunk = LibXR::min(size, capacity - current_head);
        LibXR::Memory::FastCopy(
            reinterpret_cast<void*>(data),
            reinterpret_cast<const void*>(queue_handle_ + current_head),
            first_chunk * sizeof(Data));
 
        if (size > first_chunk)
        {
          LibXR::Memory::FastCopy(reinterpret_cast<void*>(data + first_chunk),
                                  reinterpret_cast<const void*>(queue_handle_),
                                  (size - first_chunk) * sizeof(Data));
        }
      }
 
      size_t new_head = (current_head + size) % capacity;
 
      if (head_.compare_exchange_weak(current_head, new_head, std::memory_order_acq_rel,
                                      std::memory_order_relaxed))
      {
        return ErrorCode::OK;
      }
    }
  }

◆ PopWithReader()

template<typename Data >

template<typename Reader >

ErrorCode LibXR::LockFreeQueue< Data >::PopWithReader	(	size_t	size,
		Reader &&	reader )

inline

通过读取器回调弹出固定长度数据（单消费者） / Pop fixed-size data via reader callback (single consumer)

Parameters

size	需要弹出的元素数 / Number of elements to pop
reader	读取器：签名 ErrorCode(const Data* buffer, size_t chunk_size) Reader callback signature: ErrorCode(const Data* buffer, size_t chunk_size)

Note: 语义对齐 PopBatch：数据不足返回 EMPTY；仅当整段读取成功后才提交 head。 Semantics align with PopBatch: returns EMPTY when data is insufficient; head is committed only after whole range is read successfully.

Definition at line 329 of file lockfree_queue.hpp.

  {
    static_assert(std::is_invocable_v<Reader&, const Data*, size_t>,
                  "PopWithReader reader must be callable as "
                  "ErrorCode(const Data* buffer, size_t chunk_size)");
    using ReaderRet = std::invoke_result_t<Reader&, const Data*, size_t>;
    static_assert(std::is_convertible_v<ReaderRet, ErrorCode>,
                  "PopWithReader reader return type must be convertible to ErrorCode");
 
    if (size == 0U)
    {
      return ErrorCode::OK;
    }
 
    const auto current_head = head_.load(std::memory_order_relaxed);
    const auto current_tail = tail_.load(std::memory_order_acquire);
    const size_t capacity = LENGTH + 1;
    const size_t available = (current_tail >= current_head)
                                 ? (current_tail - current_head)
                                 : (capacity - current_head + current_tail);
 
    if (available < size)
    {
      return ErrorCode::EMPTY;
    }
 
    const size_t first_chunk = LibXR::min(size, capacity - static_cast<size_t>(current_head));
    Reader& reader_ref = reader;
    const ErrorCode first_ec = reader_ref(queue_handle_ + current_head, first_chunk);
    if (first_ec != ErrorCode::OK)
    {
      return first_ec;
    }
 
    if (size > first_chunk)
    {
      const ErrorCode second_ec = reader_ref(queue_handle_, size - first_chunk);
      if (second_ec != ErrorCode::OK)
      {
        return second_ec;
      }
    }
 
    head_.store((current_head + size) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ Push()

template<typename Data >

template<typename ElementData = Data>

ErrorCode LibXR::LockFreeQueue< Data >::Push ( ElementData && item )

inline

向队列中推入数据 / Pushes data into the queue

Parameters

item	要插入的元素 / Element to be inserted

Returns: 操作结果，成功返回 ErrorCode::OK，队列满返回 ErrorCode::FULL / Operation result: returns ErrorCode::OK on success, ErrorCode::FULL if the queue is full

Definition at line 78 of file lockfree_queue.hpp.

  {
    const auto CURRENT_TAIL = tail_.load(std::memory_order_relaxed);
    const auto NEXT_TAIL = Increment(CURRENT_TAIL);
 
    if (NEXT_TAIL == head_.load(std::memory_order_acquire))
    {
      return ErrorCode::FULL;
    }
 
    queue_handle_[CURRENT_TAIL] = std::forward<ElementData>(item);
    tail_.store(NEXT_TAIL, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ PushBatch()

template<typename Data >

ErrorCode LibXR::LockFreeQueue< Data >::PushBatch	(	const Data *	data,
		size_t	size )

inline

批量推入数据 / Pushes multiple elements into the queue

Parameters

data	数据数组指针 / Pointer to the data array
size	数据个数 / Number of elements

Returns: 操作结果，成功返回 ErrorCode::OK，队列满返回 ErrorCode::FULL / Operation result: returns ErrorCode::OK on success, ErrorCode::FULL if the queue is full

Definition at line 225 of file lockfree_queue.hpp.

  {
    auto current_tail = tail_.load(std::memory_order_relaxed);
    auto current_head = head_.load(std::memory_order_acquire);
 
    size_t capacity = LENGTH + 1;
    size_t free_space = (current_tail >= current_head)
                            ? (capacity - (current_tail - current_head) - 1)
                            : (current_head - current_tail - 1);
 
    if (free_space < size)
    {
      return ErrorCode::FULL;
    }
 
    size_t first_chunk = LibXR::min(size, capacity - current_tail);
    LibXR::Memory::FastCopy(reinterpret_cast<void*>(queue_handle_ + current_tail),
                            reinterpret_cast<const void*>(data),
                            first_chunk * sizeof(Data));
 
    if (size > first_chunk)
    {
      LibXR::Memory::FastCopy(reinterpret_cast<void*>(queue_handle_),
                              reinterpret_cast<const void*>(data + first_chunk),
                              (size - first_chunk) * sizeof(Data));
    }
 
    tail_.store((current_tail + size) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ PushWithWriter()

template<typename Data >

template<typename Writer >

ErrorCode LibXR::LockFreeQueue< Data >::PushWithWriter	(	size_t	size,
		Writer &&	writer )

inline

通过写入器回调写入固定长度数据（单生产者） / Push fixed-size data via writer callback (single producer)

Parameters

size	需要写入的元素数 / Number of elements to write
writer	写入器：签名 ErrorCode(Data* buffer, size_t chunk_size) Writer callback signature: ErrorCode(Data* buffer, size_t chunk_size)

Note: 语义对齐 PushBatch：空间不足返回 FULL；仅当整段写入成功后才提交 tail。 Semantics align with PushBatch: returns FULL when space is insufficient; tail is committed only after whole range is written successfully.

Definition at line 269 of file lockfree_queue.hpp.

  {
    static_assert(std::is_invocable_v<Writer&, Data*, size_t>,
                  "PushWithWriter writer must be callable as "
                  "ErrorCode(Data* buffer, size_t chunk_size)");
    using WriterRet = std::invoke_result_t<Writer&, Data*, size_t>;
    static_assert(std::is_convertible_v<WriterRet, ErrorCode>,
                  "PushWithWriter writer return type must be convertible to ErrorCode");
 
    if (size == 0U)
    {
      return ErrorCode::OK;
    }
 
    const auto current_tail = tail_.load(std::memory_order_relaxed);
    const auto current_head = head_.load(std::memory_order_acquire);
    const size_t capacity = LENGTH + 1;
    const size_t free_space =
        (current_tail >= current_head) ? (capacity - (current_tail - current_head) - 1)
                                       : (current_head - current_tail - 1);
 
    if (free_space < size)
    {
      return ErrorCode::FULL;
    }
 
    const size_t first_chunk = LibXR::min(size, capacity - static_cast<size_t>(current_tail));
    Writer& writer_ref = writer;
    const ErrorCode first_ec = writer_ref(queue_handle_ + current_tail, first_chunk);
    if (first_ec != ErrorCode::OK)
    {
      return first_ec;
    }
 
    if (size > first_chunk)
    {
      const ErrorCode second_ec = writer_ref(queue_handle_, size - first_chunk);
      if (second_ec != ErrorCode::OK)
      {
        return second_ec;
      }
    }
 
    tail_.store((current_tail + size) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ Reset()

template<typename Data >

void LibXR::LockFreeQueue< Data >::Reset ( )

inline

重置队列 / Resets the queue

该方法清空队列并将头尾指针重置为 0。 This method clears the queue and resets the head and tail pointers to 0.

Definition at line 488 of file lockfree_queue.hpp.

  {
    head_.store(0, std::memory_order_relaxed);
    tail_.store(0, std::memory_order_relaxed);
  }

◆ Size()

template<typename Data >

size_t LibXR::LockFreeQueue< Data >::Size ( ) const

inline

获取当前队列中的元素数量 / Returns the number of elements currently in the queue

Definition at line 498 of file lockfree_queue.hpp.

  {
    const auto CURRENT_HEAD = head_.load(std::memory_order_acquire);
    const auto CURRENT_TAIL = tail_.load(std::memory_order_acquire);
    return (CURRENT_TAIL >= CURRENT_HEAD) ? (CURRENT_TAIL - CURRENT_HEAD)
                                          : ((LENGTH + 1) - CURRENT_HEAD + CURRENT_TAIL);
  }

Field Documentation

◆ head_

template<typename Data >

std::atomic<uint32_t> LibXR::LockFreeQueue< Data >::head_

private

Definition at line 517 of file lockfree_queue.hpp.

◆ LENGTH

template<typename Data >

const size_t LibXR::LockFreeQueue< Data >::LENGTH

private

Definition at line 519 of file lockfree_queue.hpp.

◆ queue_handle_

template<typename Data >

Data* LibXR::LockFreeQueue< Data >::queue_handle_

private

Definition at line 520 of file lockfree_queue.hpp.

◆ tail_

template<typename Data >

std::atomic<uint32_t> LibXR::LockFreeQueue< Data >::tail_

private

Definition at line 518 of file lockfree_queue.hpp.

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

src/structure/lockfree_queue.hpp

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ LockFreeQueue()

◆ ~LockFreeQueue()

Member Function Documentation

◆ AlignUp()

◆ EmptySize()

◆ Increment()

◆ MaxSize()

◆ operator[]()

◆ Peek()

◆ PeekBatch()

◆ Pop() [1/3]

◆ Pop() [2/3]

◆ Pop() [3/3]

◆ PopBatch()

◆ PopWithReader()

◆ Push()

◆ PushBatch()

◆ PushWithWriter()

◆ Reset()

◆ Size()

Field Documentation

◆ head_

◆ LENGTH

◆ queue_handle_

◆ tail_