cancel.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
 
#include <asio/any_io_executor.hpp>
#include <asio/cancellation_signal.hpp>
#include <asio/post.hpp>
 
#include <algorithm>
#include <atomic>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <utility>
#include <vector>
 
namespace neograph::graph {
 
class NEOGRAPH_API CancelledException : public std::runtime_error {
public:
    CancelledException()
        : std::runtime_error("neograph: run cancelled") {}
    explicit CancelledException(const std::string& detail)
        : std::runtime_error("neograph: run cancelled — " + detail) {}
};
 
class NEOGRAPH_API CancelToken {
public:
    CancelToken() = default;
 
    CancelToken(const CancelToken&) = delete;
    CancelToken& operator=(const CancelToken&) = delete;
 
    void cancel() noexcept {
        if (cancelled_.exchange(true, std::memory_order_acq_rel)) {
            return;  // already cancelled
        }
 
        // Snapshot the executor under the mutex; bind_executor and
        // cancel may race across threads.
        asio::any_io_executor ex_snapshot;
        {
            std::lock_guard<std::mutex> lk(mu_);
            ex_snapshot = ex_;
        }
        if (ex_snapshot) {
            // Post the emit onto the strand that owns the signal —
            // emit() is documented as not thread-safe across executors.
            // The engine binds the same executor it co_spawns on.
            asio::post(ex_snapshot, [this]() {
                sig_.emit(asio::cancellation_type::all);
            });
        }
 
        // v1.0 (9c): the legacy ``add_cancel_hook`` / hooks_ iteration
        // is gone — ``fork()`` is the canonical primitive for nested
        // cancel scopes, and every internal ``run_sync`` cascades via
        // a forked child.
 
        // Cascade to live children produced by ``fork()``.
        // Snapshot live shared_ptrs under the lock, then call
        // ``cancel()`` on each outside the lock — the recursive call
        // would otherwise re-enter ``children_mu_`` if a grandchild
        // exists, and lock_guard isn't reentrant.
        std::vector<std::shared_ptr<CancelToken>> live_children;
        {
            std::lock_guard<std::mutex> lk(children_mu_);
            live_children.reserve(children_.size());
            for (auto& w : children_) {
                if (auto sp = w.lock()) {
                    live_children.push_back(std::move(sp));
                }
            }
            // Don't bother pruning expired entries here — fork() does
            // it on the next call, and after this cancel() the parent
            // is "done" anyway.
        }
        for (auto& child : live_children) {
            child->cancel();
        }
    }
 
    [[nodiscard]] bool is_cancelled() const noexcept {
        return cancelled_.load(std::memory_order_acquire);
    }
 
    void bind_executor(asio::any_io_executor ex) {
        bool fire_immediately = false;
        {
            std::lock_guard<std::mutex> lk(mu_);
            ex_ = std::move(ex);
            fire_immediately = cancelled_.load(std::memory_order_acquire);
        }
        if (fire_immediately) {
            std::lock_guard<std::mutex> lk(mu_);
            if (ex_) {
                asio::post(ex_, [this]() {
                    sig_.emit(asio::cancellation_type::all);
                });
            }
        }
    }
 
    asio::cancellation_slot slot() noexcept {
        return sig_.slot();
    }
 
    void throw_if_cancelled(const std::string& detail = {}) const {
        if (is_cancelled()) {
            throw CancelledException(detail);
        }
    }
 
    [[nodiscard]] std::shared_ptr<CancelToken> fork() {
        // Construct child outside any lock — its ctor takes no lock.
        // shared_ptr ctor allocates the control block; cheap relative
        // to a real cancel scope (one io_context spin-up downstream).
        auto child = std::shared_ptr<CancelToken>(new CancelToken());
 
        {
            std::lock_guard<std::mutex> lk(children_mu_);
            // Opportunistic prune: every fork() is a natural place to
            // drop expired weak_ptrs from previous children that have
            // already been released by their owners. Bounds the
            // children_ vector at the live-fan-out width even on
            // long-lived parents (e.g. an engine with 1000 LLM calls
            // per run, each forking once).
            children_.erase(
                std::remove_if(
                    children_.begin(), children_.end(),
                    [](const std::weak_ptr<CancelToken>& w) {
                        return w.expired();
                    }),
                children_.end());
            children_.push_back(child);
        }
 
        // Eager propagation: parent already cancelled → child sees
        // the polling flag immediately. ``bind_executor`` on the
        // child will then fire its signal at the next co_await.
        if (cancelled_.load(std::memory_order_acquire)) {
            child->cancel();
        }
        return child;
    }
 
private:
    std::atomic<bool>        cancelled_{false};
    mutable std::mutex       mu_;        // guards ex_ vs cancel() race
    asio::any_io_executor    ex_;        // bound by engine before HTTP I/O
    asio::cancellation_signal sig_;      // for asio operation cancel
 
    // Hierarchical cascade list. Each entry is a child token
    // produced by ``fork()``. ``cancel()`` walks live children and
    // cascades. Stored as ``weak_ptr`` so a child that goes out of
    // scope (its run_sync returned, run completed) is automatically
    // pruned on the next ``cancel()`` / ``fork()`` traversal.
    mutable std::mutex children_mu_;
    std::vector<std::weak_ptr<CancelToken>> children_;
};
 
// v1.0 (9d): the `current_cancel_token()` thread_local +
// `CurrentCancelTokenScope` RAII smuggling channel is gone.
// `RunContext::cancel_token` (on `NodeInput::ctx`) is the only cancel
// channel. Engine threads it through every dispatch.
 
}  // namespace neograph::graph