SpscFixedQueue.hpp

// SPDX-FileCopyrightText: 2026 Vincent Leroy
// SPDX-License-Identifier: MIT
//
// This file is part of dfx.
//
// Licensed under the MIT License. See the LICENSE file in the project root
// for full license information.
 
#pragma once
 
// C++ includes
#include <array>
#include <atomic>
#include <new>
#include <optional>
 
// Project includes
#include "../CompilerSupport.hpp"
#include "../CopyMoveControl.hpp"
 
namespace dfx::Utils
{
template<typename T, std::size_t Capacity>
class SpscFixedQueue
{
    static_assert(Capacity > 1);
 
public:
    // We add 1 to the capacity of the internal container because of the way we handle
    // the detection of isEmpty / isFull. isEmpty is readItr == writeItr and isFull
    // is readItr == (writeItr + 1) so if we didn't add this extra capacity
    // isFull would return true at Capacity - 1 which would be unexpected
    // from the user point of view
    using container_type  = std::array<std::byte, (Capacity + 1) * sizeof(T)>;
    using value_type      = T;
    using size_type       = typename container_type::size_type;
    using reference       = value_type &;
    using const_reference = value_type const &;
 
public:
    SpscFixedQueue();
    ~SpscFixedQueue();
 
    DFX_DISABLE_COPY_AND_MOVE(SpscFixedQueue);
 
public:
    [[nodiscard]] bool empty() const noexcept;

    [[nodiscard]] bool full() const noexcept;

    [[nodiscard]] size_type size() const noexcept;

    [[nodiscard]] constexpr size_type capacity() const noexcept { return Capacity; }
 
public:
    template<typename U>
    requires std::is_same_v<std::decay_t<U>, T>
    bool tryPush(U && value);

    std::optional<T> tryPop();
 
private:
    alignas(T) container_type _queue;
 
private:
    using atomic_itr = std::atomic<typename container_type::iterator>;
 
#ifdef __cpp_lib_hardware_interference_size
    DFX_DIAGNOSTIC_PUSH
    // We ignore this warning because this queue is not part of any public API
    DFX_DIAGNOSTIC_IGNORE("-Winterference-size")
    static constexpr auto alignSize = std::hardware_destructive_interference_size;
    DFX_DIAGNOSTIC_POP
#else
    static constexpr auto alignSize = 64;
#endif
 
    struct alignas(alignSize) ReadSide
    {
        atomic_itr itr;
    } _read;
 
    struct alignas(alignSize) WriteSide
    {
        atomic_itr itr;
    } _write;
};
 
template<typename T, std::size_t Capacity>
inline SpscFixedQueue<T, Capacity>::SpscFixedQueue()
    : _read  { std::begin(_queue) }
    , _write { std::begin(_queue) }
{
}
 
template<typename T, std::size_t Capacity>
inline SpscFixedQueue<T, Capacity>::~SpscFixedQueue()
{
    if constexpr (!std::is_trivially_destructible_v<T>)
    {
        // We are in the dtor: no need for heavy synchronization mechanism
        for (auto rItr = _read.itr.load(std::memory_order_relaxed); rItr != _write.itr.load(std::memory_order_relaxed);)
        {
            reinterpret_cast<T *>(rItr)->~T(); // Placement new == explicit call to dtor
 
            rItr += sizeof(T);
            if (rItr == std::end(_queue))
                rItr = std::begin(_queue);
        }
    }
}
 
template<typename T, std::size_t Capacity>
inline bool SpscFixedQueue<T, Capacity>::empty() const noexcept
{
    return size() == 0;
}
 
template<typename T, std::size_t Capacity>
inline bool SpscFixedQueue<T, Capacity>::full() const noexcept
{
    return size() == capacity();
}
 
template<typename T, std::size_t Capacity>
inline SpscFixedQueue<T, Capacity>::size_type SpscFixedQueue<T, Capacity>::size() const noexcept
{
    using const_iterator = container_type::const_iterator;
 
    auto const rItr = const_iterator(_read.itr.load(std::memory_order_acquire));
    auto const wItr = const_iterator(_write.itr.load(std::memory_order_acquire));
 
    auto const rPos = static_cast<size_type>(std::distance(std::begin(_queue), rItr)) / sizeof(T);
    auto const wPos = static_cast<size_type>(std::distance(std::begin(_queue), wItr)) / sizeof(T);
 
    return wPos >= rPos ? (wPos - rPos) : (Capacity + 1 + wPos - rPos);
}
 
template<typename T, std::size_t Capacity>
template<typename U>
requires std::is_same_v<std::decay_t<U>, T>
inline bool SpscFixedQueue<T, Capacity>::tryPush(U && value)
{
    auto const wItr = _write.itr.load(std::memory_order_relaxed);
    auto nextWItr   = wItr + sizeof(T);
    if (nextWItr == std::end(_queue))
        nextWItr = std::begin(_queue);
    auto const isQueueFull = [nextWItr, this] noexcept { return nextWItr == _read.itr.load(std::memory_order_acquire); };
 
    if (isQueueFull())
        return false;
 
    new (wItr) T(std::forward<U>(value));
 
    _write.itr.store(nextWItr, std::memory_order_release);
    return true;
}
 
template<typename T, std::size_t Capacity>
inline std::optional<T> SpscFixedQueue<T, Capacity>::tryPop()
{
    auto const rItr         = _read.itr.load(std::memory_order_relaxed);
    auto const isQueueEmpty = [&rItr, this] noexcept { return rItr == _write.itr.load(std::memory_order_acquire); };
 
    if (isQueueEmpty())
        return std::nullopt;
 
    auto nextRItr = rItr + sizeof(T);
    if (nextRItr == std::end(_queue))
        nextRItr = std::begin(_queue);
 
    auto value = std::move(*std::launder(reinterpret_cast<T *>(rItr)));
 
    if constexpr (!std::is_trivially_destructible_v<T>)
        reinterpret_cast<T *>(rItr)->~T(); // Placement new == explicit call to dtor
 
    _read.itr.store(nextRItr, std::memory_order_release);
 
    return value;
}
} // !namespace dfx::Utils