sdk/include/ring_buffer.hh

#pragma once
#include <limits>
#include <locks.hh>
#include <thread.h>
#include <utils.hh>

/**
 * A simple single-producer, single-consumer ring buffer that can be made
 * multi-producer and / or multi-consumer by adding locking.  The first two
 * template parameters are the type of the message and the size of the buffer
 * (in messages).  The next template arguments are the locks to use for
 * protecting the producer and consumer ends.
 *
 * In the uncontended (or single-producer, single-consumer) case where the
 * producer and consumer are balanced and the ring is neither full nor empty,
 * this will not use any cross-compartment calls.  When the queue is full,
 * producers will block on a futex.  When the queue is empty, consumers will
 * block on a futex.  On transitions to non-empty and non-full, the queue will
 * issue a `futex_wake` cross-compartment call. If locks are used, they may
 * introduce additional cross-compartment calls.
 *
 * Note: The size must be a power of two.
 */
template<typename Message,
         size_t BufferSize,
         typename PushLock = NoLock,
         typename PopLock  = NoLock>
class RingBuffer
{
	/// The ring buffer.
	std::array<Message, BufferSize> ring;
	/// The lock protecting the push end, if one exists.
	[[no_unique_address]] PushLock pushLock;
	/// The lock protecting the pop end, if one exists.
	[[no_unique_address]] PopLock popLock;
	/// Free-running producer counter.
	cheriot::atomic<uint32_t> producer;
	/// Free-running consumer counter.
	cheriot::atomic<uint32_t> consumer;

	static_assert(BufferSize < std::numeric_limits<uint32_t>::max() / 2,
	              "The buffer size cannot be more than half the range of the "
	              "counter types or overflow will give incorrect values");
	static_assert((1 << utils::log2<BufferSize>()) == BufferSize,
	              "Buffer size must be a power of two");

	/**
	 * Helper to check if the ring is full.  The counters are free running and
	 * so the displacement between them (wrapping for overflow) will be the
	 * number of elements in the buffer if the ring is full.
	 */
	bool is_full()
	{
		return producer - consumer == BufferSize;
	}

	/**
	 * Check if the queue is empty.  If the consumer counter has caught up with
	 * the producer then this ring is empty.
	 */
	bool is_empty()
	{
		return producer == consumer;
	}

	/**
	 * Helper to convert the counter value to an array index.  The counters are
	 * never reset and so this is simple modulo arithmetic.  Power of two sizes
	 * are more efficient because they become bit masks.
	 */
	size_t counter_to_index(size_t counter)
	{
		return counter % BufferSize;
	}

	public:
	/**
	 * Push an element into the ring.  The caller is responsible for ensuring
	 * that, if this contains pointers, they have the global permission and so
	 * the move operation will not fault.
	 */
	void push(Message &&message)
	{
		bool wasEmpty;
		{
			LockGuard g{pushLock};
			// Wait for the queue to not be full
			while (is_full())
			{
				consumer.wait(producer - BufferSize);
			}
			auto i   = counter_to_index(producer);
			ring[i]  = std::move(message);
			wasEmpty = is_empty();
			producer++;
		}
		if (wasEmpty)
		{
			producer.notify_all();
		}
	}

	/**
	 * Pop the next message from the queue.
	 */
	Message pop()
	{
		Message result;
		bool    wasFull;
		{
			LockGuard g{popLock};
			// Wait for the queue to not be empty
			while (is_empty())
			{
				producer.wait(consumer);
			}
			auto i  = counter_to_index(consumer);
			result  = std::move(ring[i]);
			wasFull = is_full();
			consumer++;
		}
		if (wasFull)
		{
			consumer.notify_all();
		}
		return result;
	}
};

// Make sure that locks consume no space if not used.
static_assert(sizeof(RingBuffer<uint32_t, 2>) == 16);