C++ lock-free, MPMC Ring buffer in C++20

Question

I have some performance critical inter-thread messaging code in C++. Multiple producers, one or more consumers.

Profiling dozens of iterations of this messaging code over several years of development, I'm usually fighting mutex contention. If not a direct mutex on a std::queue, then a heap mutex on the data pointed to by the pointers in whatever concurrent_queue I'm using. If I use big (~256 byte) structs to hold data to the consumers to avoid or minimise heap usage, I end up often memcopy bound from move assignment.

As an attempt to solve all my problems at once, I've written an allocation free, lock free, MPMC ring buffer. Object are produced and consumed without being even moved. No mutexes. No allocations. No memcopies. Only atomic operations.

I've written some simple tests, and implemented it in the performance critical code. Seems to work, and seems to offer a decent performance improvement in the case my boss cares about. I doubt this is as good as I can get it.

I haven't torture tested it. I haven't tested on non MS compilers. I figure this code is hard enough that I should really ask for pointers before getting too deep in.

I am interested in any feedback anyone has.

Extremely trivialised use case example:

  // buffer contains 1 << 8 (256) std::strings
  RingBuffer<std::string, 8> buffer;

  // Producer example
  {
    auto producer = buffer.TryProduce();
    if (!producer)  {... handle if we produce too quickly ...}

    producer->clear(); // We re-use any existing capacity, so there may be a string here.
    producer->push_back('A');
    producer->push_back(.... lots more typically....);

    // Production completes by the destruction of the producer object.
    // As soon as this destructor runs, the std::string can be consumed.
  }
  

  // Consumer example
  {
    auto consumer = buffer.TryConsume();

    if (!consumer) {... handle buffer underrun - ie no instructions in the queue ...}

    if (!consumer->empty() &&
        (*consumer)[0] == 'A')
    {
      // Do something with this instruction that's prefixed with "A".

      // It's safe to read from any other memory in "*consumer" for as long as we have it
      // in scope.
    }

    // When consumer is destroyed, that memory is made available for producers
    // to write to.
  }

Main header file:

#pragma once
///////////////////////////////////////////////////////////////////////////////
//
// Implements an allocation free (after initial construction), lock-free,
// parallel, multi-producer, multi-consumer ring buffer. Any data
// type T is supported, so long as it's default constructible.
//
// The type doesn't even need to be movable / copyable, as the TryProduce
// and TryConsume methods do not force the values to be copied or moved,
// although that is a likely enough case that the TryPushBack and TryPopFront
// helpers are included. Values are typically consumed from the same address 
// they're produced in.
//
// Items added to the end of the buffer and guaranteed to be popped out in
// order, or, more strictly, consumption is guaranteed to commence in the
// same order that production commenced.
//
// Eg (single consumer):
//
//  T 1: |---Produce A--------|   |---Produce B--------|
//  T 2:  |--Produce C -----|   |----produce D------|
//  T 3: idle.................|Consume A|Consume C|.|Consume D|Consume B|
//
//  Or (multiple consumers):
//  T 3: idle.................|Consume A|............|Consume D|
//  T 4: idle..................|Consume C|.............|Consume B|
//
// LGPL - (c) [email protected].
//
///////////////////////////////////////////////////////////////////////////////

#include <atomic>
#include <memory>

namespace rb
{
  template<class T, uint8_t BitsForItemCount> class RingBuffer
  {
  public:
    static_assert(std::is_default_constructible_v<T>);
    static_assert(BitsForItemCount > 0);
    static constexpr uint32_t ItemCount = 1 << BitsForItemCount;
    enum class State : uint8_t;

    struct Consumer;

    struct Producer;

    RingBuffer();

    // Will attempt to read from the buffer. If the buffer was empty, the
    // Consumer will return false in boolean context. While the Consumer is in
    // scope it is guaranteed that the memory your reading will remain
    // untouched and will not be re-used by any producers.
    Consumer TryConsume();

    // Will attempt to reserve a spot in the buffer for you, and if it
    // succeeded, you can take your time populating it (eg with a complex
    // assignment move operation) in safety knowing that no reader will access
    // it until the Producer goes out of scope.
    //
    // If no space in the buffer was available, the producer will
    // return false in a boolean context.
    //
    // There is no way to cancel production once this method has been called,
    // if this is an issue for you A) Don't call it until you know for sure or
    // B) set up a no-op value of T that your consumers will safely skip over.
    //
    // If you try to cancel by letting an exception escape (e.g. your move
    // assignment operator throws), that is very bad, as the T may be left in
    // an invalid state and then submitted to a consumer. This results in an
    // assertion failure in debug if attempted.
    Producer TryProduce();

    // ------------------

    // Simple helpers - it's quite common to want to use this buffer with a
    // type that implements fast move:

    // Moves Item into the write pointer of the ring buffer, if it'll fit.
    // Returns whether the move occurred.
    bool TryPushBack(T&& Item);

    // Moves an item from the front of the buffer to ItemDestination, if there
    // is one to read. Returns whether the move occurred.
    bool TryPopFront(T& ItemDestination);

    enum class State : uint8_t
    {
      Empty,
      Populating,
      Queued,
      Reading
    };

  private:

    static constexpr uint32_t ItemDiv =
      (uint64_t(uint32_t(-1)) + 1) / ItemCount;

    static_assert(
      (ItemDiv & (ItemDiv - 1)) == 0,
      "ItemDiv should always be a power of 2. Otherwise won't wrap correctly.");

    struct Consumer
    {
      explicit operator bool() const { return myValue; }

      auto& Get() const { return *myValue; }
      auto& Get() { return *myValue; }

      operator const T&() const { return Get(); }
      operator T&() { return Get(); }

      T& operator*() { return Get(); }
      const T& operator*() const { return Get(); }

      T* operator->() { return &Get(); }
      const T* operator->() const { return &Get(); }

      Consumer(const Consumer& DontCopy) = delete;
      Consumer(Consumer&& Move);

      Consumer& operator=(const Consumer& DontCopy) = delete;
      Consumer& operator=(Consumer&& Move);

      void Release();

      ~Consumer() { Release(); }
      Consumer() = default;
      Consumer(T* Data, std::atomic_uint8_t* State)
      : myValue(Data), myState(State){};

    private:

      T* myValue = nullptr;
      std::atomic_uint8_t* myState = nullptr;
    };


    struct Producer
    {
      explicit operator bool() const { return myValue; }

      auto& Get() const { return *myValue; }
      auto& Get() { return *myValue; }

      operator const T&() const { return Get(); }
      operator T&() { return Get(); }

      T& operator*() { return Get(); }
      const T& operator*() const { return Get(); }

      T* operator->() { return &Get(); }
      const T* operator->() const { return &Get(); }

      Producer(const Producer& DontCopy) = delete;
      Producer(Producer&& Move);

      Producer& operator=(const Producer& DontCopy) = delete;
      Producer& operator=(Producer&& Move);

      void Release();

      ~Producer();
      Producer() = default;
      Producer(T* Data, std::atomic_uint8_t* State)
      : myValue(Data), myState(State){};

    private:

      T* myValue = nullptr;
      std::atomic_uint8_t* myState = nullptr;
    };

    std::unique_ptr<T[]> myData;
    std::unique_ptr<std::atomic_uint8_t[]> myStates;

    std::atomic_uint32_t myNextRead = 0;
    std::atomic_uint32_t myNextWrite = 0;
  };

}

Template implementations file:

template<class T, uint8_t BC>
  RingBuffer<T, BC>::RingBuffer()
  : myData(std::make_unique<T[]>(ItemCount)),
    myStates(std::make_unique<std::atomic_uint8_t[]>(ItemCount)),
    myNextRead(0),
    myNextWrite(0)
  {
  }

  template<class T, uint8_t BC>
  typename RingBuffer<T, BC>::Consumer RingBuffer<T, BC>::TryConsume()
  {
    uint8_t timeout = 1;
    while (timeout++)
    {
      auto toRead = myNextRead.load(std::memory_order_acquire);
      auto toWrite = myNextWrite.load(std::memory_order_acquire);

      if (toRead == toWrite)
      {
        // Buffer is empty.
        return Consumer(nullptr, nullptr);
      }

      auto readPtr = myData.get() + (toRead / ItemDiv);
      auto statePtr = myStates.get() + (toRead / ItemDiv);

      auto oldState = uint8_t(State::Queued);
      if (statePtr->compare_exchange_strong(
            oldState, uint8_t(State::Reading), std::memory_order_release))
      {
        // We've marked it as reading successfully.

        // Advance the read pointer for the next read. We do it by a large
        // power of two so that it wraps around at the size of the buffer -
        // otherwise we end up having to do a compare exchange if the counter
        // is at end.
        myNextRead.fetch_add(ItemDiv, std::memory_order_release);

        return Consumer(readPtr, statePtr);
      }

      // We were unable to mark the item for "reading" from "queued", that
      // means it was:
      //  - still being populated by a writer.
      //  - given to another consumer on a different thread and the
      //    myNextRead value was incremented by the other thread.
      //
      // Loop back around and try again a few hundred times - otherwise
      // we fail.
    }
    return Consumer(nullptr, nullptr);
  }

  template<class T, uint8_t BC>
  typename RingBuffer<T, BC>::Producer RingBuffer<T, BC>::TryProduce()
  {
    uint8_t timeout = 1;
    while (timeout++)
    {
      auto toRead = myNextRead.load(std::memory_order_acquire);
      auto toWrite = myNextWrite.load(std::memory_order_acquire);

      if (toRead == toWrite + ItemDiv)
      {
        // Buffer is full.
        return Producer(nullptr, nullptr);
      }

      auto writePtr = myData.get() + (toWrite / ItemDiv);
      auto statePtr = myStates.get() + (toWrite / ItemDiv);

      auto oldState = uint8_t(State::Empty);
      if (statePtr->compare_exchange_strong(
            oldState, uint8_t(State::Populating), std::memory_order_release))
      {
        // We've marked it as populating successfully.

        // Advance the write pointer for the next write. We do it by a a
        // large power of two so that it wraps around at the size of the
        // buffer - otherwise we end up having to do a compare exchange if
        // the counter is at end.
        myNextWrite.fetch_add(ItemDiv, std::memory_order_release);

        return Producer(writePtr, statePtr);
      }

      // We were unable to mark the item for "writing" from "empty", that
      // means it was:
      //  - still being read by a reader.
      //  - given to another producer on a different thread and the toWrite
      //    value is about to increment.
      //
      // Loop back around and try again a few hundred times - otherwise
      // we fail.
    }
    return Producer(nullptr, nullptr);
  }


  // Moves Item into the write pointer of the ring buffer, if it'll fit.
  // Returns whether the move occurred.
  template<class T, uint8_t BitsForItemCount>
  inline bool RingBuffer<T, BitsForItemCount>::TryPushBack(T&& Item)
  {
    auto producer = TryProduce();
    if (!producer) return false;
    producer.Get() = std::move(Item);
    return true;
  }

  // Moves an item from the front of the buffer to ItemDestination, if there
  // is one to read. Returns whether the move occurred.
  template<class T, uint8_t BitsForItemCount>
  inline bool RingBuffer<T, BitsForItemCount>::TryPopFront(T& ItemDestination)
  {
    auto consumer = TryConsume();
    if (!consumer) return false;
    ItemDestination = std::move(consumer.Get());
    return true;
  }


  template<class T, uint8_t BitsForItemCount>
  typename RingBuffer<T, BitsForItemCount>::Producer::Producer&
  RingBuffer<T, BitsForItemCount>::Producer::operator=(Producer&& Move)
  {
    Release();
    myValue = Move.myValue;
    myState = Move.myState;
    Move.myValue = nullptr;
    Move.myState = nullptr;
    return *this;
  }


  template<class T, uint8_t BitsForItemCount>
  inline RingBuffer<T, BitsForItemCount>::Producer::Producer(Producer&& Move)
  {
    myValue = Move.myValue;
    myState = Move.myState;
    Move.myValue = nullptr;
    Move.myState = nullptr;
  }
  template<class T, uint8_t BitsForItemCount>
  inline void RingBuffer<T, BitsForItemCount>::Producer::Release()
  {
    if (myValue)
    {
      myState->store(uint8_t(State::Queued), std::memory_order_release);
      myValue = nullptr;
    }
  }
  template<class T, uint8_t BitsForItemCount>
  inline RingBuffer<T, BitsForItemCount>::Producer::~Producer()
  {

    if (dbgN::IsDebugFull() && myValue && std::uncaught_exception())
    {
      ASSERTF_UNREACHABLE(R"(
Exception thrown during a buffer locked for production. Did a move constructor
throw? Why would you do that? This will result in partial data being
transmitted into the buffer and sent to consumers, which will probably
cause issues. (no - we can't rewind the buffer, other producers may of already
started on the next element and we can't break ordering guarentees) Don't use
exceptions to leave the scope! if you really love exceptions and can't do
without them for this tiny region of performance sensitive code - Catch, write
a no-op to the buffer that your consumers will skip over safely, Release(),
and then rethrow)");
    }

    Release();
  }
  template<class T, uint8_t BitsForItemCount>
  inline RingBuffer<T, BitsForItemCount>::Consumer::Consumer(Consumer&& Move)
  {
    myValue = Move.myValue;
    myState = Move.myState;
    Move.myValue = nullptr;
    Move.myState = nullptr;
  }

  template<class T, uint8_t BitsForItemCount>
  typename RingBuffer<T, BitsForItemCount>::Consumer&
  RingBuffer<T, BitsForItemCount>::Consumer::operator=(Consumer&& Move)
  {
    Release();
    myValue = Move.myValue;
    myState = Move.myState;
    Move.myValue = nullptr;
    Move.myState = nullptr;
  }

  template<class T, uint8_t BitsForItemCount>
  inline void RingBuffer<T, BitsForItemCount>::Consumer::Release()
  {
    if (myValue)
    {
      myState->store(uint8_t(State::Empty), std::memory_order_release);
      myValue = nullptr;
    }
  }

Answer 1

I'm mostly going to play advocate for the devil here.

Lock-free doesn't mean fast

There is a rather persistent misconception that lock-free algorithms are faster than locking algorithms. However, that may not be true. Modern mutex implementations are extemely fast in the uncontended case, and when there is a lot of contention they use a system call that lets the kernel wait for the mutex to become unlocked. A system call definitely has a lot of overhead, but your solution is to spin 255 times. Atomic operations are not free, so in the contended case with many threads trying to get access, this might waste a lot of CPU time.

You really should try to prove your theory that a lock-free implementation is faster than one using mutexes by running benchmarks.

Add Produce() and Consume() functions

If the ringbuffer is contended, you spin up to 255 times retrying the operation before giving up, and then you just pass the problem to the caller. The caller does not want to deal with this stuff, and will invariably do something suboptimal, like calling std::this_thread::yield() or std::this_thread::sleep_for(some_random_timeout). Basically, it will either wait too short, wasting more CPU cycles, or will wait too long causing the CPU to be underutilized. Do try to solve this problem, and then implement the solution in class RingBuffer.

Possible solutions are spinning indefinitely but using some kind of exponential backoff (this might even be done without wasting CPU cycles if your processor supports something like the umwait instruction), or using a condition variable to wait after spinning some number of times. There have been lots of efforts over time to find the most optimal way to lock, see for example this LWN article about ticket spinlocks.

About State::Populating

Your design allows a producer to reserve an entry in the ringbuffer, then fill it in at leisure, and then have the guarantee that it can immediately transition that entry to State::Queued. That's perhaps nice for the producers, but not so nice for the consumers. Consider that we have two producers, A and B, and A called TryProduce() first. Now there are two entries in the ringbuffer in state State::Populating. But now suppose that B finishes populating its entry much faster than A. The consumers unfortunately cannot do anything with B's entry, they have to wait for A to finish populating before they can progress. This means that throughput is now limited by the producer that populates its entries the slowest. If you have many producers and the time needed to populate varies a lot, it might end up being slower than just having the producers allocate memory, and have the ringbuffer just store std::unique_ptr<T>s.

There are alternatives; instead of allocating each entry, you could have each producer have a pre-allocated array in which it populates entries. You could take this even further, and have a per-producer ringbuffer, and make TryConsumer() transparently try to pick an element from a producer with a non-empty buffer.

Use if with init-statements

You can make some code more concise by moving variables initialization into if-statements, for example:

if (auto consumer = TryConsume(); consumer) {
   ItemDestination = std::move(consumer.Get());
   return true;
} else {
   return false;
}

Pass size of ringbuffer as std::size_t

Instead of passing uint8_t BitsForItemCount, just use std::size_t ItemCount as the size parameter. This matches the way std::array works, as well as many other fixed-size containers from third-party libraries. You already have a static_assert that will prevent non-POT sizes to be used.

Regardless, also add a static_assert to limit ItemCount to \$2^{32}\$, since myNextRead and myNextWrite are only 32 bits.

Naming things

The names Consumer and Producer are badly chosen; those classes represent "items" or "elements" of the ringbuffer. So perhaps ConsumeItem and ProduceItem would be better, or maybe Consumable/Producable (although the latter sounds weird as well).

Answer 2

  template<class T, uint8_t BitsForItemCount> class RingBuffer

This is missing

#include <cstdint>
using std::uint8_t

That said, I really dislike headers that populate the global namespace, so I would prefer to see std::uint8_t (and its friends) written in full. Is there a reason we need an exactly 8-bit type, or would std::uint_fast8_t be a better choice?

uint8_t timeout = 1;
while (timeout++)

This looks like a for loop. I would prefer a countdown, or perhaps a count up to a constant, rather than leaning on the type overflow to end the loop - explicit is better than implicit.

I had a quick look over the rest, and didn't see anything that jumped out at me, except the many cast of State variables - perhaps suggesting that plain enum : std::uint8_t may be easier than enum class for that type.

Answer 3

Not lock-free

You've replaced locks with busy-waiting, but that doesn't make the algorithm lock-free. In TryConsume, the first consumer to win the race for the to-be-read item is then responsible for advancing the read pointer. Your other consumers wait for this advancement in a busy-loop -- making no progress.

In order for the algorithm to be lock-free, other consumers would either have to steal the item from the first consumer, or declare the item already consumed and consume the next one.

ABA problem

It is possible for multiple consumers to capture the same value of the read pointer and contend for the same item. The idea of "locking" the item by atomically setting its state to Reading after capturing the read pointer doesn't work unfortunately: the item can transition from Queued to Reading to Empty to Populating and back to Queued, all before the first consumer gets to attempt the Queued->Reading transition. At which point it succeeds and returns an item out-of-order.