parallel_query_executor.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 "query_result_stream.h"
 
#include <kcenon/pacs/core/dicom_dataset.h>
#include <kcenon/pacs/services/query_scp.h>
 
#include <kcenon/common/patterns/result.h>
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <vector>
 
namespace kcenon::pacs::storage {
class index_database;
}  // namespace kcenon::pacs::storage
 
namespace kcenon::pacs::services {
 
template <typename T>
using Result = kcenon::common::Result<T>;
 
using VoidResult = kcenon::common::VoidResult;
 
#ifdef PACS_WITH_DATABASE_SYSTEM
 
struct query_request {
    query_level level = query_level::study;
 
    core::dicom_dataset query_keys;
 
    std::string calling_ae;
 
    std::string query_id;
 
    int priority = 0;
};
 
struct query_execution_result {
    std::string query_id;
 
    bool success = false;
 
    std::string error_message;
 
    std::unique_ptr<query_result_stream> stream;
 
    std::chrono::milliseconds execution_time{0};
 
    bool cancelled = false;
 
    bool timed_out = false;
};
 
struct parallel_executor_config {
    size_t max_concurrent = 4;
 
    std::chrono::milliseconds default_timeout{0};
 
    size_t page_size = 100;
 
    bool enable_priority = true;
};
 
class parallel_query_executor {
public:
    // =========================================================================
    // Construction
    // =========================================================================
 
    explicit parallel_query_executor(storage::index_database* db,
                                     const parallel_executor_config& config = {});
 
    ~parallel_query_executor();
 
    // Non-copyable
    parallel_query_executor(const parallel_query_executor&) = delete;
    auto operator=(const parallel_query_executor&) -> parallel_query_executor& = delete;
 
    // Movable
    parallel_query_executor(parallel_query_executor&&) noexcept;
    auto operator=(parallel_query_executor&&) noexcept -> parallel_query_executor&;
 
    // =========================================================================
    // Configuration
    // =========================================================================
 
    void set_max_concurrent(size_t max) noexcept;
 
    [[nodiscard]] auto max_concurrent() const noexcept -> size_t;
 
    void set_default_timeout(std::chrono::milliseconds timeout) noexcept;
 
    [[nodiscard]] auto default_timeout() const noexcept -> std::chrono::milliseconds;
 
    // =========================================================================
    // Batch Execution
    // =========================================================================
 
    [[nodiscard]] auto execute_all(std::vector<query_request> queries)
        -> std::vector<query_execution_result>;
 
    // =========================================================================
    // Single Query with Timeout
    // =========================================================================
 
    [[nodiscard]] auto execute_with_timeout(const query_request& query,
                                            std::chrono::milliseconds timeout)
        -> Result<std::unique_ptr<query_result_stream>>;
 
    [[nodiscard]] auto execute(const query_request& query)
        -> Result<std::unique_ptr<query_result_stream>>;
 
    // =========================================================================
    // Cancellation
    // =========================================================================
 
    void cancel_all() noexcept;
 
    [[nodiscard]] auto is_cancelled() const noexcept -> bool;
 
    void reset_cancellation() noexcept;
 
    // =========================================================================
    // Statistics
    // =========================================================================
 
    [[nodiscard]] auto queries_executed() const noexcept -> size_t;
 
    [[nodiscard]] auto queries_succeeded() const noexcept -> size_t;
 
    [[nodiscard]] auto queries_failed() const noexcept -> size_t;
 
    [[nodiscard]] auto queries_timed_out() const noexcept -> size_t;
 
    [[nodiscard]] auto queries_in_progress() const noexcept -> size_t;
 
    void reset_statistics() noexcept;
 
private:
    // =========================================================================
    // Private Implementation
    // =========================================================================
 
    [[nodiscard]] auto execute_query_internal(const query_request& query,
                                              std::chrono::milliseconds timeout)
        -> query_execution_result;
 
    [[nodiscard]] auto create_stream(const query_request& query)
        -> Result<std::unique_ptr<query_result_stream>>;
 
    // =========================================================================
    // Member Variables
    // =========================================================================
 
    storage::index_database* db_;
 
    parallel_executor_config config_;
 
    std::atomic<bool> cancelled_{false};
 
    std::atomic<size_t> queries_executed_{0};
    std::atomic<size_t> queries_succeeded_{0};
    std::atomic<size_t> queries_failed_{0};
    std::atomic<size_t> queries_timed_out_{0};
    std::atomic<size_t> queries_in_progress_{0};
 
    mutable std::mutex mutex_;
};
 
#endif  // PACS_WITH_DATABASE_SYSTEM
 
}  // namespace kcenon::pacs::services