我有一个类存储一些传入实时数据的最新值(大约 1.5 亿个事件/秒)。

假设它看起来像这样:

class DataState 
{
    Event latest_event;

  public:
  //pushes event atomically
  void push_event(const Event __restrict__* e);
  //pulls event atomically
  Event pull_event();
};

我需要能够以原子方式推送事件并以严格的顺序保证拉取它们。现在，我知道我可以使用自旋锁，但考虑到大量事件发生率(超过 1 亿/秒)和高度并发，我更愿意使用无锁操作。

问题是 Event大小为 64 字节。没有 CMPXCHG64B任何当前 X86 CPU 上的指令(截至 2016 年 8 月)。所以如果我使用 std::atomic<Event>我必须链接到 libatomic它在引擎盖下使用互斥体(太慢)。

所以我的解决方案是原子地交换指向值的指针。问题是动态内存分配成为这些事件发生率的瓶颈。所以...我定义了一个我称之为“环分配器”的东西:

/// @brief Lockfree Static short-lived allocator used for a ringbuffer
/// Elements are guaranteed to persist only for "size" calls to get_next()
template<typename T> class RingAllocator {
  T *arena;
  std::atomic_size_t arena_idx;
  const std::size_t arena_size;
 public:
  /// @brief Creates a new RingAllocator
  /// @param size The number of elements in the underlying arena. Make this large enough to avoid overwriting fresh data
  RingAllocator<T>(std::size_t size) : arena_size(size)
  {
  //allocate pool
  arena = new T[size];
  //zero out pool
  std::memset(arena, 0, sizeof(T) * size);
  arena_idx = 0;
  }

  ~RingAllocator()
  {
  delete[] arena;
  }

  /// @brief Return next element's pointer. Thread-safe
  /// @return pointer to next available element
  T *get_next()
  {
      return &arena[arena_idx.exchange(arena_idx++ % arena_size)];
  }
};

然后我可以让我的 DataState 类看起来像这样:

class DataState 
{
    std::atomic<Event*> latest_event;
    RingAllocator<Event> event_allocator;
  public:
  //pushes event atomically
  void push_event(const Event __restrict__* e)
  {
      //store event
      Event *new_ptr = event_allocator.get_next()
      *new_ptr = *e;
      //swap event pointers
      latest_event.store(new_ptr, std::memory_order_release);
  }
  //pulls event atomically
  Event pull_event()
  {
      return *(latest_event.load(std::memory_order_acquire));
  }
};

只要我将环分配器的大小设置为可以并发调用函数的最大线程数，就不会有覆盖 pull_event 可能返回的数据的风​​险。此外，一切都 super 本地化，因此间接不会导致缓存性能不佳。这种方法有什么可能的缺陷吗？

最佳答案

DataState 类:

我以为它会是一个堆栈或队列，但事实并非如此，所以 push/pull 看起来不是很好的方法名称。 (否则实现完全是伪造的)。

它只是一个闩锁，可让您读取任何线程存储的最后一个事件。

没有什么可以阻止连续两次写入覆盖从未被读取的元素。也没有什么可以阻止您阅读同一个元素两次。

如果您只需要在某个地方复制小块数据，环形缓冲区似乎是一种不错的方法。但是如果你不想失去事件，我认为你不能这样使用它。相反，只需获取一个环形缓冲区条目，然后复制到它并在那里使用它。所以唯一的原子操作应该是递增环形缓冲区位置索引。

环形缓冲区

您可以使 get_next() 更有效率。此行执行原子后增量 (fetch_add) 和原子交换:

return &arena[arena_idx.exchange(arena_idx++ % arena_size)];

我什至不确定它是否安全，因为 xchg 可能会从另一个线程踩到 fetch_add。无论如何，即使它是安全的，也不是理想的。

你不需要那个。确保 arena_size 始终是 2 的幂，那么您就不需要对共享计数器取模。你可以放手，让每个线程取模供自己使用。它最终会换行，但它是一个二进制整数，所以它会以 2 的幂换行，这是你的竞技场大小的倍数。

我建议存储一个 AND 掩码而不是一个大小，这样就没有 % 编译成 and 指令以外的任何东西的风险，即使它是不是编译时常量。这确保我们避免使用 64 位整数 div 指令。

template<typename T> class RingAllocator {
  T *arena;
  std::atomic_size_t arena_idx;
  const std::size_t size_mask;   // maybe even make this a template parameter?
 public:
  RingAllocator<T>(std::size_t size) 
    : arena_idx(0),  size_mask(size-1)
  {
     // verify that size is actually a power of two, so the mask is all-ones in the low bits, and all-zeros in the high bits.
     // so that i % size == i & size_mask for all i
   ...
  }

  ...
  T *get_next() {
      size_t idx = arena_idx.fetch_add(1, std::memory_order_relaxed);  // still atomic, but we don't care which order different threads take blocks in
      idx &= size_mask;   // modulo our local copy of the idx
      return &arena[idx];
  }
};

如果您使用 calloc 而不是 new + memset，那么分配 arena 会更有效率。操作系统在将它们提供给用户空间进程之前已经将页面归零(以防止信息泄漏)，因此将它们全部写入只是浪费工作。

  arena = new T[size];
  std::memset(arena, 0, sizeof(T) * size);

  // vs.

  arena = (T*)calloc(size, sizeof(T));

自己编写页面确实会导致它们出错，因此它们都连接到真实的物理页面，而不只是系统范围共享物理零页面的写时复制映射(就像它们在 new/malloc/调用)。在 NUMA 系统上，选择的物理页面可能取决于哪个线程实际接触了该页面，而不是哪个线程进行了分配。但是由于您要重用池，第一个编写页面的核心可能不是最终使用它最多的核心。

也许可以在微基准测试/性能计数器中寻找一些东西。

关于c++ - 用于成员变量无锁更新的环形分配器？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/39195250/