File size: 3,664 Bytes
f86fb3b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | #pragma once
#include <atomic>
#include <cstddef>
#include <optional>
#include <utility>
#include <memory>
#include <thread>
#include "cache_line.hpp"
namespace quant::core {
/**
* Lock-Free Multi-Producer Single-Consumer (MPSC) Bounded Ring Buffer Queue.
* Based on Dmitry Vyukov's bounded queue algorithm.
*
* Allows multiple concurrent network / gateway client threads to push orders
* into a single matching engine thread without mutexes.
*/
template <typename T, size_t Capacity = 65536>
class alignas(CACHELINE_SIZE) LockFreeMPSCQueue {
static_assert(Capacity >= 2, "Capacity must be at least 2");
static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be a power of two");
struct Cell {
std::atomic<size_t> sequence;
T data;
};
public:
LockFreeMPSCQueue() : tail_(0), head_(0) {
buffer_ = std::make_unique<Cell[]>(Capacity);
for (size_t i = 0; i < Capacity; ++i) {
buffer_[i].sequence.store(i, std::memory_order_relaxed);
}
}
~LockFreeMPSCQueue() = default;
LockFreeMPSCQueue(const LockFreeMPSCQueue&) = delete;
LockFreeMPSCQueue& operator=(const LockFreeMPSCQueue&) = delete;
/**
* Push an item into the queue. Safe for multiple concurrent producers.
*/
template <typename... Args>
bool emplace(Args&&... args) {
Cell* cell;
size_t pos = tail_.load(std::memory_order_relaxed);
for (;;) {
cell = &buffer_[pos & MASK];
size_t seq = cell->sequence.load(std::memory_order_acquire);
intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
if (diff == 0) {
if (tail_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
break;
}
} else if (diff < 0) {
return false; // Queue is full
} else {
pos = tail_.load(std::memory_order_relaxed);
}
}
cell->data = T(std::forward<Args>(args)...);
cell->sequence.store(pos + 1, std::memory_order_release);
return true;
}
bool push(const T& item) { return emplace(item); }
bool push(T&& item) { return emplace(std::move(item)); }
/**
* Pop an item from the queue. Strictly single-consumer.
*/
bool pop(T& out_item) {
Cell* cell = &buffer_[head_ & MASK];
size_t seq = cell->sequence.load(std::memory_order_acquire);
intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(head_ + 1);
if (diff == 0) {
out_item = std::move(cell->data);
cell->sequence.store(head_ + Capacity, std::memory_order_release);
++head_;
return true;
}
return false; // Queue is empty
}
std::optional<T> pop() {
T item;
if (pop(item)) return item;
return std::nullopt;
}
[[nodiscard]] size_t size() const noexcept {
size_t t = tail_.load(std::memory_order_relaxed);
size_t h = head_;
return (t >= h) ? (t - h) : 0;
}
[[nodiscard]] bool empty() const noexcept {
return size() == 0;
}
[[nodiscard]] constexpr size_t capacity() const noexcept {
return Capacity;
}
private:
static constexpr size_t MASK = Capacity - 1;
// Tail: shared and modified by multiple producers via CAS
alignas(CACHELINE_SIZE) std::atomic<size_t> tail_;
// Head: accessed and modified strictly by the single consumer
alignas(CACHELINE_SIZE) size_t head_;
std::unique_ptr<Cell[]> buffer_;
};
} // namespace quant::core
|