Hanhan's Blog

可以自己简单实现

只有两个线程，一个用于生产（`push`操作），另一个用于消费（`pop`操作），没有其他线程参与，并且生产者和消费者不会同时对同一个变量（`mFront`或`mRear`）进行并发修改。这种情况下，通常不会出现数据竞争和数据不一致的问题。

原因分析

生产者和消费者的职责明确：
- 生产者线程 只会修改 mRear，即生产者向队列中添加元素时只会改变队列的尾部指针。
- 消费者线程 只会修改 mFront，即消费者从队列中移除元素时只会改变队列的头部指针。
没有竞争条件：
- 在这种设计中，生产者线程和消费者线程不会同时操作同一个队列指针（mFront或mRear），因此不存在两个线程同时访问和修改同一个内存位置的情况。这避免了数据竞争。
逻辑上的同步：
- 由于生产者只在队列未满时写入，消费者只在队列非空时读取，这种操作本身在逻辑上已经形成了同步。例如：
  - 生产者不会覆盖消费者尚未读取的数据，因为它只在队列未满时写入。
  - 消费者不会读取无效数据，因为它只在队列非空时读取。

潜在的风险

内存可见性：
- 在某些平台上，由于内存模型的原因，生产者线程写入的数据可能不会立刻对消费者线程可见，除非有内存屏障或适当的同步机制。这可能会导致消费者读取到旧数据。
缓存一致性：
- 如果处理器使用了缓存，生产者的写入操作可能被缓存，而消费者读取的则是旧数据。虽然在大多数现代处理器上，这种情况不常见，但在某些架构下可能会发生。
编译器优化：
- 编译器优化可能会改变代码的执行顺序，如果没有同步机制，可能导致不可预见的行为。虽然你的代码看起来是正确的，但在某些极端情况下，编译器的优化可能会破坏这种假设。

总结

在特定场景下（一个生产者线程和一个消费者线程），由于它们各自只操作队列的不同部分（一个操作mRear，一个操作mFront），理论上不会出现数据竞争和错误行为。

但是，为了保险起见，在实际应用中通常还是建议使用轻量级的同步机制（如std::atomic）来确保数据的正确性和可见性，尤其是在需要跨平台支持或长时间运行的应用中。

//
// Created by yuyi.di on 2024/4/30.
//
#pragma once
#include "vector"
template<class T>
class RingBuffer {
public:
    RingBuffer(unsigned int size): mSize(size), mFront(0), mRear(0)
    {
        mData.resize(size, 0);
    }
    ~RingBuffer()
    {
        mData.clear();
    }

    inline bool isEmpty() const
    {
        return mFront == mRear;
    }

    inline bool isFull() const
    {
        return mFront == (mRear + 1) % mSize;
    }

    bool push(const T& val)
    {
        if (isFull()) return false;
        mData[mRear] = val;
        mRear = (mRear + 1) % mSize;
        return true;
    }

    bool push(const T* val)
    {
        if (isFull()) return false;
        mData[mRear] = *val;
        mRear = (mRear + 1) % mSize;
        return true;
    }

    inline bool pop(T& val)
    {
        if (isEmpty()) return false;
        val = mData[mFront];
        mFront = (mFront + 1) % mSize;
        return true;
    }

    inline unsigned int front() const
    {
        return mFront;
    }

    inline unsigned int rear() const
    {
        return mRear;
    }

    inline unsigned int size() const
    {
        return mSize;
    }

private:
    unsigned int mSize;
    unsigned int mFront;
    unsigned int mRear;
    std::vector<T> mData;
};

#include <iostream>
#include <thread>
#include "RingBuffer.h"

RingBuffer<int> buffer(5);

void producerFunc()
{
    for (int i = 0; i < 10; ++i)
    {
        while (buffer.isFull()) {}  // 等待缓冲区不满
        buffer.push(i);
        std::cout << "Produced: " << i << std::endl;
    }
}

void consumerFunc()
{
    int val;
    for (int i = 0; i < 10; ++i)
    {
        while (buffer.isEmpty()) {}  // 等待缓冲区非空
        buffer.pop(val);
        std::cout << "Consumed: " << val << std::endl;
    }
}

int main()
{
    std::thread producer(producerFunc);
    std::thread consumer(consumerFunc);

    producer.join();
    consumer.join();

    return 0;
}

Folly::LockFreeRingBuffer实现

1. Cursor类

  // 用于环形缓冲区的不透明指针，可以指向过去或未来的写入位置。
  // 它允许相对移动，但无法在绝对位置上进行移动  
  // 用户可以使用 Cursor 对象在环形缓冲区中轻松导航，查看先前的写入或者预测未来的写入位置  
  struct Cursor {
    explicit Cursor(uint64_t initialTicket) noexcept : ticket(initialTicket) {}

    /// Returns true if this cursor now points to a different
    /// write, false otherwise.
    bool moveForward(uint64_t steps = 1) noexcept {
      uint64_t prevTicket = ticket;
      ticket += steps;
      return prevTicket != ticket;
    }

    /// Returns true if this cursor now points to a previous
    /// write, false otherwise.
    bool moveBackward(uint64_t steps = 1) noexcept {
      uint64_t prevTicket = ticket;
      if (steps > ticket) {
        ticket = 0;
      } else {
        ticket -= steps;
      }
      return prevTicket != ticket;
    }

   protected: // for test visibility reasons
    uint64_t ticket;
    friend class LockFreeRingBuffer;
  };

class LockFreeRingBuffer: boost::noncopyable {
  // 检查模板类型 T 是否具有无异常默认构造函数
  // 如果类型 T 具有无异常默认构造函数，即可以通过 T t; 进行构造且不会抛出异常
  static_assert(
      std::is_nothrow_default_constructible<T>::value,
      "Element type must be nothrow default constructible");
  // 检查类型 T 是否是平凡可复制的类型（trivially copyable）
  // 如果类型 T 是平凡可复制的类型，即可以通过简单的内存复制来实现复制操作，而无需执行用户定义的拷贝构造函数和析构函数
  static_assert(
      folly::is_trivially_copyable<T>::value,
      "Element type must be trivially copyable");

2. Slot类

环形缓冲区槽位的模板类


namespace detail {
template <typename T, template <typename> class Atom>
class RingBufferSlot {
 public:
  explicit RingBufferSlot() noexcept
    : sequencer_()
    , data()
  {
  }

  void write(const uint32_t turn, T& value) noexcept {
    Atom<uint32_t> cutoff(0);
    sequencer_.waitForTurn(turn * 2, cutoff, false);

    // Change to an odd-numbered turn to indicate write in process
    sequencer_.completeTurn(turn * 2);

    data = std::move(value);
    sequencer_.completeTurn(turn * 2 + 1);
    // At (turn + 1) * 2
  }

  bool waitAndTryRead(T& dest, uint32_t turn) noexcept {
    uint32_t desired_turn = (turn + 1) * 2;
    Atom<uint32_t> cutoff(0);
    if (sequencer_.tryWaitForTurn(desired_turn, cutoff, false) !=
        TurnSequencer<Atom>::TryWaitResult::SUCCESS) {
      return false;
    }
    memcpy(&dest, &data, sizeof(T));

    // if it's still the same turn, we read the value successfully
    return sequencer_.isTurn(desired_turn);
  }

  bool tryRead(T& dest, uint32_t turn) noexcept {
    // The write that started at turn 0 ended at turn 2
    if (!sequencer_.isTurn((turn + 1) * 2)) {
      return false;
    }
    memcpy(&dest, &data, sizeof(T));

    // if it's still the same turn, we read the value successfully
    return sequencer_.isTurn((turn + 1) * 2);
  }

 private:
  TurnSequencer<Atom> sequencer_;
  T data;
}; // RingBufferSlot

这个类 RingBufferSlot 是在命名空间 detail 中实现的，它是一个用于环形缓冲区槽位的模板类。这个类具有以下主要功能：

默认构造函数初始化了 sequencer_ 和 data 成员。

write 函数：
- bool waitAndTryRead(T& dest, uint32_t turn) noexcept：这个函数尝试等待并读取环形缓冲区槽位中的数据。
  - 首先计算所需读取的轮次 desired_turn。
  - 使用 TurnSequencer 对象尝试等待指定的轮次（desired_turn）。
  - 如果等待失败，则返回 false；否则使用 memcpy 将数据从 data 复制到目标 dest 槽位中。
  - 最后检查是否仍处于相同的轮次。
tryRead 函数：
- TurnSequencer<Atom> sequencer_：用于管理轮次的辅助对象。
- T data：存储槽位中的数据。

作用：

3. `TurnSequencer`

TurnSequencer 的作用是管理和控制环形缓冲区中数据槽位的访问顺序，确保读取和写入操作按照预定的轮次顺序进行。主要功能包括：

等待特定轮次：TurnSequencer 允许线程等待特定的轮次到来，以确保在正确的时机执行读取或写入操作。
完成轮次：一旦特定轮次到来，TurnSequencer 可以通知相关线程完成当前轮次的操作，以便其他线程可以继续执行后续操作。
检查轮次状态：TurnSequencer 提供了检查当前轮次状态的功能，以确保只有符合条件的线程可以执行读取或写入操作。
协调并发访问：通过使用 TurnSequencer，可以有效地协调多个线程对环形缓冲区槽位的并发访问，避免竞态条件和数据不一致性问题。
提供封装接口：TurnSequencer 将轮次管理的逻辑封装在一个独立的对象中，简化了对环形缓冲区的操作，使得代码更加清晰和可维护。

总的来说，TurnSequencer 在环形缓冲区中扮演着重要的角色，确保数据访问的顺序性和一致性，从而提高系统的并发性能和稳定性。

TurnSequencer 允许线程根据单调递增（带环绕）的“轮次”值排序它们的执行顺序。
提供的两个操作是等待轮次 T 和移动到下一个轮次。
每个正在等待轮次 T 的线程必须在该轮次标记为完成之前到达（对于 MPMCQueue，只有一个线程等待任何特定轮次，因此这显然成立）。
TurnSequencer 的 state_ 保存了当前轮次的 26 位（向左移动 6 位），以及一个 6 位饱和值，
记录了最大等待者减去当前轮次。
预期并处理轮次空间的环绕。
这使我们能够在执行 FUTEX_WAKE 操作时原子地调整未完成等待者的数量。
将此策略与 sem_t 的单独 num_waiters 字段进行比较，后者在等待线程得到调度之后才递减，而在此期间可能已经发生更多的入队操作，使得 FUTEX_WAKE 调用毫无意义。
TurnSequencer 直接使用 futex()。它针对最高等待的轮次比当前轮次高 32 或更少的情况进行了优化。我们使用了 FUTEX_WAIT_BITSET 变体，它允许我们在单个 futex 中嵌入 32 个独立的唤醒通道。请参阅 http://locklessinc.com/articles/futex_cheat_sheet 进行说明。
我们只需要精确跟踪当前轮次与其后的 32 个轮次中最大等待者之间的差值，因为轮次 t+32 处的等待者将在轮次 t 处唤醒。在那时，它们可以使用更高的基数调整差值。由于我们需要编码 0 到 32 （含）的等待者差值，因此我们使用 6 位。实际上，我们存储了高达 63 的等待者差值，因为这可能会稍微减少 CAS 操作的次数。
为了完全避免一些 futex() 调用，TurnSequencer 在等待之前使用自适应自旋截止。分别跟踪每个 TurnSequencer 的自适应自旋截止的开销（和收敛速率）将是不切实际的，因此实际存储作为参数传递并以原子方式更新。这还使调用方能够针对不同操作（例如读取与写入）使用不同的自适应截止值。为了避免争用，当调用方请求时才更新自适应截止值。

RingBuffer

目录

原因分析

潜在的风险

总结

Folly::LockFreeRingBuffer实现

1. Cursor类

2. Slot类

3. `TurnSequencer`

Comments 0

目录

原因分析

潜在的风险

总结

Folly::LockFreeRingBuffer实现

1. Cursor类

2. Slot类

3. TurnSequencer

Comments 0

3. `TurnSequencer`