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

Question

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:

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. ...

Answer 1

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;
// update so readers are already informed:
w_ptr.store(w, std::memory_order_relaxed);

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.

Why do my tests work?

Most likely, you are testing this on an x86 based system. The x86 architecture has certain memory ordering guarantees that make most memory barriers unnecessary. So your code will in fact operate correctly when run on an x86 architecture.