coordinator.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/types.h>
#include <neograph/graph/checkpoint.h>
#include <neograph/graph/scheduler.h>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
 
namespace neograph::graph {
 
class GraphState;
 
struct ResumeContext {
    bool have_cp = false;
 
    std::string checkpoint_id;
    json channel_values;         
    int start_step = 0;          
    CheckpointPhase phase = CheckpointPhase::Completed;
    std::vector<std::string> next_nodes;
 
    std::unordered_map<std::string, NodeResult> replay_results;
 
    BarrierState barrier_state;
};
 
class NEOGRAPH_API CheckpointCoordinator {
public:
    CheckpointCoordinator(std::shared_ptr<CheckpointStore> store,
                          std::string thread_id);
 
    bool enabled() const noexcept { return store_ != nullptr && !thread_id_.empty(); }
 
    const std::shared_ptr<CheckpointStore>& store() const noexcept { return store_; }
    const std::string& thread_id() const noexcept { return thread_id_; }
 
    std::string save_super_step(const GraphState& state,
                                const std::string& current_node,
                                const std::vector<std::string>& next_nodes,
                                CheckpointPhase phase,
                                int step,
                                const std::string& parent_id,
                                const BarrierState& barrier_state) const;
 
    ResumeContext load_for_resume() const;
 
    asio::awaitable<ResumeContext> load_for_resume_async() const;
 
    void record_pending_write(const std::string& parent_cp_id,
                              const std::string& task_id,
                              const std::string& task_path,
                              const std::string& node_name,
                              const NodeResult& nr,
                              int step) const;
 
    void clear_pending_writes(const std::string& parent_cp_id) const;
 
    // ── Async peers (Stage 3 / Sem 3.6 incremental) ─────────────────────
    //
    // Each routes to the matching CheckpointStore::*_async so the
    // coroutine engine path doesn't block the io_context on
    // checkpoint I/O. Behaviour identical to the sync versions.
 
    asio::awaitable<std::string> save_super_step_async(
        const GraphState& state,
        const std::string& current_node,
        const std::vector<std::string>& next_nodes,
        CheckpointPhase phase,
        int step,
        const std::string& parent_id,
        const BarrierState& barrier_state) const;
 
    asio::awaitable<void> record_pending_write_async(
        const std::string& parent_cp_id,
        const std::string& task_id,
        const std::string& task_path,
        const std::string& node_name,
        const NodeResult& nr,
        int step) const;
 
    asio::awaitable<void> clear_pending_writes_async(
        const std::string& parent_cp_id) const;
 
private:
    std::shared_ptr<CheckpointStore> store_;
    std::string thread_id_;
};
 
} // namespace neograph::graph