pipeline_coordinator.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/pacs/network/pipeline/pipeline_job_types.h>
#include <kcenon/pacs/network/pipeline/metrics/pipeline_metrics.h>
#include <kcenon/pacs/integration/thread_pool_interface.h>
#include <kcenon/pacs/core/result.h>
 
#include <array>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
 
namespace kcenon::pacs::network::pipeline {
 
struct pipeline_config {
    size_t net_io_workers = 4;
 
    size_t protocol_workers = 2;
 
    size_t execution_workers = 8;
 
    size_t encode_workers = 2;
 
    size_t max_queue_depth = 10000;
 
    std::chrono::milliseconds shutdown_timeout{500};
 
    bool enable_metrics = true;
 
    std::string name_prefix = "pipeline";
 
    [[nodiscard]] auto get_workers_for_stage(pipeline_stage stage) const noexcept
        -> size_t {
        switch (stage) {
            case pipeline_stage::network_receive:
            case pipeline_stage::network_send:
                return net_io_workers;
 
            case pipeline_stage::pdu_decode:
            case pipeline_stage::dimse_process:
                return protocol_workers;
 
            case pipeline_stage::storage_query_exec:
                return execution_workers;
 
            case pipeline_stage::response_encode:
                return encode_workers;
 
            default:
                return 1;
        }
    }
};
 
// Forward declarations for job types
class pipeline_job_base;
 
class pipeline_coordinator {
public:
    using job_completion_callback = std::function<void(const job_context&, bool success)>;
 
    using backpressure_callback = std::function<void(pipeline_stage, size_t queue_depth)>;
 
    // =========================================================================
    // Construction
    // =========================================================================
 
    pipeline_coordinator();
 
    explicit pipeline_coordinator(const pipeline_config& config);
 
    ~pipeline_coordinator();
 
    // Non-copyable, non-movable
    pipeline_coordinator(const pipeline_coordinator&) = delete;
    pipeline_coordinator& operator=(const pipeline_coordinator&) = delete;
    pipeline_coordinator(pipeline_coordinator&&) = delete;
    pipeline_coordinator& operator=(pipeline_coordinator&&) = delete;
 
    // =========================================================================
    // Lifecycle Management
    // =========================================================================
 
    [[nodiscard]] auto start() -> VoidResult;
 
    [[nodiscard]] auto stop() -> VoidResult;
 
    [[nodiscard]] auto is_running() const noexcept -> bool;
 
    // =========================================================================
    // Job Submission
    // =========================================================================
 
    [[nodiscard]] auto submit_to_stage(pipeline_stage stage,
                                       std::unique_ptr<pipeline_job_base> job)
        -> VoidResult;
 
    [[nodiscard]] auto submit_to_stage(pipeline_stage stage,
                                       std::unique_ptr<pipeline_job_base> job,
                                       const job_context& ctx) -> VoidResult;
 
    [[nodiscard]] auto submit_task(pipeline_stage stage,
                                   std::function<void()> task) -> VoidResult;
 
    // =========================================================================
    // Stage Management
    // =========================================================================
 
    [[nodiscard]] auto get_stage_pool(pipeline_stage stage)
        -> std::shared_ptr<kcenon::pacs::integration::thread_pool_interface>;
 
    [[nodiscard]] auto get_queue_depth(pipeline_stage stage) const noexcept
        -> size_t;
 
    [[nodiscard]] auto is_backpressure_active(pipeline_stage stage) const noexcept
        -> bool;
 
    // =========================================================================
    // Configuration & Callbacks
    // =========================================================================
 
    [[nodiscard]] auto get_config() const noexcept -> const pipeline_config&;
 
    void set_job_completion_callback(job_completion_callback callback);
 
    void set_backpressure_callback(backpressure_callback callback);
 
    // =========================================================================
    // Metrics
    // =========================================================================
 
    [[nodiscard]] auto get_metrics() noexcept -> pipeline_metrics&;
 
    [[nodiscard]] auto get_metrics() const noexcept -> const pipeline_metrics&;
 
    void reset_metrics() noexcept;
 
    // =========================================================================
    // Statistics
    // =========================================================================
 
    [[nodiscard]] auto get_total_worker_count() const noexcept -> size_t;
 
    [[nodiscard]] auto get_total_pending_jobs() const noexcept -> size_t;
 
    [[nodiscard]] auto generate_job_id() noexcept -> uint64_t;
 
private:
    // Stage thread pool array
    static constexpr size_t stage_count =
        static_cast<size_t>(pipeline_stage::stage_count);
 
    using pool_ptr = std::shared_ptr<kcenon::pacs::integration::thread_pool_interface>;
    std::array<pool_ptr, stage_count> stage_pools_;
 
    // Configuration
    pipeline_config config_;
 
    // State
    std::atomic<bool> running_{false};
    mutable std::mutex mutex_;
 
    // Metrics
    pipeline_metrics metrics_;
 
    // Callbacks
    job_completion_callback job_completion_callback_;
    backpressure_callback backpressure_callback_;
 
    // Job ID generator
    std::atomic<uint64_t> next_job_id_{1};
 
    // Internal methods
    [[nodiscard]] auto create_stage_pool(pipeline_stage stage) -> pool_ptr;
    void notify_job_completion(const job_context& ctx, bool success);
    void check_backpressure(pipeline_stage stage);
};
 
class pipeline_job_base {
public:
    virtual ~pipeline_job_base() = default;
 
    [[nodiscard]] virtual auto execute(pipeline_coordinator& coordinator)
        -> VoidResult = 0;
 
    [[nodiscard]] virtual auto get_context() const noexcept
        -> const job_context& = 0;
 
    [[nodiscard]] virtual auto get_context() noexcept -> job_context& = 0;
 
    [[nodiscard]] virtual auto get_name() const -> std::string = 0;
 
protected:
    pipeline_job_base() = default;
 
    // Non-copyable but movable
    pipeline_job_base(const pipeline_job_base&) = delete;
    pipeline_job_base& operator=(const pipeline_job_base&) = delete;
    pipeline_job_base(pipeline_job_base&&) = default;
    pipeline_job_base& operator=(pipeline_job_base&&) = default;
};
 
}  // namespace kcenon::pacs::network::pipeline