当前位置：首页 > news >正文

C++编程：无锁环形队列 (LockFreeRingQueue)的简单实现、测试和分析

news 文章来源：https://blog.csdn.net/stallion5632/article/details/141270923 2024/9/22 5:36:01

文章目录

- 0. 概述
- 1. 无锁环形队列概述
- - 1.1 无锁环形队列的特点
  - 1.2 无锁环形队列的优势与局限
- 2. `LockFreeRingQueue` 实现
- - 2.1 `Enqueue` 操作流程图
  - 2.2 `Dequeue` 操作流程图
- 3. 核心实现细节
- - 3.1 环形队列的大小调整
  - 3.2 索引计算
  - 3.3 原子操作与CAS
  - 3.4 线程让步
- 4. 测试示例程序
- - 4.1 实现代码
  - 4.2 代码解释：
  - 4.3 运行程序：
  - 4.4 执行结果：
- 5. 单元测试
- - 5.1 测试代码解释：
  - 5.2 测试运行：
- 6. 总结

0. 概述

在现代多线程编程中，高效的并发数据结构对于提升系统性能至关重要，尤其是在处理高并发场景时。本文将详细介绍一种无锁环形队列 (LockFreeRingQueue) 的实现，并探讨其在实际应用中的优势与局限。
本文详细测试代码 lock_free_ring_queue

1. 无锁环形队列概述

无锁环形队列是一种高效的并发数据结构，适用于多线程环境下的生产者-消费者模型。它通过使用原子操作（如CAS操作）来避免锁的使用，从而消除了锁竞争带来的性能开销和潜在的死锁风险。

1.1 无锁环形队列的特点

线程安全：通过原子操作保证数据的正确性。
高效性：避免了锁竞争，减少了线程上下文切换的开销。
避免ABA问题：设计时特别考虑了ABA问题的影响，通过合理的索引管理避免了这一问题。

1.2 无锁环形队列的优势与局限

优势：

高并发性：无锁结构通过避免锁的使用，使多个线程可以并发执行，提高了吞吐量。
低延迟：在高并发场景下，无锁结构减少了线程竞争带来的上下文切换，降低了系统延迟。
避免死锁：由于没有锁的存在，自然避免了死锁问题。

局限：

复杂性：无锁结构通常比锁机制实现更复杂，容易引入难以调试的并发错误。
硬件依赖性：原子操作（如CAS）通常依赖于底层硬件支持，在不同平台上表现可能有所不同。
有限应用场景：无锁队列并不适合所有场景，在某些情况下（如低并发或非实时系统），传统的锁机制可能更为适合。

2. `LockFreeRingQueue` 实现

下面是一个基于C++实现的无锁环形队列的实现，该队列支持多生产者和多消费者线程的并发访问。

#ifndef RING_QUEUE_HPP
#define RING_QUEUE_HPP#include <atomic>
#include <memory>
#include <stdexcept>
#include <thread>template <typename T>
class LockFreeRingQueue {public:// Constructor that initializes the ring queue with the specified sizeexplicit LockFreeRingQueue(uint32_t size);// Default destructor~LockFreeRingQueue() = default;// Disable copy constructor and assignment operatorLockFreeRingQueue(const LockFreeRingQueue&) = delete;LockFreeRingQueue& operator=(const LockFreeRingQueue&) = delete;// Enqueue operation to add an element to the queuebool Enqueue(const T& data);// Dequeue operation to remove an element from the queuebool Dequeue(T* data);// Check if the queue is emptybool IsEmpty() const noexcept;// Check if the queue is fullbool IsFull() const noexcept;// Get the current size of the queueuint32_t Size() const noexcept;private:// Check if the given number is a power of twostatic bool IsPowerOfTwo(uint32_t num) noexcept;// Calculate the ceiling power of two greater than or equal to the given numberstatic uint32_t CeilPowerOfTwo(uint32_t num) noexcept;// Round up the given number to the nearest power of twostatic uint32_t RoundupPowerOfTwo(uint32_t num) noexcept;// Get the index within the queueuint32_t IndexOfQueue(uint32_t index) const noexcept;private:const uint32_t size_;                     // Size of the queue, must be a power of twostd::atomic<uint32_t> length_;            // Current length of the queuestd::atomic<uint32_t> read_index_;        // Index for the consumer to readstd::atomic<uint32_t> write_index_;       // Index for the producer to writestd::atomic<uint32_t> last_write_index_;  // Last confirmed write indexstd::unique_ptr<T[]> queue_;              // Array to store the queue elements
};template <typename T>
LockFreeRingQueue<T>::LockFreeRingQueue(uint32_t size): size_(size <= 1U           ? 2U: IsPowerOfTwo(size) ? size: RoundupPowerOfTwo(size)),length_(0U),read_index_(0U),write_index_(0U),last_write_index_(0U),queue_(std::make_unique<T[]>(size_)) {if (size == 0U) {throw std::out_of_range("Queue size must be greater than 0");}
}template <typename T>
bool LockFreeRingQueue<T>::Enqueue(const T& data) {uint32_t current_read_index;uint32_t current_write_index;do {current_read_index = read_index_.load(std::memory_order_relaxed);current_write_index = write_index_.load(std::memory_order_relaxed);// Check if the queue is fullif (IndexOfQueue(current_write_index + 1U) == IndexOfQueue(current_read_index)) {return false;  // Queue is full}} while (!write_index_.compare_exchange_weak(current_write_index, current_write_index + 1U, std::memory_order_release,std::memory_order_relaxed));queue_[IndexOfQueue(current_write_index)] = data;// Confirm the write operationwhile (!last_write_index_.compare_exchange_weak(current_write_index, current_write_index + 1U,std::memory_order_release, std::memory_order_relaxed)) {std::this_thread::yield();  // Yield CPU to avoid busy-waiting}length_.fetch_add(1U, std::memory_order_relaxed);return true;
}template <typename T>
bool LockFreeRingQueue<T>::Dequeue(T* data) {if (data == nullptr) {throw std::invalid_argument("Null pointer passed to Dequeue");}uint32_t current_read_index;uint32_t current_last_write_index;do {current_read_index = read_index_.load(std::memory_order_relaxed);current_last_write_index = last_write_index_.load(std::memory_order_relaxed);// Check if the queue is emptyif (IndexOfQueue(current_last_write_index) == IndexOfQueue(current_read_index)) {return false;  // Queue is empty}*data = queue_[IndexOfQueue(current_read_index)];if (read_index_.compare_exchange_weak(current_read_index, current_read_index + 1U, std::memory_order_release,std::memory_order_relaxed)) {length_.fetch_sub(1U, std::memory_order_relaxed);return true;}} while (true);
}template <typename T>
bool LockFreeRingQueue<T>::IsEmpty() const noexcept {return length_.load(std::memory_order_relaxed) == 0U;
}template <typename T>
bool LockFreeRingQueue<T>::IsFull() const noexcept {uint32_t next_write_index = IndexOfQueue(write_index_.load(std::memory_order_relaxed) + 1U);return next_write_index == read_index_.load(std::memory_order_acquire);
}template <typename T>
uint32_t LockFreeRingQueue<T>::Size() const noexcept {return length_.load(std::memory_order_relaxed);
}template <typename T>
bool LockFreeRingQueue<T>::IsPowerOfTwo(uint32_t num) noexcept {return (num != 0U) && ((num & (num - 1U)) == 0U);
}template <typename T>
uint32_t LockFreeRingQueue<T>::CeilPowerOfTwo(uint32_t num) noexcept {num |= (num >> 1U);num |= (num >> 2U);num |= (num >> 4U);num |= (num >> 8U);num |= (num >> 16U);return num - (num >> 1U);
}template <typename T>
uint32_t LockFreeRingQueue<T>::RoundupPowerOfTwo(uint32_t num) noexcept {return CeilPowerOfTwo((num - 1U) << 1U);
}template <typename T>
uint32_t LockFreeRingQueue<T>::IndexOfQueue(uint32_t index) const noexcept {return index & (size_ - 1U);
}#endif  // RING_QUEUE_HPP

2.1 `Enqueue` 操作流程图

2.2 `Dequeue` 操作流程图

3. 核心实现细节

3.1 环形队列的大小调整

环形队列的大小通常为2的幂次，以便可以通过位运算快速计算索引。RoundUpPowerOfTwo 函数将任意正整数向上调整为最接近的2的幂次。

确保环形队列的大小是 2 的幂次方主要有以下几个原因:

简化索引计算:
- 在环形队列中,元素的位置通过取模运算来计算。
- 如果队列的大小是 2 的幂次方,那么取模运算可以被位运算 & 替代,这个操作更加高效。
避免对齐问题:
- 对于某些硬件架构来说,内存访问的效率会受到内存对齐的影响。
- 如果队列大小是 2 的幂次方,那么它就能很好地符合内存对齐要求,从而提高访问效率。
位操作优化:
- 很多处理器都针对 2 的幂次方大小的数据提供了优化的指令集。
- 比如,判断一个数是否是 2 的幂次方可以用位操作 (num & (num - 1)) == 0 来实现,这比普通的除法要高效得多。
缓存友好性:
- 当队列大小是 2 的幂次方时,元素在内存中的分布更加规则有序。
- 这有利于利用缓存,减少缓存未命中的情况,提高整体性能。

3.2 索引计算

通过IndexOfQueue 函数，索引计算可以使用位与操作（&）而不是取模运算（%），从而提升计算速度。

3.3 原子操作与CAS

通过 std::atomic_compare_exchange_weak 原子操作实现无锁队列的核心逻辑。CAS（Compare-And-Swap）是无锁数据结构的基础，用于确保多个线程在修改共享数据时不会引发数据竞争。

3.4 线程让步

在 Enqueue 操作中，当多个线程尝试修改相同的共享变量时，失败的线程可以选择让出CPU时间片，以减少竞争和等待时间。

4. 测试示例程序

下面是一个简单的C++示例程序，演示了如何使用 LockFreeRingQueue 类。该程序将进行一些基本的入队和出队操作，然后在多线程环境下测试队列的使用。

4.1 实现代码

#include <iostream>
#include <thread>#include "lock_free_ring_queue.h"  // 假设你的类定义在这个文件中void SingleProducerSingleConsumerExample() {LockFreeRingQueue<int> queue(4);// 单生产者线程入队std::thread producer([&queue]() {for (int i = 0; i < 4; ++i) {if (queue.Enqueue(i)) {std::cout << "Producer enqueued: " << i << std::endl;} else {std::cout << "Queue is full, cannot enqueue: " << i << std::endl;}}});// 单消费者线程出队std::thread consumer([&queue]() {int value = 0;for (int i = 0; i < 4; ++i) {while (!queue.Dequeue(&value)) {std::this_thread::yield();  // 等待队列中有数据}std::cout << "Consumer dequeued: " << value << std::endl;}});producer.join();consumer.join();
}void MultiProducerMultiConsumerExample() {LockFreeRingQueue<int> queue(8);auto producer = [&queue](int id) {for (int i = 0; i < 4; ++i) {int value = id * 10 + i;if (queue.Enqueue(value)) {std::cout << "Producer " << id << " enqueued: " << value << std::endl;} else {std::cout << "Queue is full, Producer " << id << " cannot enqueue: " << value << std::endl;}}};auto consumer = [&queue](int id) {int value = 0;for (int i = 0; i < 4; ++i) {while (!queue.Dequeue(&value)) {std::this_thread::yield();  // 等待队列中有数据}std::cout << "Consumer " << id << " dequeued: " << value << std::endl;}};std::thread producers[2];std::thread consumers[2];// 启动两个生产者线程for (int i = 0; i < 2; ++i) {producers[i] = std::thread(producer, i);}// 启动两个消费者线程for (int i = 0; i < 2; ++i) {consumers[i] = std::thread(consumer, i);}for (auto &producer : producers) {producer.join();}for (auto &consumer : consumers) {consumer.join();}
}int main() {std::cout << "Single Producer, Single Consumer Example:" << std::endl;SingleProducerSingleConsumerExample();std::cout << "\nMulti Producer, Multi Consumer Example:" << std::endl;MultiProducerMultiConsumerExample();return 0;
}

4.2 代码解释：

SingleProducerSingleConsumerExample:
- 创建了一个大小为 4 的队列。
- 使用一个生产者线程将数字 0-3 入队。
- 使用一个消费者线程从队列中出队并打印结果。
MultiProducerMultiConsumerExample:
- 创建了一个大小为 8 的队列。
- 启动两个生产者线程，每个线程将其线程 ID 和循环计数拼接成值并入队。
- 启动两个消费者线程，从队列中出队并打印结果。

4.3 运行程序：

将上面的代码保存为 main.cpp，并确保你已经编译和链接了 LockFreeRingQueue 类的实现。

编译示例：

g++ -std=c++14 main.cpp -pthread -o lock_free_ring_queue_example
./lock_free_ring_queue_example

4.4 执行结果：

$ ./lock_free_ring_queue_example 
Single Producer, Single Consumer Example:
Producer enqueued: Consumer dequeued: 0
0
Producer enqueued: 1
Producer enqueued: 2
Producer enqueued: 3
Consumer dequeued: 1
Consumer dequeued: 2
Consumer dequeued: 3Multi Producer, Multi Consumer Example:
Producer 0 enqueued: 0
Producer 0 enqueued: 1
Producer 0 enqueued: 2
Producer 0 enqueued: 3
Consumer 1 dequeued: 0
Consumer 1 dequeued: 1
Consumer 1 dequeued: 2
Consumer 1 dequeued: 3
Producer 1 enqueued: 10
Producer 1 enqueued: 11
Producer 1 enqueued: 12
Producer 1 enqueued: 13
Consumer 0 dequeued: 10
Consumer 0 dequeued: 11
Consumer 0 dequeued: 12
Consumer 0 dequeued: 13

5. 单元测试

下面是使用gtest编写的LockFreeRingQueue类的完整单元测试。测试涵盖了基本功能，包括队列的初始化、入队、出队、边界条件，以及在多线程环境下的行为。

#include "lock_free_ring_queue.h"  // 假设你的类定义在这个文件中
#include <gtest/gtest.h>#include <algorithm>
#include <memory>
#include <thread>
#include <vector>class LockFreeRingQueueTest : public ::testing::Test {protected:void SetUp() override {// 初始化队列大小queue_size_ = 64;queue_ = std::make_unique<LockFreeRingQueue<int>>(queue_size_);}std::unique_ptr<LockFreeRingQueue<int>> queue_;uint32_t queue_size_;
};// 测试队列初始化
TEST_F(LockFreeRingQueueTest, Initialization) {EXPECT_EQ(queue_->Size(), 0U);EXPECT_TRUE(queue_->IsEmpty());
}// 测试入队和出队单个元素
TEST_F(LockFreeRingQueueTest, SingleEnqueueDequeue) {int value_in = 42;int value_out = 0;EXPECT_TRUE(queue_->Enqueue(value_in));EXPECT_EQ(queue_->Size(), 1U);EXPECT_FALSE(queue_->IsEmpty());EXPECT_TRUE(queue_->Dequeue(&value_out));EXPECT_EQ(value_out, value_in);EXPECT_EQ(queue_->Size(), 0U);EXPECT_TRUE(queue_->IsEmpty());
}// 测试队列满时入队
TEST_F(LockFreeRingQueueTest, EnqueueFullQueue) {for (uint32_t i = 0; i < queue_size_ - 1; ++i) {  // 注意减1EXPECT_TRUE(queue_->Enqueue(static_cast<int>(i)));}EXPECT_EQ(queue_->Size(), queue_size_ - 1);EXPECT_FALSE(queue_->Enqueue(100));  // 队列已满，入队失败
}// 测试空队列出队
TEST_F(LockFreeRingQueueTest, DequeueEmptyQueue) {int value_out = 0;EXPECT_FALSE(queue_->Dequeue(&value_out));  // 队列为空，出队失败
}// 多线程测试
TEST_F(LockFreeRingQueueTest, MultiThreadedEnqueueDequeue) {const int num_threads = 4;const int num_elements_per_thread = 10;auto enqueue_function = [&](int thread_id) {for (int i = 0; i < num_elements_per_thread; ++i) {queue_->Enqueue(thread_id * num_elements_per_thread + i);}};auto dequeue_function = [&](int thread_id, int* result_array) {for (int i = 0; i < num_elements_per_thread; ++i) {int value = 0;while (!queue_->Dequeue(&value)) {std::this_thread::yield();}result_array[thread_id * num_elements_per_thread + i] = value;}};std::vector<std::thread> threads;int results[num_threads * num_elements_per_thread] = {0};for (int i = 0; i < num_threads; ++i) {threads.emplace_back(enqueue_function, i);}for (auto& thread : threads) {thread.join();}threads.clear();for (int i = 0; i < num_threads; ++i) {threads.emplace_back(dequeue_function, i, results);}for (auto& thread : threads) {thread.join();}EXPECT_EQ(queue_->Size(), 0U);EXPECT_TRUE(queue_->IsEmpty());// 检查所有元素是否都已被成功出队std::sort(std::begin(results), std::end(results));for (int i = 0; i < num_threads * num_elements_per_thread; ++i) {EXPECT_EQ(results[i], i);}
}// 边界条件测试：初始化大小为1的队列
TEST(LockFreeRingQueueBoundaryTest, InitializationWithSizeOne) {LockFreeRingQueue<int> small_queue(1);EXPECT_EQ(small_queue.Size(), 0U);EXPECT_TRUE(small_queue.IsEmpty());int value_in = 99;EXPECT_TRUE(small_queue.Enqueue(value_in));EXPECT_FALSE(small_queue.Enqueue(value_in));  // 队列应该已经满了
}// 边界条件测试：入队和出队仅一个元素
TEST(LockFreeRingQueueBoundaryTest, SingleElementQueue) {LockFreeRingQueue<int> small_queue(1);int value_in = 123;int value_out = 0;EXPECT_TRUE(small_queue.Enqueue(value_in));EXPECT_FALSE(small_queue.Enqueue(value_in));  // 队列已满EXPECT_TRUE(small_queue.Dequeue(&value_out));EXPECT_EQ(value_out, value_in);EXPECT_FALSE(small_queue.Dequeue(&value_out));  // 队列为空
}int main(int argc, char** argv) {::testing::InitGoogleTest(&argc, argv);return RUN_ALL_TESTS();
}

5.1 测试代码解释：

基本功能测试：
- Initialization: 检查队列是否正确初始化。
- SingleEnqueueDequeue: 测试单个元素的入队和出队操作。
- EnqueueFullQueue: 测试在队列已满时的入队操作。
- DequeueEmptyQueue: 测试在队列为空时的出队操作。
多线程测试：
- MultiThreadedEnqueueDequeue: 使用多个线程测试队列的入队和出队操作。每个线程分别执行入队和出队操作，最后检查所有元素是否正确出队。
边界条件测试：
- InitializationWithSizeOne: 测试初始化大小为1的队列。
- SingleElementQueue: 测试大小为1的队列的入队和出队操作。

5.2 测试运行：

将上面的代码保存为一个测试文件（例如lock_free_ring_queue_test.cpp），并确保你已经安装了gtest库。然后编译并运行测试。