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:
- 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.
- Readability. Can this be understood and modified by someone else (future-me, in particular)?
- 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 eachget_read_buffer()
call? If so, should I?