sync_policies.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 <condition_variable>
#include <deque>
#include <memory>
#include <mutex>
 
#include <kcenon/thread/core/error_handling.h>
#include <kcenon/thread/core/job.h>
#include <kcenon/thread/core/safe_hazard_pointer.h>
#include <kcenon/thread/interfaces/queue_capabilities.h>
 
namespace kcenon::thread::policies {
 
struct sync_policy_tag {};
 
class mutex_sync_policy {
public:
    using policy_tag = sync_policy_tag;
 
    [[nodiscard]] static constexpr auto get_capabilities() -> queue_capabilities {
        return queue_capabilities{
            .exact_size = true,
            .atomic_empty_check = true,
            .lock_free = false,
            .wait_free = false,
            .supports_batch = true,
            .supports_blocking_wait = true,
            .supports_stop = true
        };
    }
 
    mutex_sync_policy() : notify_(true), stop_(false) {}
 
    ~mutex_sync_policy() = default;
 
    // Non-copyable, non-movable
    mutex_sync_policy(const mutex_sync_policy&) = delete;
    mutex_sync_policy& operator=(const mutex_sync_policy&) = delete;
    mutex_sync_policy(mutex_sync_policy&&) = delete;
    mutex_sync_policy& operator=(mutex_sync_policy&&) = delete;
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& value) -> common::VoidResult {
        if (!value) {
            return common::error_info{-105, "cannot enqueue null job", "thread_system"};
        }
 
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push_back(std::move(value));
 
        if (notify_.load(std::memory_order_relaxed)) {
            condition_.notify_one();
        }
 
        return common::ok();
    }
 
    [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>> {
        std::unique_lock<std::mutex> lock(mutex_);
 
        condition_.wait(lock, [this] {
            return !queue_.empty() || stop_.load(std::memory_order_relaxed);
        });
 
        if (queue_.empty()) {
            return common::error_info{-121, "queue is stopped or empty", "thread_system"};
        }
 
        auto value = std::move(queue_.front());
        queue_.pop_front();
        return value;
    }
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>> {
        std::lock_guard<std::mutex> lock(mutex_);
 
        if (queue_.empty()) {
            return common::error_info{-121, "queue is empty", "thread_system"};
        }
 
        auto value = std::move(queue_.front());
        queue_.pop_front();
        return value;
    }
 
    [[nodiscard]] auto empty() const -> bool {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }
 
    [[nodiscard]] auto size() const -> std::size_t {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.size();
    }
 
    auto clear() -> void {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.clear();
    }
 
    auto stop() -> void {
        stop_.store(true, std::memory_order_release);
        condition_.notify_all();
    }
 
    [[nodiscard]] auto is_stopped() const -> bool {
        return stop_.load(std::memory_order_acquire);
    }
 
    auto set_notify(bool notify) -> void {
        notify_.store(notify, std::memory_order_relaxed);
    }
 
private:
    mutable std::mutex mutex_;
    std::condition_variable condition_;
    std::deque<std::unique_ptr<job>> queue_;
    std::atomic<bool> notify_;
    std::atomic<bool> stop_;
};
 
class lockfree_sync_policy {
public:
    using policy_tag = sync_policy_tag;
 
    [[nodiscard]] static constexpr auto get_capabilities() -> queue_capabilities {
        return queue_capabilities{
            .exact_size = false,
            .atomic_empty_check = false,
            .lock_free = true,
            .wait_free = false,
            .supports_batch = false,
            .supports_blocking_wait = false,
            .supports_stop = false
        };
    }
 
    lockfree_sync_policy() : shutdown_(false), approximate_size_(0) {
        auto dummy = new node(nullptr);
        head_.store(dummy, std::memory_order_relaxed);
        tail_.store(dummy, std::memory_order_relaxed);
    }
 
    ~lockfree_sync_policy() {
        shutdown_.store(true, std::memory_order_release);
 
        // Drain remaining nodes
        node* current = head_.load(std::memory_order_acquire);
        while (current != nullptr) {
            node* next = current->next.load(std::memory_order_acquire);
            delete current;
            current = next;
        }
    }
 
    // Non-copyable, non-movable
    lockfree_sync_policy(const lockfree_sync_policy&) = delete;
    lockfree_sync_policy& operator=(const lockfree_sync_policy&) = delete;
    lockfree_sync_policy(lockfree_sync_policy&&) = delete;
    lockfree_sync_policy& operator=(lockfree_sync_policy&&) = delete;
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& value) -> common::VoidResult {
        if (!value) {
            return common::error_info{-105, "cannot enqueue null job", "thread_system"};
        }
 
        if (shutdown_.load(std::memory_order_acquire)) {
            return common::error_info{-122, "queue is shutting down", "thread_system"};
        }
 
        auto new_node = new node(std::move(value));
 
        while (true) {
            node* tail = tail_.load(std::memory_order_acquire);
            node* next = tail->next.load(std::memory_order_acquire);
 
            if (tail == tail_.load(std::memory_order_acquire)) {
                if (next == nullptr) {
                    if (tail->next.compare_exchange_weak(next, new_node,
                            std::memory_order_release,
                            std::memory_order_relaxed)) {
                        tail_.compare_exchange_strong(tail, new_node,
                            std::memory_order_release,
                            std::memory_order_relaxed);
                        approximate_size_.fetch_add(1, std::memory_order_relaxed);
                        return common::ok();
                    }
                } else {
                    tail_.compare_exchange_weak(tail, next,
                        std::memory_order_release,
                        std::memory_order_relaxed);
                }
            }
        }
    }
 
    [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>> {
        while (true) {
            node* head = head_.load(std::memory_order_acquire);
            node* tail = tail_.load(std::memory_order_acquire);
            node* next = head->next.load(std::memory_order_acquire);
 
            if (head == head_.load(std::memory_order_acquire)) {
                if (head == tail) {
                    if (next == nullptr) {
                        return common::error_info{-121, "queue is empty", "thread_system"};
                    }
                    tail_.compare_exchange_weak(tail, next,
                        std::memory_order_release,
                        std::memory_order_relaxed);
                } else {
                    if (next != nullptr) {
                        auto value = std::move(next->data);
                        if (head_.compare_exchange_weak(head, next,
                                std::memory_order_release,
                                std::memory_order_relaxed)) {
                            approximate_size_.fetch_sub(1, std::memory_order_relaxed);
                            delete head;
                            return value;
                        }
                    }
                }
            }
        }
    }
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>> {
        return dequeue();
    }
 
    [[nodiscard]] auto empty() const -> bool {
        node* head = head_.load(std::memory_order_acquire);
        node* next = head->next.load(std::memory_order_acquire);
        return next == nullptr;
    }
 
    [[nodiscard]] auto size() const -> std::size_t {
        return approximate_size_.load(std::memory_order_relaxed);
    }
 
    auto clear() -> void {
        while (true) {
            auto result = dequeue();
            if (result.is_err()) {
                break;
            }
        }
    }
 
    auto stop() -> void {
        shutdown_.store(true, std::memory_order_release);
    }
 
    [[nodiscard]] auto is_stopped() const -> bool {
        return shutdown_.load(std::memory_order_acquire);
    }
 
    auto set_notify(bool /*notify*/) -> void {
        // No-op: lock-free queue doesn't use condition variables
    }
 
private:
    struct node {
        std::unique_ptr<job> data;
        std::atomic<node*> next{nullptr};
 
        explicit node(std::unique_ptr<job>&& job_data)
            : data(std::move(job_data)) {}
    };
 
    std::atomic<node*> head_;
    std::atomic<node*> tail_;
    std::atomic<bool> shutdown_;
    mutable std::atomic<std::size_t> approximate_size_;
};
 
class adaptive_sync_policy {
public:
    using policy_tag = sync_policy_tag;
 
    enum class mode {
        mutex,     
        lock_free  
    };
 
    [[nodiscard]] auto get_capabilities() const -> queue_capabilities {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_sync_policy::get_capabilities();
        }
        return lockfree_sync_policy::get_capabilities();
    }
 
    explicit adaptive_sync_policy(mode initial_mode = mode::mutex)
        : current_mode_(initial_mode)
        , mutex_policy_(std::make_unique<mutex_sync_policy>())
        , lockfree_policy_(std::make_unique<lockfree_sync_policy>()) {}
 
    ~adaptive_sync_policy() = default;
 
    // Non-copyable, non-movable
    adaptive_sync_policy(const adaptive_sync_policy&) = delete;
    adaptive_sync_policy& operator=(const adaptive_sync_policy&) = delete;
    adaptive_sync_policy(adaptive_sync_policy&&) = delete;
    adaptive_sync_policy& operator=(adaptive_sync_policy&&) = delete;
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& value) -> common::VoidResult {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_policy_->enqueue(std::move(value));
        }
        return lockfree_policy_->enqueue(std::move(value));
    }
 
    [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>> {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_policy_->dequeue();
        }
        return lockfree_policy_->dequeue();
    }
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>> {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_policy_->try_dequeue();
        }
        return lockfree_policy_->try_dequeue();
    }
 
    [[nodiscard]] auto empty() const -> bool {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_policy_->empty();
        }
        return lockfree_policy_->empty();
    }
 
    [[nodiscard]] auto size() const -> std::size_t {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_policy_->size();
        }
        return lockfree_policy_->size();
    }
 
    auto clear() -> void {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            mutex_policy_->clear();
        } else {
            lockfree_policy_->clear();
        }
    }
 
    auto stop() -> void {
        mutex_policy_->stop();
        lockfree_policy_->stop();
    }
 
    [[nodiscard]] auto is_stopped() const -> bool {
        if (current_mode_.load(std::memory_order_acquire) == mode::mutex) {
            return mutex_policy_->is_stopped();
        }
        return lockfree_policy_->is_stopped();
    }
 
    auto set_notify(bool notify) -> void {
        mutex_policy_->set_notify(notify);
    }
 
    [[nodiscard]] auto current_mode() const -> mode {
        return current_mode_.load(std::memory_order_acquire);
    }
 
    auto switch_mode(mode target_mode) -> void {
        current_mode_.store(target_mode, std::memory_order_release);
    }
 
private:
    std::atomic<mode> current_mode_;
    std::unique_ptr<mutex_sync_policy> mutex_policy_;
    std::unique_ptr<lockfree_sync_policy> lockfree_policy_;
};
 
} // namespace kcenon::thread::policies