无锁队列实现 / 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

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

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 25 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.

Definition at line 40 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 54 of file lockfree_queue.hpp.

54{ 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 27 of file lockfree_queue.hpp.

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

◆ EmptySize()

template<typename Data >

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

inline

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

Definition at line 373 of file lockfree_queue.hpp.

373{ return LENGTH - Size(); }

LibXR::LockFreeQueue::Size

size_t Size() const

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

Definition lockfree_queue.hpp:362

◆ Increment()

template<typename Data >

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

inlineprivate

Definition at line 381 of file lockfree_queue.hpp.

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

◆ 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 61 of file lockfree_queue.hpp.

61{ 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 196 of file lockfree_queue.hpp.

  {
    const auto CURRENT_HEAD = head_.load(std::memory_order_acquire);
    if (CURRENT_HEAD == tail_.load(std::memory_order_acquire))
    {
      return ErrorCode::EMPTY;
    }
 
    item = queue_handle_[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 308 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;
      }
 
      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 167 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_acquire,
                                      std::memory_order_relaxed))
      {
        return ErrorCode::OK;
      }
      current_head = head_.load(std::memory_order_relaxed);
    }
  }

◆ 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).

Note: 使用 std::atomic_thread_fence(std::memory_order_acquire) 确保正确读取数据，防止 CPU 乱序执行带来的数据读取问题。 (Uses std::atomic_thread_fence(std::memory_order_acquire) to ensure proper data reading and prevent CPU reordering issues).

Definition at line 137 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_acquire,
                                      std::memory_order_relaxed))
      {
        std::atomic_thread_fence(std::memory_order_acquire);
        item = queue_handle_[current_head];
        return ErrorCode::OK;
      }
      current_head = head_.load(std::memory_order_relaxed);
    }
  }

◆ 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 94 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_acquire,
                                      std::memory_order_relaxed))
      {
        item = queue_handle_[current_head];
        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 255 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_acquire,
                                      std::memory_order_relaxed))
      {
        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 71 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 216 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;
  }

◆ 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 352 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 362 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 376 of file lockfree_queue.hpp.

◆ LENGTH

template<typename Data >

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

private

Definition at line 378 of file lockfree_queue.hpp.

◆ queue_handle_

template<typename Data >

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

private

Definition at line 379 of file lockfree_queue.hpp.

◆ tail_

template<typename Data >

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

private

Definition at line 377 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()

◆ operator[]()

◆ Peek()

◆ PeekBatch()

◆ Pop() [1/3]

◆ Pop() [2/3]

◆ Pop() [3/3]

◆ PopBatch()

◆ Push()

◆ PushBatch()

◆ Reset()

◆ Size()

Field Documentation

◆ head_

◆ LENGTH

◆ queue_handle_

◆ tail_