Single Producer Single Consumer lockless ring buffer implementation

Question

I am writing a simple ring buffer for my own education. Below is a crack at a strategy described in http://www.cse.cuhk.edu.hk/~pclee/www/pubs/ancs09poster.pdf : Producer and Consumer keep local copies of the write and read indexes on different cache lines and to the extent possible avoid touching the shared versions of the same.

Performance seems to be on par with the boost spspc queue. Any pointers on how to improve this are appreciated. I am using volatile variables rather than std::atomic because, for reasons I do not understand, performance is better with volatile.

Any insight there would be very welcome. I understand that without std::atomic the code might only work on x86_64.

#include <array>
#include <atomic>
#include <limits>
#include <uchar.h>

// Single Producer Single Consumer ring buffer.
// based on http://www.cse.cuhk.edu.hk/~pclee/www/pubs/ancs09poster.pdf

template <class T, size_t MAX_SZ = 10>
class RingBuffer {

   static const size_t cache_line = 64;

 public:
   RingBuffer()
       : // Shared control variables
         shared_r(0), shared_w(0),
         // Consumer state
         consumer_w(0), consumer_r(0),
         // Producer state
         producer_r(0), producer_w(0), uncommited_writes(0) {}

   // Called only by the single producer thread
   // -----------------------------------------
   template <class... ARG>
   bool emplace_enqueue_one(ARG &&... arg) {
      auto result = emplace_enqueue_batch(std::forward<ARG>(arg)...);
      commit_writes();
      return result;
   }

   template <class... ARG>
   bool emplace_enqueue_batch(ARG &&... arg) {

      // Where would the write position be after we enqueue this element?
      size_t next_w = calc_next(producer_w);

      // We always keep an empty slot between the read and write
      // positions, rather than fill our entire buffer. We do this to
      // be able to distinguish between empty (w == r) and full
      // (next(w) == r) buffers. Since we are consulting the
      // producer's copy of the shared read position (producer_r), not
      // the actual read position (shared_r), we might get a false
      // positive (that is we might think we are full when we are not)
      // but not a false negative (that is we think the queue is not
      // full we are right)
      if (next_w == producer_r) {
         // At this point we might be full. To be sure we need to do
         // the more expensive read of the shared read position
         // variable
         size_t actual_r = get_shared_r();
         if (next_w == actual_r) {
            // We are indeed full. At this point we might have to
            // force a commit so that the consumer can see (and drain)
            // uncommited writes.
            commit_writes();
            return false;
         } else
            // We are not actually full, update our local copy of the
            // read position and carry on.
            producer_r = actual_r;
      }

      // Enqueue
      new (&buffer[producer_w]) T(std::forward<ARG>(arg)...);

      // Update our copy of the write position but do not actually
      // update the shared write position. We leave it up to the
      // caller as to when the writes should be visible to the
      // consumer. This allows the caller to amortize the expensive
      // update fo the shared_w variable over multiple writes.
      producer_w = next_w;
      uncommited_writes++;
      return true;
   }

   void commit_writes() {
      if (uncommited_writes) {
         uncommited_writes = 0;
         set_shared_w(producer_w);
      }
   }

   // Called only by the single consumer thread
   // -----------------------------------------
   template <class C>
   size_t consume_one(C &&c) {
      return consume_(std::forward<C>(c), 1);
   }

   template <class C>
   size_t consume_all(C &&c) {
      return consume_(std::forward<C>(c), std::numeric_limits<size_t>::max());
   }

 private:
   template <class C>
   size_t consume_(C c, size_t max_consume_count) {
      size_t consumed_count = 0;
      while (consumed_count < max_consume_count) {
         // Could we be empty?
         if (consumer_w == consumer_r) {
            // We could, but to be sure we have to do the expensive
            // read of the shared write position.
            size_t actual_w = get_shared_w();
            if (consumer_r == actual_w) {
               // We are actually empty. If we managed to read
               // anything so far then update the shared read
               // position.
               if (consumed_count)
                  set_shared_r(consumer_r);
               return consumed_count;
            } else
               // We were not actually empty. Update our copy of the
               // write position. We will do the read below.
               consumer_w = actual_w;
         }
         consumed_count++;
         c(buffer[consumer_r]);
         buffer[consumer_r].~T();
         consumer_r = calc_next(consumer_r);
      }
      // If we reach this point that means we were able to consume
      // max_consume_count items, so we need to update the shared_r
      // position.
      set_shared_r(consumer_r);
      return consumed_count;
   }
   size_t calc_next(size_t p) const {
      if (p < (MAX_SZ - 1))
         return p + 1;
      else
         return 0;
   }

   size_t get_shared_r() { return shared_r; }
   void set_shared_r(size_t r) { shared_r = r; }
   size_t get_shared_w() { return shared_w; }
   void set_shared_w(size_t w) { shared_w = w; }

   // cacheline 1 : shared control variables
   // read position is known to be larger or equal than this
   volatile size_t shared_r;
   // write position is known to be larger or equal than this
   volatile size_t shared_w;
   char padding1[cache_line - 2 * sizeof(size_t)];

   // cacheline 2: consumer state
   size_t consumer_w; // last known write position (to the consumer)
   size_t consumer_r; // current consumer read position
   char padding2[cache_line - 2 * sizeof(size_t)];

   // cacheline 3: producer state
   size_t producer_r;        // last known read position (to the producer)
   size_t producer_w;        // current producer write position
   size_t uncommited_writes; // how far ahead is producer_w from shared_w
   char padding3[cache_line - 3 * sizeof(size_t)];

   // cache line 5: start of actual buffer
   std::array<T, MAX_SZ> buffer;
};

EDIT: In response to feedback I changed the code above to use atomics like this:

   size_t get_shared_r() { return shared_r.load(std::memory_order_acquire); }
   void set_shared_r(size_t r) { shared_r.store(r, std::memory_order_release); }
   size_t get_shared_w() { return shared_w.load(std::memory_order_acquire); }
   void set_shared_w(size_t w) { shared_w.store(w, std::memory_order_release); }

I believe I do not require anything stronger than this (though I could be wrong - I am new to this sort of programming). It is my understanding however, and I have confirmed it by inspecting the generated assembler, that in x86_64, loads always acquire and stores always release. Thus I expected performance to be the same as with the volatile case. Unfortunately this is not the case, the atomic version is roughly twice as slow as the volatile one. I am scratching my head on this one.

Answer 1

Performance is better with volatile because its unsafe as it doesn't generate a memory barrier and thus has a data race and should not be used in at multi threaded context.

As such the code is broken for its intended purpose and could be considered of topic for this site. However I will answer, as this is a common fault at heart of many bugs:

volatile does not mean thread safe or atomic!

Thread safety includes many aspects such as making sure CPU cores see a consistent view of the main memory (e.g. invaliding or updating cache lines cross core) and ensuring that memory operations occur at all (you'd be surprised how much compilers can optimize away, especially clang) and in the right order and atomically. Out of these properties volatile only guarantees order and that the accesses occur. You need memory barriers etc for the rest.

Volatile only means that the compiler will emit all reads and writes to the variables as written, thus ignoring optimization opportunities and not reordering operations to volatile memory.

The only reason ever to use volatile is when you're interfacing hardware such as micro controller registers or device drivers.

I repeat, volatile is almost always wrong AND/OR slow. Unless you're writing drivers or firmware, forget that it ever existed.

Answer 2

std::size_t is consistently misspelt throughout the code. It seems your implementation declares size_t as well as std::size_t (which it is allowed to do), but you must not depend on that.

As Emily says, it's wrong to use volatile where you should use an atomic type.

std::array<T, MAX_SZ> buffer;

If T is expensive to construct, then this is a poor choice. You'll need to allocate uninitialized memory and construct in-place as std::vector does. Actually, the logic is all over the place here, because I see that you do construct in-place, but overwrite existing objects without destructing them. That is totally wrong and dangerous.

template <class T, std::size_t MAX_SZ = 10>

Please use ALL_CAPS naming only for macros. Using for well-behaved C++ identifiers dilutes the warning message that is conveyed by all-caps.

   // cacheline 2: consumer state
   std::size_t consumer_w; // last known write position (to the consumer)
   std::size_t consumer_r; // current consumer read position
   char padding2[cache_line - 2 * sizeof(std::size_t)];

Rather than counting the size of the variables in the cache line, it is simpler to use an anonymous union:

// cacheline 2: consumer state
union {
    char padding2[cache_line];
    struct {
        std::size_t w = 0; // last known write position (to the consumer)
        std::size_t r = 0; // current consumer read position
    } consumer;
};

Similarly for the other cache-aligned blocks. Or consider specifying alignment explicitly using alignas().

Not sure why we have these private members for simple expressions:

   std::size_t get_shared_r();
   void set_shared_r(std::size_t r);
   std::size_t get_shared_w();
   void set_shared_w(std::size_t w);

Just write the expressions directly in the code, rather than cluttering the class with these functions.

Answer 3

A few things in addition to already two great answers.

'Theoretically' this is as fast a single-producer single-consumer ring buffer can can get in a modern processor (larger read/write commit batches takes us closer to a single non-shared memory read/write which is extremely fast).

But your implementation has issues.

1. Using std::array is not only not recommended here but wrong if T is not trivially destructible. You are in-place constructing an element already constructed in std::array. An alternative was to use something like aligned_storage but that's deprecated in C++23 for good reasons. So you can try alignas(alignof(T)) std::byte storage[sizeof(T)*size];

2. Both your reader and writer atomic counters/cursors (shared_r & shared_w) are sharing the cache line (false sharing). This is the worst we can do here as far as throughput is concerned. Place them on different cache lines and throughput will improve dramatically.

3. As far as I understand having uncommitted_writes doesn't give you any performance advantage but adds needless pain to the code-reader. This is accessed in the writer thread right along with accessing shared_w. So you can replace it with

   auto writer_pos = shared_w.load(std::memory_acquire);
   auto uncommited_writes = producer_w - writer_pos;
   

   Why would producer try to commit when there is nothing to commit? What's the point of having if(uncommitted_writes)?

4. Those small helpers if you insist to keep, make them noexcept and/or const where applicable.

5. The whole API design can be improved. That has less to do with C++ but that's a lot to write about.

6.         if (consumer_r == actual_w) { // JUST BREAK out of the while loop here
              // We are actually empty. If we managed to read
              // anything so far then update the shared read
              // position.
              if (consumed_count)
                 set_shared_r(consumer_r);
              return consumed_count;