Lock-free triple buffer

Question

I have a single producer and single consumer, and the producer never stops, but the consumer might not keep up. There's no need to consume every item, as long as we always access the most-recently produced item, and never process an item twice. In other words, we should drop items if the consumer isn't keeping up, but wait for the producer if it is.

#ifndef TRIPLE_BUFFER_HPP
#define TRIPLE_BUFFER_HPP

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <mutex>

template<typename T>
class triple_buffer
{
    // the actual buffer
    T buffer[3] = {};

    // the roles the buffers currently have
    // read and write buffers are private to each side
    // the available buffer is passed between them
    T* readbuffer = &buffer[0];
    T* writebuffer = &buffer[1];
    std::atomic<T*> available = &buffer[2];

    // the last fully-written buffer waiting ready for reader
    std::atomic<T*> next_read_buf = nullptr;

    // When the reader catches up, it needs to wait for writer (slow path only)
    std::mutex read_queue_mutex = {};
    std::condition_variable read_queue = {};

public:

    // Writer interface

    // Writer has ownership of this buffer (this function never blocks).
    T *get_write_buffer()
    {
        return writebuffer;
    }

    // Writer releases ownership of its buffer.
    void set_write_complete()
    {
        // give back the write buffer
        auto *written = writebuffer;
        writebuffer = available.exchange(writebuffer);
        // mark it as written
        next_read_buf.store(written, std::memory_order_release);
        // notify any waiting reader
        read_queue.notify_one();
    }

    // Reader interface

    // Reader gets ownership of the buffer, until the next call of
    // get_read_buffer().
    T *get_read_buffer(std::chrono::milliseconds timeout = std::chrono::milliseconds::max())
    {
        auto const timeout_time = std::chrono::steady_clock::now() + timeout;

        // get the written buffer, waiting if necessary
        auto *b = next_read_buf.exchange(nullptr);
        while (b != readbuffer) {
            // it could be the available buffer
            readbuffer = available.exchange(readbuffer);
            if (b == readbuffer) {
                // yes, that's it
                return readbuffer;
            }
            // else we need to wait for writer
            b = nullptr;
            std::unique_lock lock{read_queue_mutex};
            auto test = [this,&b]{ b = next_read_buf.exchange(nullptr); return b; };
            if (!read_queue.wait_until(lock, timeout_time, test)) {
                return nullptr;
            }
        }

        return readbuffer;
    }


    // The unit test helper is enabled only if <gtest.h> is included before this header.
    // It's not available (or necessary) in production code.
#ifdef TEST
    // N.B. not thread-safe - only call this when reader and writer are idle
    void test_invariant(const char *file, int line) const
    {
        const std::set<const T*> buffers{&buffer[0], &buffer[1], &buffer[2]};
        const std::set<const T*> roles{readbuffer, available, writebuffer};
        auto const fail = buffers != roles
            || next_read_buf && !buffers.count(next_read_buf)
            || next_read_buf == writebuffer;
        if (fail) {
            auto name = [this](const T *slot){
                if (slot == &buffer[0]) { return "buffer[0]"; }
                if (slot == &buffer[1]) { return "buffer[1]"; }
                if (slot == &buffer[2]) { return "buffer[2]"; }
                if (slot == nullptr) { return "(null)"; }
                return "(invalid)";
            };
            ADD_FAILURE_AT(file, line) <<
                "Buffer/role mismatch:\n"
                "Buffers = " << &buffer[0] << ", " << &buffer[1] << ", " << &buffer[2] << "\n"
                "Read = " << readbuffer << " = " << name(readbuffer) << "\n"
                "Available = " << available  << " = "<< name(available) << "\n"
                "Write = " << writebuffer << " = " << name(writebuffer) << "\n"
                "Next Read = " << next_read_buf << " = " << name(next_read_buf) << "\n";
        }
    }
#endif
};

#endif // TRIPLE_BUFFER_HPP

I created this with help from a set of unit tests:

#include <gtest/gtest.h>

#include <triple-buffer.hpp>

#include <set>


#define EXPECT_INVARIANT(obj) (obj.test_invariant(__FILE__, __LINE__))


TEST(triple_buffer, no_write_read_returns_null)
{
    triple_buffer<int> buffer;
    EXPECT_INVARIANT(buffer);
    EXPECT_EQ(buffer.get_read_buffer({}), nullptr);
    EXPECT_INVARIANT(buffer);
}

TEST(triple_buffer, write_once_read_once)
{
    triple_buffer<int> buffer;
    auto *write0 = buffer.get_write_buffer();
    EXPECT_INVARIANT(buffer);
    EXPECT_NE(write0, nullptr);
    EXPECT_EQ(buffer.get_read_buffer({}), nullptr);
    EXPECT_INVARIANT(buffer);

    buffer.set_write_complete();
    EXPECT_INVARIANT(buffer);
    // having written, we can read
    EXPECT_EQ(buffer.get_read_buffer({}), write0);
    EXPECT_INVARIANT(buffer);

    // another read should block/fail
    EXPECT_EQ(buffer.get_read_buffer({}), nullptr);
    EXPECT_INVARIANT(buffer);
}

TEST(triple_buffer, write_twice_read)
{
    triple_buffer<int> buffer;
    auto *write0 = buffer.get_write_buffer();
    EXPECT_NE(write0, nullptr);
    buffer.set_write_complete();

    auto *write1 = buffer.get_write_buffer();
    EXPECT_NE(write1, nullptr);
    EXPECT_NE(write1, write0);
    buffer.set_write_complete();

    // read should get newest
    EXPECT_EQ(buffer.get_read_buffer({}), write1);
    // another read should block/fail
    EXPECT_EQ(buffer.get_read_buffer({}), nullptr);
}

TEST(triple_buffer, write_read_write2)
{
    triple_buffer<int> buffer;
    auto *write0 = buffer.get_write_buffer();
    EXPECT_NE(write0, nullptr);
    buffer.set_write_complete();

    // read should get newest
    auto *read0 = buffer.get_read_buffer({});
    EXPECT_EQ(read0, write0);

    auto *write1 = buffer.get_write_buffer();
    EXPECT_NE(write1, nullptr);
    EXPECT_NE(write1, write0);
    buffer.set_write_complete();

    auto *write2 = buffer.get_write_buffer();
    EXPECT_NE(write2, nullptr);
    EXPECT_NE(write2, write1);
    EXPECT_NE(write2, read0);   // don't touch reader's buffer
    buffer.set_write_complete();

    // read should get newest
    auto *read1 = buffer.get_read_buffer({});
    EXPECT_EQ(read1, write2);
}

TEST(triple_buffer, write_read_write3)
{
    triple_buffer<int> buffer;
    auto *write0 = buffer.get_write_buffer();
    EXPECT_NE(write0, nullptr);
    buffer.set_write_complete();

    // read should get newest
    auto *read0 = buffer.get_read_buffer({});
    EXPECT_EQ(read0, write0);

    auto *write1 = buffer.get_write_buffer();
    EXPECT_NE(write1, nullptr);
    EXPECT_NE(write1, write0);
    buffer.set_write_complete();

    auto *write2 = buffer.get_write_buffer();
    EXPECT_NE(write2, nullptr);
    EXPECT_NE(write2, write1);
    EXPECT_NE(write2, read0);   // don't touch reader's buffer
    buffer.set_write_complete();

    auto *write3 = buffer.get_write_buffer();
    EXPECT_EQ(write3, write1);  // we should be overwriting the old unread value
    EXPECT_NE(write3, read0);   // still don't touch reader's buffer
    buffer.set_write_complete();

    // read should get newest
    auto *read1 = buffer.get_read_buffer({});
    EXPECT_EQ(read1, write3);
}


TEST(triple_buffer, read_during_write)
{
    triple_buffer<int> buffer;
    auto *write0 = buffer.get_write_buffer();
    EXPECT_NE(write0, nullptr);
    buffer.set_write_complete();

    auto *write1 = buffer.get_write_buffer();

    // read should get complete buffer
    auto *read0 = buffer.get_read_buffer({});
    EXPECT_EQ(read0, write0);

    buffer.set_write_complete();
    auto *write2 = buffer.get_write_buffer();
    EXPECT_NE(write2, nullptr);
    EXPECT_NE(write2, write1);
    EXPECT_NE(write2, read0);   // don't touch reader's buffer

    auto *read1 = buffer.get_read_buffer({});
    EXPECT_EQ(read1, write1);

    buffer.set_write_complete();
}

Obviously, the unit tests are limited, being single threaded. In particular, they don't cover the wait_until() call or provoke any races. I feel I should probably create some module tests that can be slower than unit tests (I consider 1ms runtime the absolute upper limit for a unit-test), but don't have any experience using slower tests like that.

Anyway, my main concerns with the code (both the class and its tests) are:

1. Correctness. I believe it's correct in all cases, but concurrent code is notoriously hard to fully reason about, and I'd appreciate any insight into what I've missed.
2. Readability. Can this be understood and modified by someone else (future-me, in particular)?
3. Given I know there's only one reader, do I really need read_queue_mutex to be a member, or can I get away with instantiating a new one in each get_read_buffer() call? If so, should I?

Answer 1

Answers to your questions

Correctness. I believe it's correct in all cases, but concurrent code is notoriously hard to fully reason about, and I'd appreciate any insight into what I've missed.

It don't see any correctness issues.

Readbility. Can this be understood and modified by someone else (future-me, in particular)?

Lock-free code is always hard to follow. I think the get_write_buffer() and set_write_complete() functions are relatively easy to understand. The hardest to follow is get_read_buffer(). I would explain the invariants in the code:

• readbuffer is always owned by the consumer
• writebuffer is always owned by the producer
• available is never owned by either
• you can atomically exchange read/writebuffer with available without breaking the above invariants

And then also explain the trick: next_read_buf is set by the producer after it is finished with the buffer, and it also swaps writebuffer and available, after which next_read_buf == available. The consumer can atomically take over available in this case.

The wait is necessary in two cases: next_read_buf is nullptr (producer didn't produce anything yet since last consume), or if it just exchanged available and writebuffer, but had not written to next_read_buf yet.

Given I know there's only one reader, do I really need read_queue_mutex to be a member, or can I get away with instantiating a new one in each get_read_buffer() call? If so, should I?

You can instantiate a new one in each call. Either you pay the small price of instantiating a new std::mutex every call, or you pay the price of keeping a std::mutex around all the time. I don't think it matters much for performance. I would keep it like it is, so that the mutex and condition variable are kept close together in the source code.

This problem is solved in C++20 with std::atomic_wait().

Consider using RAII to convey buffer ownership

A producer has to do two steps: get a pointer to a buffer to write to, and then release that buffer when it is done. That sounds like a job for RAII. Similar to std::lock_guard, consider adding a way to get a handle for the write buffer that will automatically do set_write_complete() when its lifetime ends. You could even do the same for the read buffer for symmetry's sake.