55 lines
2.7 KiB
C++
55 lines
2.7 KiB
C++
#pragma once
|
|
#include <array>
|
|
#include <atomic>
|
|
#include <cstddef>
|
|
#include <cstdint>
|
|
|
|
/* Ограниченная очередь с несколькими producers. Producer не меняет позицию consumer.
|
|
* Резервирование и публикация разделены sequence каждого слота; callback не ждёт занятый слот.
|
|
* reset допустим только до запуска всех участников. Последние 8 слотов зарезервированы для команд. */
|
|
template<class T, size_t Capacity> class AudioEventQueue {
|
|
static_assert(Capacity > 8);
|
|
struct Slot { std::atomic<uint64_t> sequence{0}; T value{}; };
|
|
std::array<Slot, Capacity> m_slots;
|
|
alignas(64) std::atomic<uint64_t> m_write{0};
|
|
alignas(64) std::atomic<uint64_t> m_read{0};
|
|
public:
|
|
AudioEventQueue() { reset(); }
|
|
void reset() {
|
|
m_write = 0; m_read = 0;
|
|
for (size_t i = 0; i < Capacity; ++i) m_slots[i].sequence = i;
|
|
}
|
|
bool push(const T& value, bool control = false) {
|
|
uint64_t position = m_write.load(std::memory_order_relaxed);
|
|
for (int attempt = 0; attempt < 8; ++attempt) {
|
|
uint64_t read = m_read.load(std::memory_order_acquire);
|
|
if (read > position) { position = m_write.load(std::memory_order_relaxed); continue; }
|
|
if (position - read >= Capacity - (control ? 0 : 8)) return false;
|
|
auto& slot = m_slots[position % Capacity];
|
|
int64_t difference = (int64_t)(slot.sequence.load(std::memory_order_acquire) - position);
|
|
if (difference < 0) return false;
|
|
if (difference == 0 && m_write.compare_exchange_weak(position, position + 1, std::memory_order_relaxed)) {
|
|
slot.value = value;
|
|
slot.sequence.store(position + 1, std::memory_order_release);
|
|
return true;
|
|
}
|
|
position = m_write.load(std::memory_order_relaxed);
|
|
}
|
|
return false;
|
|
}
|
|
bool pop(T& value) {
|
|
uint64_t position = m_read.load(std::memory_order_relaxed);
|
|
auto& slot = m_slots[position % Capacity];
|
|
if (slot.sequence.load(std::memory_order_acquire) != position + 1) return false;
|
|
value = slot.value;
|
|
slot.sequence.store(position + Capacity, std::memory_order_release);
|
|
m_read.store(position + 1, std::memory_order_release);
|
|
return true;
|
|
}
|
|
size_t size() const {
|
|
uint64_t read = m_read.load(std::memory_order_acquire);
|
|
uint64_t count = m_write.load(std::memory_order_acquire) - read;
|
|
return (size_t)(count > Capacity ? Capacity : count); // Приблизительный снимок, включая ещё не опубликованные слоты.
|
|
}
|
|
};
|