concurrent_queue.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <mutex>
#include <optional>
#include <type_traits>
 
namespace kcenon::thread {
 
namespace detail {
 
template <typename T>
class concurrent_queue {
    static_assert(std::is_move_constructible_v<T>, "T must be move constructible");
 
public:
    concurrent_queue() {
        auto* dummy = new node{};
        head_ = dummy;
        tail_ = dummy;
    }
 
    ~concurrent_queue() {
        shutdown();
        // Drain remaining items
        while (try_dequeue()) {
        }
        // Delete the dummy node
        delete head_;
    }
 
    // Non-copyable and non-movable
    concurrent_queue(const concurrent_queue&) = delete;
    concurrent_queue& operator=(const concurrent_queue&) = delete;
    concurrent_queue(concurrent_queue&&) = delete;
    concurrent_queue& operator=(concurrent_queue&&) = delete;
 
    void enqueue(T value) {
        if (shutdown_.load(std::memory_order_acquire)) {
            return;
        }
 
        auto* new_node = new node{std::move(value)};
 
        {
            std::lock_guard<std::mutex> lock(tail_mutex_);
            tail_->next = new_node;
            tail_ = new_node;
        }
 
        size_.fetch_add(1, std::memory_order_release);
        notify_one();
    }
 
    [[nodiscard]] auto try_dequeue() -> std::optional<T> {
        std::lock_guard<std::mutex> lock(head_mutex_);
 
        node* old_head = head_;
        node* next = old_head->next;
 
        if (next == nullptr) {
            return std::nullopt;  // Queue is empty
        }
 
        // Extract value from next node
        std::optional<T> value;
        if (next->data.has_value()) {
            value = std::move(*next->data);
            next->data.reset();
        }
 
        // Advance head (old_head becomes the new dummy)
        head_ = next;
        delete old_head;
 
        size_.fetch_sub(1, std::memory_order_release);
        return value;
    }
 
    template <typename Rep, typename Period>
    [[nodiscard]] auto wait_dequeue(const std::chrono::duration<Rep, Period>& timeout)
        -> std::optional<T> {
        // First try without waiting
        if (auto value = try_dequeue()) {
            return value;
        }
 
        // Wait for notification
        std::unique_lock<std::mutex> lock(cv_mutex_);
        auto deadline = std::chrono::steady_clock::now() + timeout;
 
        while (!shutdown_.load(std::memory_order_acquire)) {
            // Try to dequeue
            lock.unlock();
            if (auto value = try_dequeue()) {
                return value;
            }
            lock.lock();
 
            // Check if we should stop waiting
            if (std::chrono::steady_clock::now() >= deadline) {
                return std::nullopt;
            }
 
            // Wait for notification or timeout
            cv_.wait_until(lock, deadline);
        }
 
        // Final attempt after shutdown
        return try_dequeue();
    }
 
    [[nodiscard]] auto wait_dequeue() -> std::optional<T> {
        return wait_dequeue(std::chrono::hours{24 * 365});
    }
 
    [[nodiscard]] auto empty() const noexcept -> bool {
        return size_.load(std::memory_order_acquire) == 0;
    }
 
    [[nodiscard]] auto size() const noexcept -> std::size_t {
        return size_.load(std::memory_order_acquire);
    }
 
    void notify_one() {
        std::lock_guard<std::mutex> lock(cv_mutex_);
        cv_.notify_one();
    }
 
    void notify_all() {
        std::lock_guard<std::mutex> lock(cv_mutex_);
        cv_.notify_all();
    }
 
    void shutdown() {
        shutdown_.store(true, std::memory_order_release);
        notify_all();
    }
 
    [[nodiscard]] auto is_shutdown() const noexcept -> bool {
        return shutdown_.load(std::memory_order_acquire);
    }
 
private:
    struct node {
        std::optional<T> data;
        node* next{nullptr};
 
        node() = default;
        explicit node(T value) : data(std::move(value)) {}
    };
 
    node* head_;
    node* tail_;
 
    mutable std::mutex head_mutex_;
    mutable std::mutex tail_mutex_;
 
    std::atomic<std::size_t> size_{0};
    std::atomic<bool> shutdown_{false};
 
    // For blocking wait support
    mutable std::mutex cv_mutex_;
    std::condition_variable cv_;
};
 
}  // namespace detail
 
}  // namespace kcenon::thread