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

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

文章目录

    • 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 无锁环形队列的特点

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

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

优势

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

局限

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

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 操作流程图

Full
Not Full
Success
Fail
Success
Fail
Start enQueue
Check if Full
Return false
CAS Update Write Index
Write Data
CAS Update Last Write Index
Increment Length
End enQueue
Yield Thread

2.2 Dequeue 操作流程图

Empty
Not Empty
Success
Fail
Start deQueue
Check if Empty
Return false
Read Data from Queue
CAS Update Read Index
Decrement Length
End 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库。然后编译并运行测试。

编译示例:

g++ -std=c++14 lock_free_ring_queue_test.cpp -lgtest -lgtest_main -pthread -o lock_free_ring_queue_test
./lock_free_ring_queue_test

测试结果

$ ./lock_free_ring_queue_test 
[==========] Running 7 tests from 2 test suites.
[----------] Global test environment set-up.
[----------] 5 tests from LockFreeRingQueueTest
[ RUN      ] LockFreeRingQueueTest.Initialization
[       OK ] LockFreeRingQueueTest.Initialization (0 ms)
[ RUN      ] LockFreeRingQueueTest.SingleEnqueueDequeue
[       OK ] LockFreeRingQueueTest.SingleEnqueueDequeue (0 ms)
[ RUN      ] LockFreeRingQueueTest.EnqueueFullQueue
[       OK ] LockFreeRingQueueTest.EnqueueFullQueue(0 ms)
[ RUN      ] LockFreeRingQueueTest.DequeueEmptyQueue
[       OK ] LockFreeRingQueueTest.DequeueEmptyQueue (0 ms)
[ RUN      ] LockFreeRingQueueTest.MultiThreadedEnqueueDequeue
[       OK ] LockFreeRingQueueTest.MultiThreadedEnqueueDequeue (10 ms)
[----------] 5 tests from LockFreeRingQueueTest (10 ms total)[----------] Global test environment tear-down
[==========] 7 tests from 2 test suites ran. (10 ms total)
[  PASSED  ] 7 tests.

6. 总结

无锁环形队列在高并发场景下具有显著的性能优势,其设计充分利用了现代硬件提供的原子操作和内存模型。然而,在实际应用中,开发者需要权衡无锁结构带来的复杂性和潜在的硬件依赖问题。

http://www.lryc.cn/news/426561.html

相关文章:

  • 植物生长时为什么会扭动?科学家解开令查尔斯·达尔文困惑的千古之谜
  • SAP LE学习笔记02 - WM和库存管理(IM)之间的关系,保管Lot(Quant)
  • Span<T> 是 C# 7.2 引入的重要类型
  • Python办公自动化:初识 `openpyxl`
  • Pocketbase实战体验:内置数据库与实时功能如何超越传统MySQL
  • ChatGPT 3.5/4.0 新手使用手册(详细版)
  • 【Java学习】Stream流详解
  • Oracle(69)什么是表压缩(Table Compression)?
  • java JUC编程
  • vue3+element-plus表格分页选中加默认回显选中
  • Erupt 项目搭建
  • HarmonyOS Next 系列之列表下拉刷新和触底加载更多数据实现(十一)
  • 比特位的计算
  • SQLAlchemy 学习笔记
  • Linux内核分析(调度类和调度实体)
  • 用输入输出流(I/O)流,递归复制和删除多级文件
  • kafka监控工具EFAK
  • Page与自定义Components生命周期
  • Chain of Thought (CoT) 系列论文:大模型思维链,提升 LLM 的推理能力
  • 已解决:java.net.BindException: 地址已在使用
  • 看书标记【数据科学:R语言实战 8】
  • STM32标准库学习笔记-1.基础知识
  • Nginx:高效HTTP服务器与反向代理
  • vue3二次封装element-puls
  • 在CentOS 7上安装Apache Tomcat 8的方法
  • 深入理解分布式事务中的三阶段提交(3PC),什么是3PC,3PC原理是怎样?3PC的优化?
  • 这款新的 AI 工具会消灭 ChatGPT 吗?
  • 谷粒商城实战笔记-214~219-商城业务-认证服务-验证码防刷校验
  • 在华为服务器的openEuler系统中适配Pytorch调用NPU
  • MVCC工作原理深入解析