dc-terrain / libdc / include / dc_spsc_queue.h

#ifndef DC_SPSC_QUEUE_H
#define DC_SPSC_QUEUE_H

#include <atomic>
#include <cstddef>
#include <cstdint>
#include <new>
#include <type_traits>
#include <utility>

// Lock-free single-producer single-consumer ring buffer.
// N must be a power of 2. Usable capacity is N-1.
// Producer calls try_push(), consumer calls try_pop().
// Thread-safe without mutexes via acquire/release memory ordering.

template <typename T, size_t N>
class SPSCQueue {
	static_assert(N >= 2, "Capacity must be at least 2");
	static_assert((N & (N - 1)) == 0, "Capacity must be a power of 2");

public:
	SPSCQueue() : head_(0), tail_(0) {}

	~SPSCQueue() {
		// Destroy any remaining elements
		size_t h = head_.load(std::memory_order_relaxed);
		size_t t = tail_.load(std::memory_order_relaxed);
		while (h != t) {
			reinterpret_cast<T *>(&storage_[h])->~T();
			h = (h + 1) & mask_;
		}
	}

	SPSCQueue(const SPSCQueue &) = delete;
	SPSCQueue &operator=(const SPSCQueue &) = delete;

	// Producer: push by const ref. Returns false if full.
	bool try_push(const T &item) {
		size_t t = tail_.load(std::memory_order_relaxed);
		size_t next = (t + 1) & mask_;
		if (next == head_.load(std::memory_order_acquire)) {
			return false; // full
		}
		new (&storage_[t]) T(item);
		tail_.store(next, std::memory_order_release);
		return true;
	}

	// Producer: push by rvalue ref. Returns false if full.
	bool try_push(T &&item) {
		size_t t = tail_.load(std::memory_order_relaxed);
		size_t next = (t + 1) & mask_;
		if (next == head_.load(std::memory_order_acquire)) {
			return false; // full
		}
		new (&storage_[t]) T(std::move(item));
		tail_.store(next, std::memory_order_release);
		return true;
	}

	// Consumer: pop into out. Returns false if empty.
	bool try_pop(T &out) {
		size_t h = head_.load(std::memory_order_relaxed);
		if (h == tail_.load(std::memory_order_acquire)) {
			return false; // empty
		}
		T *elem = reinterpret_cast<T *>(&storage_[h]);
		out = std::move(*elem);
		elem->~T();
		head_.store((h + 1) & mask_, std::memory_order_release);
		return true;
	}

	// Approximate emptiness check (safe from either thread, but may be stale).
	bool empty() const {
		return head_.load(std::memory_order_acquire) ==
				tail_.load(std::memory_order_acquire);
	}

	// Approximate count (safe from either thread, may be stale). 0..N-1.
	size_t approx_size() const {
		size_t t = tail_.load(std::memory_order_acquire);
		size_t h = head_.load(std::memory_order_acquire);
		return (t - h) & mask_;
	}

private:
	static constexpr size_t mask_ = N - 1;

	// Cache-line aligned to prevent false sharing between producer and consumer.
	alignas(64) std::atomic<size_t> head_;
	alignas(64) std::atomic<size_t> tail_;

	// Uninitialized storage for T elements.
	alignas(alignof(T)) typename std::aligned_storage<sizeof(T), alignof(T)>::type storage_[N];
};

#endif // DC_SPSC_QUEUE_H