lockfree_queue.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <atomic>
#include <memory>
#include <array>
#include <type_traits>
 
namespace kcenon::logger::async {
 
template<typename T, size_t Size>
class lockfree_spsc_queue {
    static_assert((Size & (Size - 1)) == 0, "Size must be a power of 2");
    static_assert(Size > 1, "Size must be greater than 1");
 
public:
    lockfree_spsc_queue()
        : head_{0}
        , tail_{0} {
        // Initialize cell sequence numbers for ABA prevention
        for (size_t i = 0; i < Size; ++i) {
            cells_[i].sequence.store(i, std::memory_order_relaxed);
        }
    }
 
    ~lockfree_spsc_queue() {
        T item;
        while (dequeue(item)) {
            // Clean up remaining items
        }
    }
 
    bool enqueue(const T& item) {
        return enqueue_impl(item);
    }
 
    bool enqueue(T&& item) {
        return enqueue_impl(std::move(item));
    }
 
    bool dequeue(T& item) {
        const size_t pos = tail_.load(std::memory_order_relaxed);
        auto& cell = cells_[pos & mask_];
 
        const size_t seq = cell.sequence.load(std::memory_order_acquire);
        const size_t expected_seq = pos + 1;
 
        if (seq != expected_seq) {
            return false; // Queue is empty
        }
 
        item = std::move(cell.data);
        cell.sequence.store(pos + mask_ + 1, std::memory_order_release);
        tail_.store(pos + 1, std::memory_order_relaxed);
 
        return true;
    }
 
    bool empty() const {
        const size_t head_pos = head_.load(std::memory_order_acquire);
        const size_t tail_pos = tail_.load(std::memory_order_acquire);
        return head_pos == tail_pos;
    }
 
    bool full() const {
        const size_t head_pos = head_.load(std::memory_order_acquire);
        const size_t tail_pos = tail_.load(std::memory_order_acquire);
        return ((head_pos + 1) & mask_) == (tail_pos & mask_);
    }
 
    size_t size() const {
        const size_t head_pos = head_.load(std::memory_order_acquire);
        const size_t tail_pos = tail_.load(std::memory_order_acquire);
        return head_pos - tail_pos;
    }
 
    constexpr size_t capacity() const {
        return Size;
    }
 
private:
    static constexpr size_t mask_ = Size - 1;
 
    static constexpr size_t cache_line_size = 64;
 
    struct alignas(cache_line_size) cell {
        std::atomic<size_t> sequence;
        T data;
 
        cell() : sequence{0} {}
    };
 
    template<typename U>
    bool enqueue_impl(U&& item) {
        const size_t pos = head_.load(std::memory_order_relaxed);
        auto& cell = cells_[pos & mask_];
 
        const size_t seq = cell.sequence.load(std::memory_order_acquire);
        const size_t expected_seq = pos;
 
        if (seq != expected_seq) {
            return false; // Queue is full
        }
 
        cell.data = std::forward<U>(item);
        cell.sequence.store(pos + 1, std::memory_order_release);
        head_.store(pos + 1, std::memory_order_relaxed);
 
        return true;
    }
 
    // Align to cache line boundaries to prevent false sharing
    alignas(cache_line_size) std::atomic<size_t> head_;
    alignas(cache_line_size) std::atomic<size_t> tail_;
    alignas(cache_line_size) std::array<cell, Size> cells_;
};
 
template<typename T, size_t Size = 1024>
std::unique_ptr<lockfree_spsc_queue<T, Size>> make_lockfree_queue() {
    return std::make_unique<lockfree_spsc_queue<T, Size>>();
}
 
template<typename T, size_t Size>
class lockfree_mpmc_queue {
    // TODO: Implement if multi-producer scenarios are needed
    static_assert(sizeof(T) == 0, "MPMC queue not yet implemented");
};
 
} // namespace kcenon::logger::async