2. 环形缓冲区

任务目标

实现一个基于C++11的无锁单生产者-单消费者环形缓冲区，用于嵌入式环境下的数据传递，并编写吞吐量测试程序验证其性能。

核心实现

// ring_buffer.hpp
#pragma once
#include <atomic>
#include <cstddef>
#include <cstdint>

template <typename T, size_t Capacity>
class RingBuffer {
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be power of 2");
    static constexpr size_t MASK = Capacity - 1;
    
    alignas(64) std::atomic<size_t> head_{0};  // 写索引
    alignas(64) std::atomic<size_t> tail_{0};  // 读索引
    alignas(64) T buffer_[Capacity];
    
public:
    bool push(const T& item) {
        size_t h = head_.load(std::memory_order_relaxed);
        size_t t = tail_.load(std::memory_order_acquire);
        if ((h - t) == Capacity) return false;  // 满
        buffer_[h & MASK] = item;
        head_.store(h + 1, std::memory_order_release);
        return true;
    }
    
    bool pop(T& item) {
        size_t t = tail_.load(std::memory_order_relaxed);
        size_t h = head_.load(std::memory_order_acquire);
        if (t == h) return false;  // 空
        item = buffer_[t & MASK];
        tail_.store(t + 1, std::memory_order_release);
        return true;
    }
    
    size_t size() const {
        return head_.load(std::memory_order_relaxed) - 
               tail_.load(std::memory_order_relaxed);
    }
    
    bool empty() const { return size() == 0; }
    bool full() const { return size() == Capacity; }
};

吞吐量测试

// benchmark.cpp
#include "ring_buffer.hpp"
#include <chrono>
#include <cstdio>
#include <thread>

struct alignas(64) Payload { uint32_t id; uint32_t data; };

int main() {
    constexpr size_t CAP = 1024;
    constexpr size_t TOTAL = 1000000;
    RingBuffer<Payload, CAP> rb;
    std::atomic<bool> done{false};
    
    // 生产者线程
    std::thread producer([&] {
        for (size_t i = 0; i < TOTAL; ++i) {
            Payload p{static_cast<uint32_t>(i), static_cast<uint32_t>(i * 3)};
            while (!rb.push(p)) { std::this_thread::yield(); }
        }
        done.store(true, std::memory_order_release);
    });
    
    // 消费者线程
    size_t received = 0;
    auto start = std::chrono::high_resolution_clock::now();
    std::thread consumer([&] {
        Payload p;
        while (true) {
            if (rb.pop(p)) {
                ++received;
                if (received == TOTAL) break;
            } else if (done.load(std::memory_order_acquire) && rb.empty()) {
                break;
            } else {
                std::this_thread::yield();
            }
        }
    });
    
    producer.join();
    consumer.join();
    auto end = std::chrono::high_resolution_clock::now();
    
    double elapsed_ms = std::chrono::duration<double, std::milli>(end - start).count();
    double throughput = TOTAL / (elapsed_ms / 1000.0) / 1000000.0;  // Mops/s
    
    std::printf("Received: %zu / %zu\n", received, TOTAL);
    std::printf("Elapsed: %.2f ms\n", elapsed_ms);
    std::printf("Throughput: %.2f Mops/s\n", throughput);
    return 0;
}

编译与运行

g++ -O2 -std=c++11 -pthread benchmark.cpp -o benchmark
./benchmark

验证方法

  • 正确性验证：程序输出 Received: 1000000 / 1000000，表示所有数据均被成功消费，无丢失、无重复。
  • 性能验证：观察吞吐量数值。在普通x86 Linux环境下，O2优化级别通常可达数十Mops/s量级；若数值异常低（如<1 Mops/s），需检查是否因缓存行伪共享或频繁yield导致。
  • 无锁验证：代码中无任何互斥锁（mutex）或原子自旋锁，仅使用std::atomic的无锁操作（在主流架构上对size_t为无锁）。

关键设计说明

  • 容量必须为2的幂：通过位与运算 & MASK 替代取模，提升效率。
  • 内存序：push 使用 release 保证数据写入先于索引更新；pop 使用 acquire 保证索引读取先于数据读取，避免指令重排导致的数据竞争。
  • 缓存行对齐：head_ 与 tail_ 分别以 alignas(64) 对齐，避免伪共享（false sharing）导致的性能下降。
  • 适用场景：仅支持单生产者-单消费者模型。多生产者/多消费者需要额外机制（如CAS循环或分段锁），不在本案例范围内。

常见问题

  • 吞吐量偏低：检查是否开启了 -O2 优化；确认CPU核心数足够（生产者和消费者线程可并行）；尝试增大 TOTAL 以减少线程调度开销占比。
  • 死锁或卡死：确认消费者退出条件正确——当生产者完成且缓冲区为空时退出，否则可能因空转yield导致CPU占用过高但无法退出。
  • 数据错乱：检查内存序是否完整；在弱内存序架构（如ARM）上，缺失 acquire/release 会导致可见性问题。
← 上一章 📖 返回目录 下一章 →