dag_job.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 <kcenon/thread/core/job.h>
#include <kcenon/thread/core/error_handling.h>
 
#include <any>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <vector>
 
namespace kcenon::thread
{
    using job_id = std::uint64_t;
 
    constexpr job_id INVALID_JOB_ID = 0;
 
    enum class dag_job_state
    {
        pending,    
        ready,      
        running,    
        completed,  
        failed,     
        cancelled,  
        skipped     
    };
 
    [[nodiscard]] inline auto dag_job_state_to_string(dag_job_state state) -> std::string
    {
        switch (state)
        {
            case dag_job_state::pending:   return "pending";
            case dag_job_state::ready:     return "ready";
            case dag_job_state::running:   return "running";
            case dag_job_state::completed: return "completed";
            case dag_job_state::failed:    return "failed";
            case dag_job_state::cancelled: return "cancelled";
            case dag_job_state::skipped:   return "skipped";
            default:                       return "unknown";
        }
    }
 
    struct dag_job_info
    {
        job_id id{INVALID_JOB_ID};                       
        std::string name;                                 
        dag_job_state state{dag_job_state::pending};     
 
        std::vector<job_id> dependencies;                 
        std::vector<job_id> dependents;                   
 
        std::chrono::steady_clock::time_point submit_time;  
        std::chrono::steady_clock::time_point start_time;   
        std::chrono::steady_clock::time_point end_time;     
 
        std::optional<std::string> error_message;         
        std::optional<std::any> result;                   
 
        [[nodiscard]] auto get_wait_time() const -> std::chrono::milliseconds
        {
            if (start_time == std::chrono::steady_clock::time_point{})
            {
                return std::chrono::milliseconds{0};
            }
            return std::chrono::duration_cast<std::chrono::milliseconds>(start_time - submit_time);
        }
 
        [[nodiscard]] auto get_execution_time() const -> std::chrono::milliseconds
        {
            if (end_time == std::chrono::steady_clock::time_point{} ||
                start_time == std::chrono::steady_clock::time_point{})
            {
                return std::chrono::milliseconds{0};
            }
            return std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time);
        }
    };
 
    // Forward declaration
    class dag_scheduler;
 
    class dag_job : public job
    {
    public:
        explicit dag_job(const std::string& name = "dag_job");
 
        ~dag_job() override;
 
        [[nodiscard]] auto get_dag_id() const -> job_id { return dag_id_; }
 
        [[nodiscard]] auto get_state() const -> dag_job_state
        {
            return state_.load(std::memory_order_acquire);
        }
 
        auto set_state(dag_job_state new_state) -> void
        {
            state_.store(new_state, std::memory_order_release);
        }
 
        [[nodiscard]] auto try_transition_state(dag_job_state expected, dag_job_state desired) -> bool
        {
            return state_.compare_exchange_strong(expected, desired,
                                                  std::memory_order_acq_rel,
                                                  std::memory_order_acquire);
        }
 
        [[nodiscard]] auto get_dependencies() const -> const std::vector<job_id>& { return dependencies_; }
 
        auto add_dependency(job_id dependency_id) -> void
        {
            if (dependency_id != INVALID_JOB_ID)
            {
                dependencies_.push_back(dependency_id);
            }
        }
 
        auto add_dependencies(const std::vector<job_id>& dependency_ids) -> void
        {
            for (const auto& id : dependency_ids)
            {
                add_dependency(id);
            }
        }
 
        auto set_work(std::function<common::VoidResult()> work_func) -> void
        {
            work_func_ = std::move(work_func);
        }
 
        template<typename T>
        auto set_work_with_result(std::function<common::Result<T>()> work_func) -> void
        {
            work_func_ = [this, func = std::move(work_func)]() -> common::VoidResult {
                auto result = func();
                if (result.is_ok())
                {
                    set_result(result.value());
                    return common::ok();
                }
                return common::VoidResult(result.error());
            };
        }
 
        auto set_fallback(std::function<common::VoidResult()> fallback_func) -> void
        {
            fallback_func_ = std::move(fallback_func);
        }
 
        [[nodiscard]] auto get_fallback() const -> const std::function<common::VoidResult()>&
        {
            return fallback_func_;
        }
 
        [[nodiscard]] auto has_fallback() const -> bool
        {
            return fallback_func_ != nullptr;
        }
 
        template<typename T>
        auto set_result(T&& value) -> void
        {
            result_ = std::forward<T>(value);
        }
 
        template<typename T>
        [[nodiscard]] auto get_result() const -> const T&
        {
            return std::any_cast<const T&>(result_);
        }
 
        [[nodiscard]] auto has_result() const -> bool
        {
            return result_.has_value();
        }
 
        [[nodiscard]] auto get_result_any() const -> const std::any& { return result_; }
 
        auto set_error_message(const std::string& message) -> void
        {
            error_message_ = message;
        }
 
        [[nodiscard]] auto get_error_message() const -> const std::optional<std::string>&
        {
            return error_message_;
        }
 
        auto record_start_time() -> void
        {
            start_time_ = std::chrono::steady_clock::now();
        }
 
        auto record_end_time() -> void
        {
            end_time_ = std::chrono::steady_clock::now();
        }
 
        [[nodiscard]] auto get_submit_time() const -> std::chrono::steady_clock::time_point
        {
            return submit_time_;
        }
 
        [[nodiscard]] auto get_start_time() const -> std::chrono::steady_clock::time_point
        {
            return start_time_;
        }
 
        [[nodiscard]] auto get_end_time() const -> std::chrono::steady_clock::time_point
        {
            return end_time_;
        }
 
        [[nodiscard]] auto get_info() const -> dag_job_info;
 
        [[nodiscard]] auto do_work() -> common::VoidResult override;
 
        [[nodiscard]] auto to_string() const -> std::string override;
 
    private:
        static std::atomic<job_id> next_dag_id_;
 
        job_id dag_id_;
 
        std::atomic<dag_job_state> state_{dag_job_state::pending};
 
        std::vector<job_id> dependencies_;
 
        std::function<common::VoidResult()> work_func_;
 
        std::function<common::VoidResult()> fallback_func_;
 
        std::any result_;
 
        std::optional<std::string> error_message_;
 
        std::chrono::steady_clock::time_point submit_time_;
 
        std::chrono::steady_clock::time_point start_time_;
 
        std::chrono::steady_clock::time_point end_time_;
    };
 
} // namespace kcenon::thread
 
// ----------------------------------------------------------------------------
// Formatter specializations for dag_job
// ----------------------------------------------------------------------------
 
#include <kcenon/thread/utils/formatter_macros.h>
 
DECLARE_FORMATTER(kcenon::thread::dag_job)