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 <chrono>
#include <exception>
#include <functional>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <unordered_map>
#include <vector>
 
namespace neograph::graph {
 
class GraphState;
class GraphRegistry;
 
struct RunContext;
 
namespace detail {
struct InvocationSlot;
struct FailureRouting;
class AttemptQuiescence;
struct SendStep;
class AttemptEffectLog;
struct AttemptsAccess;
}  // namespace detail
 
class NEOGRAPH_API NodeExecutor {
public:
    using RetryPolicyLookup = std::function<RetryPolicy(const std::string&)>;
    using AttemptTimeoutLookup = std::function<std::optional<int>(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);
 
    NodeExecutor(const std::map<std::string, std::unique_ptr<GraphNode>>& nodes,
                 const std::vector<ChannelDef>&                           channel_defs,
                 RetryPolicyLookup                                        retry_policy_for,
                 std::shared_ptr<const GraphRegistry>                     registry,
                 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:
    friend class GraphEngine;
    friend struct detail::AttemptsAccess;
 
    asio::awaitable<std::vector<StepRouting>> run_sends_impl_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,
        std::vector<json>* failure_records,
        const detail::SendStep* send_step = nullptr);
 
    asio::awaitable<NodeResult> execute_attempts_async(
        const std::string& node_name,
        GraphState& state,
        const GraphStreamCallback& cb,
        StreamMode stream_mode,
        const RunContext& ctx,
        detail::InvocationSlot* slot);
 
    asio::awaitable<void> run_dispatched_attempt_async(
        GraphNode* node,
        GraphState& state,
        const GraphStreamCallback& cb,
        const RunContext& ctx,
        detail::InvocationSlot* slot,
        std::string_view effect_namespace,
        const std::optional<int>& attempt_timeout,
        const std::string& node_name,
        int attempt,
        std::optional<NodeResult>& ok_result,
        std::exception_ptr& failure,
        std::shared_ptr<detail::AttemptEffectLog>& effects,
        bool& attempt_stopped,
        bool& attempt_timed_out);
 
    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_;
    std::shared_ptr<const GraphRegistry>                     registry_;
    asio::thread_pool* fan_out_pool_ = nullptr;
    NodeCache*         node_cache_   = nullptr;
    mutable std::atomic<bool> warned_serial_fanout_{false};
    const detail::FailureRouting* failure_routing_ = nullptr;
    AttemptTimeoutLookup attempt_timeout_for_;
    std::shared_ptr<detail::AttemptQuiescence> attempt_quiescence_;
};
 
NEOGRAPH_API bool wait_for_node_attempts(
    std::chrono::milliseconds timeout = std::chrono::milliseconds::max());
 
} // namespace neograph::graph