graph.hpp

Go to the documentation of this file.

#pragma once
 
// ============================================================================
//  reactive/graph.hpp
// ----------------------------------------------------------------------------
//  `Graph` is the process-wide coordinator of the reactive subsystem
//  (one instance per process, reached via `Node::graph()`).
//
//  The Graph provides four things:
//    1. Transactions (batches). Multiple Property writes coalesce into a
//       single flush, notifying the UI exactly once -- eliminating the
//       "3 setters -> 3 repaints" problem entirely.
//    2. Push phase (coloring). A source change propagates MaybeDirty down
//       the DAG without triggering any computation.
//    3. Pull phase (evaluation). At flush time the affected nodes are
//       evaluated in topological order (ascending depth); nodes whose
//       upstreams did not actually move are skipped, which is what makes
//       the graph glitch-free.
//    4. Tracking context. While a Derivation computes, every reactive read
//       it performs is recorded automatically: `Property::get()` and
//       `Computed::get()` call into the active TrackingContext themselves,
//       so a Computed never declares its dependency list. The Graph turns
//       the recorded reads into Edges and reconciles the edge set after each
//       compute, which is what lets a dependency set change shape between
//       evaluations.
//
//       `dep(x)` exists as an explicit escape hatch for the case where a
//       value must be depended upon without its `get()` appearing in the
//       compute body; `reactive::untracked` is the inverse, suppressing
//       recording for reads inside its scope.
//
//  Threading model
//  ---------------
//  Single-threaded (UI thread). Every public Graph entry point asserts the
//  calling thread in Debug builds. Cross-thread updates must be posted via
//  the Dispatcher to the graph thread before invoking any API here.
//
//  Cycles
//  ------
//  If flush encounters a dependency cycle B -> ... -> B, it raises
//  `CircularDependencyError` instead of spinning or overflowing the stack.
// ============================================================================
 
#include "aria/reactive/node.hpp"
#include "aria/callback_boundary.hpp"
 
#include <algorithm>
#include <cassert>
#include <exception>
#include <functional>
#include <stdexcept>
#include <string>
#include <string_view>
#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>
 
namespace aria::reactive {

class CircularDependencyError : public std::runtime_error {
public:
    using std::runtime_error::runtime_error;
};
 
// ---------------------------------------------------------------------------
//  Flush tracing hook (used by `aria::GraphInspector`).
//  We keep the type minimal here so `graph.hpp` does not have to know
//  anything about the inspector; `inspector.hpp` packages these raw
//  parameters into a nicer `FlushEvent` struct for user callbacks.
//
//  Phase codes are stable integers so the ABI of the hook never breaks
//  across versions:
//    0 = FlushBegin   (node = nullptr)
//    1 = RoundBegin   (node = nullptr,  round = 1..N)
//    2 = Pull         (node = n,         round = 1..N)
//    3 = SkipClean    (node = n,         round = 1..N)
//    4 = Recomputed   (node = n,         round, changed)
//    5 = RoundEnd     (node = nullptr,  round)
//    6 = FlushEnd     (node = nullptr)
// ---------------------------------------------------------------------------
using FlushTraceFn = std::function<void(int phase,
                                        const Node* node,
                                        int round,
                                        bool changed)>;
 
// Registration and use are confined to the graph thread. Shared ownership
// preserves a running tracer and its state across replacement/reentrant clear.
ARIA_ABI_API std::shared_ptr<FlushTraceFn>& flush_trace_hook_() noexcept;

struct ReadRecord {
    Node* source;  
};

class TrackingContext {
public:
    using Buffer = std::vector<detail::NodeHandle>;
 
    TrackingContext() = default;
    explicit TrackingContext(Buffer& reusable) noexcept : reusable_(&reusable) {
        reads_.swap(reusable);
    }
    ~TrackingContext() {
        if (reusable_) {
            reads_.clear();
            reads_.swap(*reusable_);
        }
    }
    TrackingContext(const TrackingContext&) = delete;
    TrackingContext& operator=(const TrackingContext&) = delete;

    void record_read(Node& src) {
        // Small read sets -- linear de-dup is more than fast enough.
        for (const auto& n : reads_) {
            if (n.get() == &src) return;
        }
        reads_.emplace_back(&src);
    }
 
    [[nodiscard]] const std::vector<detail::NodeHandle>& reads() const noexcept { return reads_; }
    void                                     clear() noexcept      { reads_.clear(); }
 
private:
    Buffer reads_;
    Buffer* reusable_ = nullptr;
};

class Graph {
    // `Node::mark_downstream_maybe_dirty()` reaches into `color_stack_`
    // and `color_in_use_` to amortise the coloring buffer across
    // notifications. Symmetric to `Node`'s existing `friend class
    // Graph;` declaration — together they keep the intrusive coupling
    // between the two header-only types contained.
    friend class Node;
 
public:
    Graph() noexcept;
    ~Graph() noexcept = default;
 
    Graph(const Graph&)            = delete;
    Graph& operator=(const Graph&) = delete;
 
    // ---- Debug thread assertion (no-op in Release) -----------------------
    void assert_on_graph_thread() const noexcept {
#ifndef NDEBUG
        // First call records the thread; later calls must match.
        if (owner_thread_ == std::thread::id{}) {
            const_cast<Graph*>(this)->owner_thread_ = std::this_thread::get_id();
            return;
        }
        assert(owner_thread_ == std::this_thread::get_id()
               && "reactive::Graph accessed from a non-UI thread. "
                  "Use Dispatcher::post to marshal updates back to the graph thread.");
#endif
    }
 
    // ------------------------------------------------------------------
    //  Transactions (batches)
    // ------------------------------------------------------------------

    void begin_batch() noexcept {
        assert_on_graph_thread();
        ++batch_depth_;
    }

    void end_batch() {
        assert_on_graph_thread();
        assert(batch_depth_ > 0 && "end_batch without matching begin_batch");
        if (--batch_depth_ == 0 && !flushing_) {
            flush();
        }
    }
 
    [[nodiscard]] bool in_batch() const noexcept { return batch_depth_ > 0; }
 
    // ------------------------------------------------------------------
    //  Push phase -- invoked when a Source's value changes.
    // ------------------------------------------------------------------

    void on_source_changed(Node& src);

    void enqueue_dirty(Node& n) {
        if (n.queued_) return;
        pending_.emplace_back(&n);
        n.queued_ = true;
    }
 
    // ------------------------------------------------------------------
    //  Pull phase -- flush and evaluation
    // ------------------------------------------------------------------

    void flush();

    bool pull(Node& n);
 
    // ------------------------------------------------------------------
    //  Tracking
    // ------------------------------------------------------------------
 
    [[nodiscard]] TrackingContext* current_tracker() noexcept {
        return tracker_stack_.empty() ? nullptr : tracker_stack_.back();
    }

    void push_tracker(TrackingContext* ctx) { tracker_stack_.push_back(ctx); }
    void pop_tracker(TrackingContext* ctx) {
        assert(!tracker_stack_.empty() && tracker_stack_.back() == ctx);
        (void)ctx;
        tracker_stack_.pop_back();
    }

    void enter_untracked() { tracker_stack_.push_back(nullptr); }
    void leave_untracked() {
        assert(!tracker_stack_.empty() && tracker_stack_.back() == nullptr);
        tracker_stack_.pop_back();
    }
 
private:
    // Inner helper used by `pull`: assumes every MaybeDirty upstream of
    // `n` has already been settled, and runs the version-compare /
    // recompute leg without recursing into upstream resolution. The
    // public `pull` first walks the graph iteratively to reach this
    // precondition, then calls `pull_settle_` for each node bottom-up.
    bool pull_settle_(Node& n);
 
    // Reusable scratch buffer for `Node::mark_downstream_maybe_dirty()`.
    // The coloring walk used to allocate a fresh `std::vector<Node*>`
    // every time a Source committed a value — which is millions of
    // allocations per second on hot UI workloads. Hoisting it into the
    // (single-threaded) graph lets us amortise the buffer across every
    // notify, capacity grows monotonically to the deepest fan-out
    // observed and is reused thereafter. Exposed only to `Node` via the
    // existing `friend class Graph` declaration.
    std::vector<Node*> color_stack_;
    bool color_in_use_ = false;   // re-entrancy guard
 
    // Guard against nested flush calls. Recursive set() re-enqueues into
    // the next round instead of starting a new flush.
    bool flushing_ = false;
 
    int batch_depth_ = 0;
 
    // A node has at most one entry in pending_. Snapshotting the next
    // round resets queued_ before user code runs, permitting a new pulse.
    std::vector<detail::NodeHandle> pending_;
    std::vector<detail::NodeHandle> round_;
 
    // Active recursive `pull()` stack. When a cycle trips the
    // `Computing` re-entry check we use this to format a full
    // A → B → C → A path in the error message — vastly more useful
    // than just "node X is re-entering itself".
    std::vector<detail::NodeHandle> pulling_stack_;
 
    // Tracker stack. `nullptr` entries denote untracked scopes.
    std::vector<TrackingContext*> tracker_stack_;
 
    std::thread::id owner_thread_{};
 
    // Safety fuse against runaway propagation (which should only happen
    // when there is a bug in a user-supplied recompute function).
    static constexpr int kMaxFlushRounds = 100;
};
 
// ------------------------------------------------------------------
//  Convenience API -- user-facing entry points
// ------------------------------------------------------------------

[[nodiscard]] inline TrackingContext* current_tracker() noexcept {
    return Node::graph().current_tracker();
}

class TrackerScope {
public:
    explicit TrackerScope(TrackingContext& ctx) : ctx_(&ctx) {
        Node::graph().push_tracker(&ctx);
    }
    ~TrackerScope() { Node::graph().pop_tracker(ctx_); }
    TrackerScope(const TrackerScope&)            = delete;
    TrackerScope& operator=(const TrackerScope&) = delete;
private:
    TrackingContext* ctx_;
};

class UntrackedScope {
public:
    UntrackedScope() { Node::graph().enter_untracked(); }
    ~UntrackedScope() { Node::graph().leave_untracked(); }
    UntrackedScope(const UntrackedScope&)            = delete;
    UntrackedScope& operator=(const UntrackedScope&) = delete;
};

template<class Fn>
auto untracked(Fn&& fn) -> decltype(fn()) {
    UntrackedScope guard;
    if constexpr (std::is_void_v<decltype(fn())>) {
        std::forward<Fn>(fn)();
    } else {
        return std::forward<Fn>(fn)();
    }
}

class BatchScope {
public:
    BatchScope() { Node::graph().begin_batch(); }
    ~BatchScope() noexcept {
        try {
            Node::graph().end_batch();
        } catch (...) {
            // Avoid std::terminate during unwinding, but do not make
            // the failure invisible: route it through the lightweight
            // callback-failure sink (stderr fallback until the host
            // installs its own sink). Keep this dependency in core so
            // the header remains portable across GCC/Clang/MSVC.
            ::aria::report_callback_failure(
                std::string_view{"reactive.batch_scope.end_batch"},
                std::current_exception());
        }
    }
    BatchScope(const BatchScope&)            = delete;
    BatchScope& operator=(const BatchScope&) = delete;
};

template<class Fn>
auto batch(Fn&& fn) -> decltype(fn()) {
    BatchScope guard;
    if constexpr (std::is_void_v<decltype(fn())>) {
        std::forward<Fn>(fn)();
    } else {
        return std::forward<Fn>(fn)();
    }
}
 
}  // namespace aria::reactive