Skip to content

refactor(util): replace lock-free SPSCQueue with mutex-guarded ring buffer #145

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
K1ngst0m merged 1 commit into from
Mar 25, 2026
Merged

refactor(util): replace lock-free SPSCQueue with mutex-guarded ring buffer #145

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 33 additions & 62 deletions src/util/queues.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,115 +1,86 @@
#pragma once

#include <algorithm>
#include <atomic>
#include <cstddef>
#include <cstdlib>
#include <memory>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <type_traits>
#include <vector>
Comment on lines 3 to +7
Copy link
Copy Markdown

@qodo-code-review qodo-code-review bot Mar 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Action required

1. Tests missing required includes 🐞 Bug ✓ Correctness

tests/util/test_queues.cpp uses std::unique_ptr/std::make_unique and std::string but does not
include <memory> or <string>; this likely compiled only because queues.hpp previously included
<memory>. After this PR removes those transitive includes, the test can fail to compile depending on
the standard library/Catch2 headers.
Agent Prompt 
### Issue description
`tests/util/test_queues.cpp` uses `std::unique_ptr`, `std::make_unique`, and `std::string` but does not include `<memory>` / `<string>`. This likely became visible after `queues.hpp` stopped including `<memory>`.

### Issue Context
This is a compile portability issue caused by relying on transitive includes.

### Fix Focus Areas
- tests/util/test_queues.cpp[1-10]
  - Add `#include <memory>` and `#include <string>` (and any other directly-used standard headers not already included).

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools


namespace goggles::util {

/// @brief Single-producer, single-consumer lock-free ring buffer.
/// @brief Single-producer, single-consumer queue with fixed capacity.
///
/// `capacity` must be a power of two. Construction throws on invalid capacity or allocation
/// failure.
/// Capacity must be > 0. Construction throws `std::invalid_argument` on zero capacity.
template <typename T>
class SPSCQueue {
public:
explicit SPSCQueue(size_t capacity)
: m_capacity(capacity), m_buffer_size(capacity * 2), m_capacity_mask(m_buffer_size - 1),
m_buffer(nullptr) {
// Power-of-2 required for efficient modulo via bitwise AND
if (capacity == 0 || (capacity & (capacity - 1)) != 0) {
throw std::invalid_argument("SPSCQueue capacity must be power of 2");
: m_capacity(capacity), m_buffer(capacity), m_head(0), m_tail(0), m_size(0) {
if (capacity == 0) {
throw std::invalid_argument("SPSCQueue capacity must be > 0");
}

m_buffer = static_cast<T*>(std::aligned_alloc(alignof(T), sizeof(T) * m_buffer_size));
if (m_buffer == nullptr) {
throw std::bad_alloc();
}

m_head.store(0, std::memory_order_relaxed);
m_tail.store(0, std::memory_order_relaxed);
}

~SPSCQueue() {
while (try_pop()) {
}
if (m_buffer) {
std::free(m_buffer);
}
}
~SPSCQueue() = default;

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

auto try_push(const T& item) -> bool {
const size_t current_head = m_head.load(std::memory_order_relaxed);
const size_t current_tail = m_tail.load(std::memory_order_acquire);
const size_t current_size = (current_head - current_tail) & m_capacity_mask;
if (current_size >= m_capacity) {
std::lock_guard lock(m_mutex);
if (m_size >= m_capacity) {
return false;
}

new (&m_buffer[current_head]) T(item);
const size_t next_head = (current_head + 1) & m_capacity_mask;
m_head.store(next_head, std::memory_order_release);
m_buffer[m_head].emplace(item);
m_head = (m_head + 1) % m_capacity;
++m_size;
return true;
}

auto try_push(T&& item) -> bool {
const size_t current_head = m_head.load(std::memory_order_relaxed);
const size_t current_tail = m_tail.load(std::memory_order_acquire);
const size_t current_size = (current_head - current_tail) & m_capacity_mask;
if (current_size >= m_capacity) {
std::lock_guard lock(m_mutex);
if (m_size >= m_capacity) {
return false;
}

new (&m_buffer[current_head]) T(std::move(item));
const size_t next_head = (current_head + 1) & m_capacity_mask;
m_head.store(next_head, std::memory_order_release);
m_buffer[m_head].emplace(std::move(item));
m_head = (m_head + 1) % m_capacity;
++m_size;
Comment on lines 42 to +49
Copy link
Copy Markdown

@qodo-code-review qodo-code-review bot Mar 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Action required

2. Missing include 🐞 Bug ✓ Correctness

src/util/queues.hpp uses std::move but does not include <utility>, making the header
non-self-contained and potentially failing to compile on some toolchains/stdlib implementations.
Agent Prompt 
### Issue description
`src/util/queues.hpp` uses `std::move` but does not include `<utility>`, so the header is not self-contained.

### Issue Context
Relying on indirect includes (e.g., through `<optional>` or `<vector>`) is not guaranteed by the standard and can break builds across platforms/stdlibs.

### Fix Focus Areas
- src/util/queues.hpp[1-10]
  - Add `#include <utility>`.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

return true;
}

auto try_pop() -> std::optional<T> {
const size_t current_tail = m_tail.load(std::memory_order_relaxed);
if (current_tail == m_head.load(std::memory_order_acquire)) {
std::lock_guard lock(m_mutex);
if (m_size == 0) {
return std::nullopt;
}

T item = std::move(m_buffer[current_tail]);
m_buffer[current_tail].~T();
const size_t next_tail = (current_tail + 1) & m_capacity_mask;
m_tail.store(next_tail, std::memory_order_release);
std::optional<T> item = std::move(m_buffer[m_tail]);
m_buffer[m_tail].reset();
m_tail = (m_tail + 1) % m_capacity;
--m_size;
return item;
}

[[nodiscard]] auto size() const -> size_t {
const size_t current_head = m_head.load(std::memory_order_acquire);
const size_t current_tail = m_tail.load(std::memory_order_acquire);
return (current_head - current_tail) & m_capacity_mask;
std::lock_guard lock(m_mutex);
return m_size;
}

[[nodiscard]] auto empty() const -> bool {
const size_t current_tail = m_tail.load(std::memory_order_relaxed);
return current_tail == m_head.load(std::memory_order_acquire);
std::lock_guard lock(m_mutex);
return m_size == 0;
}

[[nodiscard]] auto capacity() const -> size_t { return m_capacity; }

private:
alignas(64) std::atomic<size_t> m_head;
alignas(64) std::atomic<size_t> m_tail;

const size_t m_capacity;
const size_t m_buffer_size;
const size_t m_capacity_mask;
T* m_buffer;
std::vector<std::optional<T>> m_buffer;
size_t m_head;
size_t m_tail;
size_t m_size;
mutable std::mutex m_mutex;
};

} // namespace goggles::util
19 changes: 11 additions & 8 deletions tests/util/test_queues.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,21 +7,28 @@
using namespace goggles::util;

TEST_CASE("SPSCQueue construction and basic properties", "[queues]") {
SECTION("Construct with power-of-2 capacity") {
SECTION("Construct with valid capacity") {
SPSCQueue<int> queue(8);
REQUIRE(queue.capacity() == 8);
REQUIRE(queue.size() == 0);
}

SECTION("Construct with non-power-of-2 capacity throws") {
REQUIRE_THROWS_AS(SPSCQueue<int>(7), std::invalid_argument);
REQUIRE_THROWS_AS(SPSCQueue<int>(10), std::invalid_argument);
SECTION("Non-power-of-2 capacity is accepted") {
SPSCQueue<int> queue(7);
REQUIRE(queue.capacity() == 7);

SPSCQueue<int> queue2(10);
REQUIRE(queue2.capacity() == 10);
}

SECTION("Minimum capacity of 1 works") {
SPSCQueue<int> queue(1);
REQUIRE(queue.capacity() == 1);
}

SECTION("Zero capacity is rejected") {
REQUIRE_THROWS_AS(SPSCQueue<int>(0), std::invalid_argument);
}
}

TEST_CASE("SPSCQueue basic operations", "[queues]") {
Expand Down Expand Up @@ -158,10 +165,6 @@ TEST_CASE("SPSCQueue FIFO ordering", "[queues]") {
}

TEST_CASE("SPSCQueue edge cases and boundary conditions", "[queues]") {
SECTION("Zero capacity is rejected") {
REQUIRE_THROWS_AS(SPSCQueue<int>(0), std::invalid_argument);
}

SECTION("Capacity 1 handles full/empty correctly") {
SPSCQueue<int> queue(1);

Expand Down
Loading