SyncQueue.hpp

// SPDX-FileCopyrightText: 2025 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 <sys/eventfd.h>
#include <unistd.h>
 
// C++ includes
#include <array>
#include <atomic>
#include <cerrno>
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <mutex>
#include <new>
#include <optional>
#include <type_traits>
 
// Project includes
#include <dfx-utilities/Exception.hpp>
#include <dfx-utilities/CompilerSupport.hpp>
 
namespace dfx::Core::details
{
struct noOpMutex
{
    constexpr void lock() noexcept {}
    constexpr void unlock() noexcept {}
    constexpr bool try_lock() noexcept { return true; }
};
 
inline ALWAYS_INLINE_ATTR void cpuRelax() noexcept
{
#ifdef PLATFORM_ARCH_X86
    __builtin_ia32_pause();
#elif PLATFORM_ARCH_ARM
    asm volatile("yield" ::: "memory");
#endif
}
 
template<typename T, bool sp, std::size_t Capacity = 1'000>
class SyncQueue
{
    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 mutex_type      = std::mutex;
    using pmutex_type     = std::conditional_t<sp, noOpMutex, std::mutex>;
    using value_type      = T;
    using size_type       = typename container_type::size_type;
    using reference       = value_type &;
    using const_reference = value_type const &;
    using opt_value_type  = std::optional<std::reference_wrapper<value_type>>;
 
public:
    SyncQueue();
    SyncQueue(SyncQueue<T, sp, Capacity> const &) = delete;
    SyncQueue(SyncQueue<T, sp, Capacity> &&) = delete;
    ~SyncQueue();
 
    SyncQueue<T, sp, Capacity> & operator=(SyncQueue<T, sp, Capacity> const &) = delete;
    SyncQueue<T, sp, Capacity> & operator=(SyncQueue<T, sp, Capacity> &&) = delete;
 
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; }
 
    int createReadEventFd(bool nonblock = false);
    [[nodiscard]] int readEventFd() const noexcept { return _readEventFd; }
 
    void forceQuit() noexcept;
 
public:
    template<typename ... Args>
    void emplace(Args && ... args);
 
    [[nodiscard]] opt_value_type front();
    [[nodiscard]] opt_value_type tryFront() noexcept;
    void pop();
 
public:
    void setSpinWaitCount(uint32_t counter) noexcept { _spinWaitCounter = counter; }
    void disableSpinWait() noexcept                  { setSpinWaitCount(0); }
    uint32_t spinWaitCount() const noexcept          { return _spinWaitCounter; }
 
private:
    void _notifyReader() noexcept;
    void _notifyWriter() noexcept;
 
private:
    alignas(T) container_type _queue;
 
private:
    using atomic_itr = std::atomic<typename container_type::iterator>;
    static_assert(atomic_itr::is_always_lock_free);
 
    static constexpr auto alignSize = 64;
 
    struct alignas(alignSize) ReadSide
    {
        atomic_itr itr;
        std::atomic<size_type> snapshot{0};
    } _read;
 
    struct alignas(alignSize) WriteSide
    {
        atomic_itr itr;
        std::atomic<size_type> snapshot{0};
    } _write;
 
    struct alignas(alignSize) Flags
    {
        std::atomic_bool forceQuit = false;
        std::atomic_bool hasWriteWaiter = false;
        std::atomic_bool hasReadWaiter = false;
    } _flags;
 
private:
    mutex_type              _mutex;
    pmutex_type             _producerMutex;
    std::condition_variable _cvReader;
    std::condition_variable _cvWriter;
 
private:
    uint32_t _spinWaitCounter = 1'000;
    int _readEventFd = -1;
};
 
template<typename T, bool sp, std::size_t Capacity>
SyncQueue<T, sp, Capacity>::SyncQueue()
    : _read  { std::begin(_queue) }
    , _write { std::begin(_queue) }
{
}
 
template<typename T, bool sp, std::size_t Capacity>
SyncQueue<T, sp, Capacity>::~SyncQueue()
{
    _flags.forceQuit.store(true, std::memory_order_release);
    if (_readEventFd != -1)
        close(_readEventFd);
 
    _notifyReader();
    _notifyWriter();
 
    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, bool sp, std::size_t Capacity>
bool SyncQueue<T, sp, Capacity>::empty() const noexcept
{
    return size() == 0;
}
 
template<typename T, bool sp, std::size_t Capacity>
bool SyncQueue<T, sp, Capacity>::full() const noexcept
{
    return size() == capacity();
}
 
template<typename T, bool sp, std::size_t Capacity>
typename SyncQueue<T, sp, Capacity>::size_type SyncQueue<T, sp, Capacity>::size() const noexcept
{
    auto const rPos = _read.snapshot.load(std::memory_order_acquire) / sizeof(T);
    auto const wPos = _write.snapshot.load(std::memory_order_acquire) / sizeof(T);
 
    return wPos >= rPos ? (wPos - rPos) : (Capacity + 1 + wPos - rPos);
}
 
template<typename T, bool sp, std::size_t Capacity>
int SyncQueue<T, sp, Capacity>::createReadEventFd(bool nonblock)
{
    if (_readEventFd != -1)
        return _readEventFd;
 
    int flags = EFD_CLOEXEC;
    if (nonblock)
        flags |= EFD_NONBLOCK;
 
    _readEventFd = eventfd(size(), flags);
    B_ASSERT(_readEventFd != -1, "Failed to create eventfd: {}", strerror(errno));
    return _readEventFd;
}
 
template<typename T, bool sp, std::size_t Capacity>
void SyncQueue<T, sp, Capacity>::forceQuit() noexcept
{
    {
        std::lock_guard<std::mutex> lock(_mutex);
        _flags.forceQuit.store(true, std::memory_order_release);
    }
 
    _cvReader.notify_all();
    _cvWriter.notify_all();
}
 
template<typename T, bool sp, std::size_t Capacity>
template<typename ... Args>
void SyncQueue<T, sp, Capacity>::emplace(Args && ... args)
{
    std::unique_lock<pmutex_type> pLock(_producerMutex, std::defer_lock);
    if constexpr (!sp)
        pLock.lock();
 
    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())
    {
        for (uint32_t counter = _spinWaitCounter; counter > 0 && isQueueFull(); --counter)
            cpuRelax();
 
        if (isQueueFull())
        {
            std::unique_lock<mutex_type> lock(_mutex);
            _flags.hasWriteWaiter.store(true, std::memory_order_release);
            _cvWriter.wait(lock, [&isQueueFull, this]() noexcept
            {
                // Stop waiting if not full anymore or force quit
                return !isQueueFull() || UNLIKELY(_flags.forceQuit.load(std::memory_order_acquire));
            });
            // No need to worry about RAII here as everything is marked noexcept
            _flags.hasWriteWaiter.store(false, std::memory_order_release);
 
            if (UNLIKELY(_flags.forceQuit.load(std::memory_order_relaxed)))
                return ;
        }
    }
 
    new (wItr) T(std::forward<Args>(args)...);
 
    {
        // The mutex has to be locked here even if we use an atomic
        // otherwise the following scenario can occur (and in fact has):
        //  - thread A waiting on _cvReader wakes spuriously
        //  - thread B call _write.writeItr store function
        //  - thread B call _cvReader notify_one function
        //  - thread A waits on _cvReader and so the notification has been missed
        // c.f. https://stackoverflow.com/a/36130475/3368370
        std::lock_guard<mutex_type> lock(_mutex);
        _write.itr.store(nextWItr, std::memory_order_release);
 
        auto offset = static_cast<size_type>(std::distance(std::begin(_queue), nextWItr));
        offset %= (Capacity + 1) * sizeof(T);
        _write.snapshot.store(offset, std::memory_order_release);
    }
 
    if (_flags.hasReadWaiter.load(std::memory_order_acquire) || _readEventFd != -1)
        _notifyReader();
}
 
template<typename T, bool sp, std::size_t Capacity>
typename SyncQueue<T, sp, Capacity>::opt_value_type SyncQueue<T, sp, Capacity>::front()
{
    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())
    {
        for (uint32_t counter = _spinWaitCounter; counter > 0 && isQueueEmpty(); --counter)
            cpuRelax();
 
        if (isQueueEmpty())
        {
            std::unique_lock<mutex_type> lock(_mutex);
            _flags.hasReadWaiter.store(true, std::memory_order_release);
            _cvReader.wait(lock, [&isQueueEmpty, this]() noexcept
            {
                // Stop waiting if not empty anymore or force quit
                return !isQueueEmpty() || UNLIKELY(_flags.forceQuit.load(std::memory_order_acquire));
            });
            // No need to worry about RAII here as everything is marked noexcept
            _flags.hasReadWaiter.store(false, std::memory_order_release);
 
            if (UNLIKELY(_flags.forceQuit.load(std::memory_order_relaxed)))
                return {};
        }
    }
 
    return *std::launder(reinterpret_cast<T *>(rItr));
}
 
template<typename T, bool sp, std::size_t Capacity>
typename SyncQueue<T, sp, Capacity>::opt_value_type SyncQueue<T, sp, Capacity>::tryFront() noexcept
{
    auto const rItr = _read.itr.load(std::memory_order_relaxed);
    if (rItr == _write.itr.load(std::memory_order_acquire)) // Queue is empty
        return {};
 
    return *std::launder(reinterpret_cast<T *>(rItr));
}
 
template<typename T, bool sp, std::size_t Capacity>
void SyncQueue<T, sp, Capacity>::pop()
{
    auto const rItr = _read.itr.load(std::memory_order_relaxed);
    B_ASSERT(rItr != _write.itr.load(std::memory_order_acquire), "Trying to pop an empty queue");
 
    auto nextRItr = rItr + sizeof(T);
    if (nextRItr == std::end(_queue))
        nextRItr = std::begin(_queue);
 
    if constexpr (!std::is_trivially_destructible_v<T>)
        reinterpret_cast<T *>(rItr)->~T(); // Placement new == explicit call to dtor
 
    {
        // The mutex has to be locked here even if we use an atomic
        // otherwise the following scenario can occur (and in fact has):
        //  - thread A waiting on _cvWriter wakes spuriously
        //  - thread B call _read.readItr store function
        //  - thread B call _cvWriter notify_one function
        //  - thread A waits on _cvWriter and so the notification has been missed
        // c.f. https://stackoverflow.com/a/36130475/3368370
        std::lock_guard<mutex_type> lock(_mutex);
        _read.itr.store(nextRItr, std::memory_order_release);
 
        auto offset = static_cast<size_type>(std::distance(std::begin(_queue), nextRItr));
        offset %= (Capacity + 1) * sizeof(T);
        _read.snapshot.store(offset, std::memory_order_release);
    }
 
    if (_flags.hasWriteWaiter.load(std::memory_order_acquire))
        _notifyWriter();
}
 
template<typename T, bool sp, std::size_t Capacity>
void SyncQueue<T, sp, Capacity>::_notifyReader() noexcept
{
    _cvReader.notify_one();
 
    if (_readEventFd != -1)
    {
        uint64_t const val = 1;
        write(_readEventFd, &val, sizeof(val));
    }
}
 
template<typename T, bool sp, std::size_t Capacity>
void SyncQueue<T, sp, Capacity>::_notifyWriter() noexcept
{
    _cvWriter.notify_one();
}
} // !namespace dfx::Core::details