# Lock-free ringbuffer with multiple readers in C++11

Basic Info

I needed a lock-free ringbuffer with multiple readers (but one writer). However, I did not want the writer to check all readers every time in order to prevent an overrun. Thus, I decided to split the buffer in two halves. If the write pointer is in one half, it is only allowed to advance to the next half if all readers are already in the same half as the writer.

This is realized by using an atomic counter named readers_left, which counts the number of readers left in the previous half.

The code to this question can be found here and the most recent version here. In the following, I'll try to explain the code bit by bit.

Classes

This class is the base for a reader and writer:

class ringbuffer_common_t
{
private:
static std::size_t calc_size(std::size_t sz);
protected:
const std::size_t size; //!< buffer size (2^n for some n)
const std::size_t size_mask; //!< = size - 1
public:
ringbuffer_common_t(std::size_t sz);
};


Next comes the writer, just called ringbuffer_t. It contains the only two atomics:

class ringbuffer_t : protected ringbuffer_common_t
{
std::atomic<std::size_t> w_ptr; //!< writer at buf[w_ptr]
//! counts number of readers left in previous buffer half
std::size_t num_readers = 0; //!< to be const after initialisation

char* const buf;

//! version for preloaded write ptr
std::size_t rl) const;

public:
//! allocating constructor
//! @param sz size of buffer being allocated
ringbuffer_t(std::size_t sz);
~ringbuffer_t();

//! size that is guaranteed to be writable one all readers
//! are up to date
std::size_t maximum_eventual_write_space() const {
return size >> 1;
}

//! returns number of bytes that can be written at least
std::size_t write_space() const;
//! writes max(cnt, write_space) of src into the buffer
//! @return number of bytes successfully written
std::size_t write(const char *src, size_t cnt);
};


And finally, the reader class. It contains a helper class read_sequence_t. If you want to read using this reader, the reader returns such a read_sequence_t object. Only one read per reader at a time is allowed (this is not checked).

class ringbuffer_reader_t : protected ringbuffer_common_t
{
const char* const buf;
ringbuffer_t* const ref;

void try_inc(std::size_t range);

class seq_base
{
const char* const buf;
std::size_t range;
protected:
public:
//! requests a read sequence of size range
buf(rb.buf),
range(range),
{
}

//! single member access
const char& operator[](std::size_t idx) {
}

std::size_t size() const { return range; }
};

class read_sequence_t : public seq_base {
public:
using seq_base::seq_base;
};

template<class Sequence>
std::size_t rs2 = rs < range ? 0 : range;
return Sequence(*this, rs2);
}

public:
//! constuctor. registers this reader at the ringbuffer
//! @note careful: this function is @a not thread-safe

}

//! returns number of bytes that can be read at least
};


Implementation

The common interface simply sets the size to some power of two. I won't show this code here.

Ctor and Dtor of the writer are simple:

ringbuffer_t::ringbuffer_t(std::size_t sz) :
ringbuffer_common_t(sz),
buf(new char[ringbuffer_common_t::size])
{
w_ptr.store(0, std::memory_order_relaxed);
}

ringbuffer_t::~ringbuffer_t()
{
delete[] buf;
}


How much can the writer write?

std::size_t ringbuffer_t::write_space_preloaded(std::size_t w,
std::size_t rl) const
{
return (((size_mask - w) & (size_mask >> 1))) // = before next half
+ ((rl == false) * (size >> 1)) // one more block?
;
}

std::size_t ringbuffer_t::write_space() const
{
}


Now, the write procedure of the writer:

std::size_t ringbuffer_t::write (const char *src, size_t cnt)
{

// size calculations
std::size_t free_cnt;
if ((free_cnt = write_space_preloaded(w, rl)) == 0) {
return 0;
}

const std::size_t to_write = cnt > free_cnt ? free_cnt : cnt;
const std::size_t cnt2 = w + to_write;

std::size_t n1, n2;
if (cnt2 > size) {
} else {
n1 = to_write;
n2 = 0;
}

if((w ^ ((w + to_write) & size_mask)) & (size >> 1)) // msb flipped
{
if(rl) throw "impossible";
}

// here starts the writing

std::copy_n(src, n1, &(buf[w]));
w = (w + n1) & size_mask;
w_ptr.store(w, std::memory_order_relaxed);

if (n2) {
std::copy_n(src + n1, n2, &(buf[w]));
w = (w + n2) & size_mask;
w_ptr.store(w, std::memory_order_relaxed);
}

}


We're almost done. Now to the reader - the Ctor is simple:

ringbuffer_reader_t::ringbuffer_reader_t(ringbuffer_t &ref) :
ringbuffer_common_t(ref.size), buf(ref.buf), ref(&ref) {
++ref.num_readers; // register at the writer
}


How much can the reader read? It is much more simple here since we know the exact position of the write pointer.

std::size_t ringbuffer_reader_t::read_space() const
{
const std::size_t

if (w > r) {
return w - r;
} else {
return (w - r + ref->size) & ref->size_mask;
}
}


Finally, after we have read something (using read_sequence_t), we need to increase the read_ptr:

void ringbuffer_reader_t::try_inc(std::size_t range)
{
// checks if highest bit flipped:
{
}
}


What works and what not

As one can see in github, I made sequential and parallel tests, all working. Valgrind saw no memory leaks. However, I have no tool to debug race conditions.

My questions are:

1. Is this code safe of any race conditions? (main question)
2. Is there anything you would improve concerning efficiency? (e.g. bit manipulation)
3. ...
• This is a red flag. I needed a lock-free ringbuffer. What you need is a ringbuffer weather you use locks or not depends. Locks may be expensive but a naively written lock-free alternative usually has throughput issues. As such when writing lock-free variants you should also write a version with locks so that you can do speed comparison tests. – Martin York Mar 22 '15 at 17:10
• @LokiAstari actually it is correct. I needed a ringbuffer, and it needed to be lock-free, because locks are usually considered non RT-safe, and RT-safety was a requirement. So that sentence should makes sense. – Johannes Mar 22 '15 at 18:42
• I am unfamiliar with the term RT-safe. – Martin York Mar 22 '15 at 19:39
• @LokiAstari What I meant was real-time-safe. I required the ringbuffers in working with audio, where the app needed to guarantee that ringbuffer functions are finished in a "short enough" time. – Johannes Mar 22 '15 at 20:05
• Sure. You don't want a short time. You basically don't want the readers to block if there is data available to read. There a couple of issues you should be aware of. 1) Writing lock-free code is harder. 2) Throughput of nieve lock-free code is a know common problem Common Pitfalls in Writing Lock-Free Algorithms. This is why I would suggest writing a locking version. Not because it will be quicker or anything. But more as a data point to compare throughput against. – Martin York Mar 22 '15 at 20:16

## Barriers missing

I know this is an old question, but I started looking at the "lock-free" tag and came across this unanswered question. I believe that your code is unsafe because it is lacking the proper memory barriers. For example, in this code snippet:

std::copy_n(src, n1, &(buf[w]));
w = (w + n1) & size_mask;

The std::memory_order_relaxed doesn't provide any protection against store reordering. So the store to w_ptr can be reordered to be before the copy to the buffer. Later on, the reader could see an updated write pointer and try to read from the buffer before the buffer contents have been stored. You should use std::memory_order_release to prevent this.
You have similar problems on the reader side, where you should use std::memory_order_acquire instead of std::memory_order_relaxed.