5
\$\begingroup\$

I'm implementing a quick-and-dirty lock-free MPMC queue in C++, based on the v3 Disruptor algorithm. The implementation is pretty simple:

  • Data to be shared between threads is stored in a ring buffer.
  • The ring buffer maintains a sequence number that points to the next available slot.
  • Consumer threads maintain their own sequence numbers indicating their current read position.
  • Producer threads claim a slot in the buffer by atomically incrementing its sequence number. To avoid overwriting data that is yet to be read, the producers block until every consumer has caught up (implementation omitted for brevity).
  • Once a producer has finished writing to its slot, it stores the current queue generation into a second ring buffer to indicate that the data has been published.
  • Consumers block until the published buffer contains the expected generation count, indicating that data is ready to be used.

I'm a bit fuzzy on memory barriers, and wanted to check that I'm using them appropriately. Writes to the data buffer must be completed before corresponding writes to the published buffer. I use a release barrier when writing to the published buffer, and an acquire barrier when reading from it. Is this correct?

template <class T, std::size_t n>
class mpmc_ring_buffer {
  static constexpr auto length = next_power_of_2(n);
  static constexpr auto mask = length - 1;
  static constexpr auto shift = log2(length);

public:
  ring_buffer () {
    for (std::size_t i = 0; i < length; ++i)
        published[i] = -1;
  }

  auto push (T && t) {
    auto seq = sequence.fetch_add(1, std::memory_order_relaxed);
    // TODO: Check for wraparound and spin while consumers catch up
    data[seq & mask] = std::forward<T &&>(t);
    published[seq & mask].store(seq >> shift, std::memory_order_release);
    return seq;
  }

  T peek (std::size_t seq) {
    while (published[seq & mask].load(std::memory_order_acquire) != static_cast<std::uint_fast8_t>(seq >> shift));
    return data[seq & mask];
  }

private:
  std::atomic_size_t sequence alignas(CACHE_LINE_SIZE) {};
  T data alignas(CACHE_LINE_SIZE) [length];
  std::atomic_uint_fast8_t published alignas(CACHE_LINE_SIZE) [length];
};
\$\endgroup\$
  • \$\begingroup\$ You'll receive better reviews if you show a complete example. For example, I recommend that you edit to show the necessary #include lines, and a main() that shows how to call your function. It can really help reviewers if they are able to compile and run your program. \$\endgroup\$ – Toby Speight Feb 6 at 9:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.