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::atomic<std::size_t> readers_left;
std::size_t num_readers = 0; //!< to be const after initialisation
char* const buf;
friend class ringbuffer_reader_t;
//! version for preloaded write ptr
std::size_t write_space_preloaded(std::size_t w,
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;
std::size_t read_ptr = 0; //!< reader at buf[read_ptr]
//! increases the @a read_ptr after reading from the buffer
void try_inc(std::size_t range);
class seq_base
{
const char* const buf;
std::size_t range;
protected:
ringbuffer_reader_t* reader_ref;
public:
//! requests a read sequence of size range
seq_base(ringbuffer_reader_t& rb, std::size_t range) :
buf(rb.buf),
range(range),
reader_ref(&rb)
{
}
//! single member access
const char& operator[](std::size_t idx) {
return *(buf + ((reader_ref->read_ptr + idx) &
reader_ref->size_mask));
}
std::size_t size() const { return range; }
};
class read_sequence_t : public seq_base {
public:
using seq_base::seq_base;
//! increases the read_ptr after reading
~read_sequence_t() { reader_ref->try_inc(size()); }
};
template<class Sequence>
Sequence _read(std::size_t range) {
std::size_t rs = read_space();
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
ringbuffer_reader_t(ringbuffer_t& ref);
read_sequence_t read(std::size_t range) {
return _read<read_sequence_t>(range);
}
//! returns number of bytes that can be read at least
std::size_t read_space() const;
};
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);
readers_left.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
{
return write_space_preloaded(w_ptr.load(std::memory_order_relaxed),
readers_left.load(std::memory_order_relaxed));
}
Now, the write procedure of the writer:
std::size_t ringbuffer_t::write (const char *src, size_t cnt)
{
std::size_t w = w_ptr.load(std::memory_order_relaxed);
std::size_t rl = readers_left.load(std::memory_order_relaxed);
// 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) {
n1 = size - w_ptr.load(std::memory_order_relaxed);
n2 = cnt2 & size_mask;
} else {
n1 = to_write;
n2 = 0;
}
// reset reader_left
if((w ^ ((w + to_write) & size_mask)) & (size >> 1)) // msb flipped
{
if(rl) throw "impossible";
readers_left.store(num_readers, std::memory_order_relaxed);
}
// here starts the writing
std::copy_n(src, n1, &(buf[w]));
w = (w + n1) & size_mask;
// update so readers are already informed:
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);
}
return to_write;
}
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
w = ref->w_ptr.load(std::memory_order_relaxed),
r = read_ptr;
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)
{
const std::size_t old_read_ptr = read_ptr;
read_ptr = (read_ptr + range) & size_mask;
// checks if highest bit flipped:
if((read_ptr ^ old_read_ptr) & (size >> 1))
{
--ref->readers_left;
}
}
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:
- Is this code safe of any race conditions? (main question)
- Is there anything you would improve concerning efficiency? (e.g. bit manipulation)
- ...
I needed a lock-free ringbuffer
. What you need is aringbuffer
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. \$\endgroup\$