executor.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/types.h>
#include <neograph/graph/node.h>
#include <neograph/graph/compiler.h>
#include <neograph/graph/coordinator.h>
#include <neograph/graph/node_cache.h>
#include <neograph/graph/scheduler.h>
 
#include <asio/thread_pool.hpp>
 
#include <atomic>
#include <functional>
#include <map>
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>
 
namespace neograph::graph {
 
class GraphState;
 
struct RunContext;
 
class NEOGRAPH_API NodeExecutor {
public:
    using RetryPolicyLookup = std::function<RetryPolicy(const std::string&)>;
 
    NodeExecutor(
        const std::map<std::string, std::unique_ptr<GraphNode>>& nodes,
        const std::vector<ChannelDef>& channel_defs,
        RetryPolicyLookup retry_policy_for,
        asio::thread_pool* fan_out_pool = nullptr,
        NodeCache* node_cache = nullptr);
 
    asio::awaitable<NodeResult> run_one_async(
        const std::string& node_name,
        int step,
        GraphState& state,
        const std::unordered_map<std::string, NodeResult>& replay,
        CheckpointCoordinator& coord,
        const std::string& parent_cp_id,
        const BarrierState& barrier_state,
        std::vector<std::string>& trace,
        const GraphStreamCallback& cb,
        StreamMode stream_mode,
        const RunContext& ctx);
 
    asio::awaitable<NodeResult> execute_node_with_retry_async(
        const std::string& node_name,
        GraphState& state,
        const GraphStreamCallback& cb,
        StreamMode stream_mode,
        const RunContext& ctx);
 
    asio::awaitable<std::vector<NodeResult>> run_parallel_async(
        const std::vector<std::string>& ready,
        int step,
        GraphState& state,
        const std::unordered_map<std::string, NodeResult>& replay,
        CheckpointCoordinator& coord,
        const std::string& parent_cp_id,
        const BarrierState& barrier_state,
        std::vector<std::string>& trace,
        const GraphStreamCallback& cb,
        StreamMode stream_mode,
        const RunContext& ctx);
 
    asio::awaitable<std::vector<StepRouting>> run_sends_async(
        const std::vector<Send>& sends,
        int step,
        GraphState& state,
        const std::unordered_map<std::string, NodeResult>& replay,
        CheckpointCoordinator& coord,
        const std::string& parent_cp_id,
        std::vector<std::string>& trace,
        const GraphStreamCallback& cb,
        StreamMode stream_mode,
        const RunContext& ctx);
 
private:
    void init_state(GraphState& state) const;
 
    void apply_input(GraphState& state, const json& input) const;
 
    void maybe_warn_serial_fanout(std::size_t width) const;
 
    const std::map<std::string, std::unique_ptr<GraphNode>>& nodes_;
    const std::vector<ChannelDef>& channel_defs_;
    RetryPolicyLookup retry_policy_for_;
    asio::thread_pool* fan_out_pool_ = nullptr;
    NodeCache*         node_cache_   = nullptr;
    mutable std::atomic<bool> warned_serial_fanout_{false};
};
 
} // namespace neograph::graph