diff --git a/CHANGELOG.md b/CHANGELOG.md index 78b6c81..4ede1dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Changed + +- **Flow Policy**: `get_path_bundle` now memoizes raw SPF results keyed by the exact inputs that can vary per call (src, dst, residual content, residual-awareness). EqualBalanced placement and rebalance rounds re-request bundles against residual state that repeats -- 94% of SPF calls in a measured place/rebalance cycle were exact input repeats, largely remove+place round-trips restoring identical bytes -- and those calls are now elided. Matching is exact (FlowGraph state stamp fast path, full residual `memcmp` content path), so all outputs are bit-identical; a corpus hash over max-flow, SPF, KSP and policy outputs is unchanged. Measured on place/rebalance churn: 31-70% faster; single EqualBalanced placement: 6-26% faster; Proportional mode skips the memo (it measured 0% repeat inputs) and is unaffected. `FlowGraph` gained an internal monotonic state stamp to support this; no public API change. + ## [0.8.0] - 2026-08-24 ### Added diff --git a/include/netgraph/core/flow_graph.hpp b/include/netgraph/core/flow_graph.hpp index b548265..f17a518 100644 --- a/include/netgraph/core/flow_graph.hpp +++ b/include/netgraph/core/flow_graph.hpp @@ -21,6 +21,17 @@ class FlowGraph { explicit FlowGraph(const StrictMultiDiGraph& g); ~FlowGraph() noexcept = default; + // Copies and moves take a FRESH uid (and version 0). uid must be unique per + // reachable instance: a shared uid would let two objects that diverge after + // copying present equal StateStamps over different residual content, and + // FlowPolicy's memo fast path would then serve a DAG cached for the other + // object's state. Moved-from objects stay valid and mutable, so moves must + // not share the uid either; the one-time cache miss this costs is nothing. + FlowGraph(const FlowGraph& other); + FlowGraph& operator=(const FlowGraph& other); + FlowGraph(FlowGraph&& other) noexcept; + FlowGraph& operator=(FlowGraph&& other) noexcept; + // Views [[nodiscard]] std::span capacity_view() const noexcept { return fs_.capacity_view(); } [[nodiscard]] std::span residual_view() const noexcept { return fs_.residual_view(); } @@ -46,6 +57,18 @@ class FlowGraph { // Inspect: return a copy of the flow's edges and amounts. [[nodiscard]] std::vector> get_flow_edges(const FlowIndex& idx) const; + // Monotonic stamp identifying the exact residual state of this FlowGraph. + // uid is process-unique per instance (never reused across FlowGraph + // lifetimes); version bumps on every mutation that can change residuals. + // Equal stamps from two observations therefore guarantee identical residual + // content, which FlowPolicy uses to elide repeated identical SPF runs. + struct StateStamp { + std::uint64_t uid; + std::uint64_t version; + bool operator==(const StateStamp&) const noexcept = default; + }; + [[nodiscard]] StateStamp state_stamp() const noexcept { return {uid_, version_}; } + // Reconstruct single path for this flow from ledger. // Returns empty vector if flow uses multipath/proportional splitting. [[nodiscard]] std::vector get_flow_path(const FlowIndex& idx) const; @@ -55,6 +78,10 @@ class FlowGraph { FlowState fs_; // Per-flow ledger: stores only edges with non-zero flow std::unordered_map>, FlowIndexHash> ledger_; + // See state_stamp(). fs_ is private and every mutation path (place, remove, + // remove_by_class, reset) bumps version_, so the stamp cannot go stale. + std::uint64_t uid_ {0}; + std::uint64_t version_ {0}; }; } // namespace netgraph::core diff --git a/include/netgraph/core/flow_policy.hpp b/include/netgraph/core/flow_policy.hpp index 32e1ea2..7079d81 100644 --- a/include/netgraph/core/flow_policy.hpp +++ b/include/netgraph/core/flow_policy.hpp @@ -209,6 +209,36 @@ class FlowPolicy { Cost best_path_cost_ { std::numeric_limits::max() }; FlowId next_flow_id_ { 0 }; + // Memo of recent raw SPF results computed by get_path_bundle(), used only in + // EqualBalanced mode. SPF is a pure function of (graph, selection, multipath, + // node_mask, src, dst, residual, edge_mask); in EB mode everything except + // (src, dst, residual, min_flow) is fixed for the lifetime of the policy (the + // min_flow-derived edge mask is Proportional-only), so those four ARE the key. + // EB placement and rebalance rounds re-request bundles against residual + // content that repeats -- 94% of SPF calls in a measured place/rebalance + // cycle were exact input repeats, largely remove+place round-trips that + // restore identical bytes -- and the memo elides those calls without changing + // any output. Matching is exact: a (uid, version) StateStamp equality is the + // fast path (unchanged FlowGraph implies unchanged content), with a full + // residual memcmp as the content path (catches the round-trips). Proportional + // mode measured 0% repeats, so it skips the memo and pays nothing. Cost gates + // and best_path_cost_ updates still run on every call, hit or miss: they + // depend on mutable policy state. + struct SpfMemoEntry { + NodeId src = -1, dst = -1; + bool with_residual = false; + FlowGraph::StateStamp stamp {0, 0}; + std::vector residual; // key copy; empty when SPF ran residual-blind + bool has_min_flow = false; // toggles require_residual; the value is Proportional-only + PredDAG dag; + Cost dst_cost = 0; + }; + // MRU order: front = most recent. Entry count adapts to graph size so the + // memo stays under ~512 KiB per policy regardless of edge count. + static constexpr std::size_t kSpfMemoMaxEntries = 24; + static constexpr std::size_t kSpfMemoMaxBytes = 512 * 1024; + std::vector spf_memo_; + // Static paths (optional): usable (mask-pruned) bundles, one flow per bundle. struct StaticBundle { PredDAG dag; // pruned to mask-surviving entries diff --git a/src/flow_graph.cpp b/src/flow_graph.cpp index c9ed488..c073e10 100644 --- a/src/flow_graph.cpp +++ b/src/flow_graph.cpp @@ -8,17 +8,63 @@ #include "netgraph/core/constants.hpp" #include +#include +#include namespace netgraph::core { +namespace { +// Process-unique instance ids for state_stamp(); never reused, so a stamp +// taken from one FlowGraph can never match a different (or later) instance. +std::uint64_t fresh_flow_graph_uid() noexcept { + static std::atomic next_uid{1}; + return next_uid.fetch_add(1, std::memory_order_relaxed); +} +} // namespace + FlowGraph::FlowGraph(const StrictMultiDiGraph& g) - : g_(&g), fs_(g) { + : g_(&g), fs_(g), uid_(fresh_flow_graph_uid()) { +} + +// See the header: copies and moves must not share the source's uid, or two +// diverging objects could present equal stamps over different residuals. +FlowGraph::FlowGraph(const FlowGraph& other) + : g_(other.g_), fs_(other.fs_), ledger_(other.ledger_), + uid_(fresh_flow_graph_uid()), version_(0) { +} + +FlowGraph& FlowGraph::operator=(const FlowGraph& other) { + if (this != &other) { + g_ = other.g_; + fs_ = other.fs_; + ledger_ = other.ledger_; + uid_ = fresh_flow_graph_uid(); + version_ = 0; + } + return *this; +} + +FlowGraph::FlowGraph(FlowGraph&& other) noexcept + : g_(other.g_), fs_(std::move(other.fs_)), ledger_(std::move(other.ledger_)), + uid_(fresh_flow_graph_uid()), version_(0) { +} + +FlowGraph& FlowGraph::operator=(FlowGraph&& other) noexcept { + if (this != &other) { + g_ = other.g_; + fs_ = std::move(other.fs_); + ledger_ = std::move(other.ledger_); + uid_ = fresh_flow_graph_uid(); + version_ = 0; + } + return *this; } Flow FlowGraph::place(const FlowIndex& idx, NodeId src, NodeId dst, const PredDAG& dag, Flow amount, FlowPlacement placement) { if (amount <= 0.0) return 0.0; + ++version_; // residuals may change from here on // Get or create ledger entry for this flow. The ledger tracks per-edge // cumulative amounts contributed by this flow (not just last placement). @@ -57,6 +103,7 @@ Flow FlowGraph::place(const FlowIndex& idx, NodeId src, NodeId dst, void FlowGraph::remove(const FlowIndex& idx) { auto it = ledger_.find(idx); if (it == ledger_.end()) return; // flow not found + ++version_; const auto& deltas = it->second; // Revert this flow's allocations from the FlowState by subtracting them. if (!deltas.empty()) { @@ -73,6 +120,7 @@ void FlowGraph::remove_by_class(FlowClass flowClass) { } void FlowGraph::reset() noexcept { + ++version_; fs_.reset(); ledger_.clear(); } diff --git a/src/flow_policy.cpp b/src/flow_policy.cpp index 3e1a773..def6672 100644 --- a/src/flow_policy.cpp +++ b/src/flow_policy.cpp @@ -19,6 +19,7 @@ #include "netgraph/core/profiling.hpp" #include +#include #include #include #include @@ -127,11 +128,71 @@ std::optional> FlowPolicy::get_path_bundle(const FlowGr opts.residual = require_residual ? residual : std::span(); opts.node_mask = node_mask_; // Use user-provided node mask opts.edge_mask = final_edge_mask; - auto res = ctx_.algorithms->spf(ctx_.graph, src, opts); - const auto& dist = res.first; - PredDAG dag = std::move(res.second); - if (dst < 0 || static_cast(dst) >= dist.size()) return std::nullopt; - Cost dst_cost = dist[static_cast(dst)]; + // dist.size() == num_nodes() for every SPF result, so the range check does + // not depend on the SPF output; hoisting it lets the memo skip the call. + if (dst < 0 || dst >= ctx_.graph.graph->num_nodes()) return std::nullopt; + + PredDAG dag; + Cost dst_cost; + const bool use_memo = (flow_placement_ == FlowPlacement::EqualBalanced); + const auto stamp = fg.state_stamp(); + const bool with_residual = !opts.residual.empty(); + std::size_t hit_idx = spf_memo_.size(); + if (use_memo) { + for (std::size_t i = 0; i < spf_memo_.size(); ++i) { + const auto& e = spf_memo_[i]; + // min_flow's VALUE never reaches SPF in EB mode (the value-derived edge + // mask is Proportional-only); only has_value() matters, via + // require_residual. Keying on the value caused misses whenever rebalance + // rounds adjusted the per-flow target against unchanged residuals. + if (e.src != src || e.dst != dst || e.with_residual != with_residual || + e.has_min_flow != min_flow.has_value()) { + continue; + } + // Fast path: same FlowGraph, no mutation since the entry was stored. + // Content path: byte-identical residuals (rebalance remove+place + // round-trips restore content while the version keeps advancing). + const bool same = !with_residual || e.stamp == stamp || + (e.residual.size() == opts.residual.size() && + std::memcmp(e.residual.data(), opts.residual.data(), + opts.residual.size() * sizeof(Cap)) == 0); + if (same) { hit_idx = i; break; } + } + } + if (hit_idx < spf_memo_.size()) { + dag = spf_memo_[hit_idx].dag; + dst_cost = spf_memo_[hit_idx].dst_cost; + // MRU: move the hit to the front so hot entries stay cheap to find. + if (hit_idx != 0) { + std::rotate(spf_memo_.begin(), spf_memo_.begin() + hit_idx, + spf_memo_.begin() + hit_idx + 1); + } + } else { + auto res = ctx_.algorithms->spf(ctx_.graph, src, opts); + dag = std::move(res.second); + dst_cost = res.first[static_cast(dst)]; + if (use_memo) { + SpfMemoEntry entry; + entry.src = src; + entry.dst = dst; + entry.with_residual = with_residual; + entry.stamp = stamp; + entry.residual.assign(opts.residual.begin(), opts.residual.end()); + entry.has_min_flow = min_flow.has_value(); + entry.dag = dag; + entry.dst_cost = dst_cost; + const std::size_t entry_bytes = + entry.residual.size() * sizeof(Cap) + + entry.dag.parent_offsets.size() * sizeof(std::int32_t) + + entry.dag.parents.size() * (sizeof(NodeId) + sizeof(EdgeId)) + + sizeof(SpfMemoEntry); + const std::size_t cap = std::clamp( + kSpfMemoMaxBytes / std::max(entry_bytes, 1), 1, + kSpfMemoMaxEntries); + spf_memo_.insert(spf_memo_.begin(), std::move(entry)); + if (spf_memo_.size() > cap) spf_memo_.resize(cap); + } + } if (dst_cost < best_path_cost_) best_path_cost_ = dst_cost; // Enforce path cost constraints: diff --git a/tests/cpp/flow_graph_tests.cpp b/tests/cpp/flow_graph_tests.cpp index 100613c..687492d 100644 --- a/tests/cpp/flow_graph_tests.cpp +++ b/tests/cpp/flow_graph_tests.cpp @@ -264,3 +264,49 @@ TEST(FlowGraph, LedgerMicroFlows_RemovalRestoresResidual) { EXPECT_NEAR(res_after[i], capv[i], 1e-9) << "Residual not restored at edge " << i; } } + +// state_stamp() uids must be unique per reachable instance. A copy that shared +// the source's uid could diverge by the same number of mutations and then +// present an equal stamp over different residual content, which would let +// FlowPolicy's SPF memo fast path serve a DAG cached for the other object. +// Moved-from FlowGraphs remain valid and mutable, so moves must not share the +// uid either. +TEST(FlowGraphStamp, CopiesAndMovesGetFreshUids) { + auto g = make_square_graph(1); + FlowGraph a(g); + const auto uid_a = a.state_stamp().uid; + + FlowGraph b(a); // copy ctor + EXPECT_NE(b.state_stamp().uid, uid_a); + + FlowGraph c(g); + c = a; // copy assign + EXPECT_NE(c.state_stamp().uid, uid_a); + EXPECT_NE(c.state_stamp().uid, b.state_stamp().uid); + + FlowGraph d(std::move(b)); // move ctor + EXPECT_NE(d.state_stamp().uid, uid_a); + + FlowGraph e(g); + e = std::move(c); // move assign + EXPECT_NE(e.state_stamp().uid, uid_a); +} + +// The end-to-end failure the uid rule prevents: two copies mutated the same +// NUMBER of times (equal versions) but with different content must never +// satisfy the stamp fast path. Equal-version divergence is exactly the case a +// shared uid would get wrong. +TEST(FlowGraphStamp, EqualVersionDivergentCopiesCompareUnequal) { + auto g = make_square_graph(1); // 0->1->2 cheap, 0->3->2 expensive + FlowGraph a(g); + FlowGraph b(a); + + // One mutation each (same version count), different targets. + PredDAG cheap = make_path_dag(g, std::array{0, 1}); + PredDAG dear = make_path_dag(g, std::array{2, 3}); + FlowIndex idx{0, 2, 0, 0}; + (void)a.place(idx, 0, 2, cheap, 0.5, FlowPlacement::Proportional); + (void)b.place(idx, 0, 2, dear, 0.5, FlowPlacement::Proportional); + + EXPECT_FALSE(a.state_stamp() == b.state_stamp()); +} diff --git a/tests/py/test_flow_policy_spf_memo.py b/tests/py/test_flow_policy_spf_memo.py new file mode 100644 index 0000000..172e95c --- /dev/null +++ b/tests/py/test_flow_policy_spf_memo.py @@ -0,0 +1,158 @@ +"""FlowPolicy's SPF memo must be invisible: identical results, fewer SPF runs. + +EqualBalanced placement and rebalance rounds re-request path bundles against +residual state that repeats (94% of SPF calls in a measured place/rebalance +cycle were exact input repeats), so get_path_bundle memoizes raw SPF results, +keyed by (src, dst, residual content, residual-awareness). These tests pin the +one way that can go wrong: a stale hit -- serving a cached DAG computed for +residuals that have since changed. Each scenario mutates residual state between +placements and asserts routing reflects the *current* residuals. +""" + +from __future__ import annotations + +import numpy as np +import pytest + +import netgraph_core as ngc + + +def _graph(num_nodes, edges): + """edges: (src, dst, cost, cap).""" + s = np.array([e[0] for e in edges], np.int32) + d = np.array([e[1] for e in edges], np.int32) + k = np.array([int(e[2]) for e in edges], np.int64) + c = np.array([float(e[3]) for e in edges], np.float64) + return ngc.StrictMultiDiGraph.from_arrays( + num_nodes, s, d, c, k, np.arange(len(edges), dtype=np.int64) + ) + + +def _eb_policy(algs, gh, **kw): + sel = ngc.EdgeSelection( + multi_edge=True, + require_capacity=True, + tie_break=ngc.EdgeTieBreak.DETERMINISTIC, + ) + cfg = ngc.FlowPolicyConfig( + path_alg=ngc.PathAlg.SPF, + flow_placement=ngc.FlowPlacement.EQUAL_BALANCED, + selection=sel, + **kw, + ) + return ngc.FlowPolicy(algs, gh, cfg) + + +def _two_path_graph(): + """0->1->3 (cost 2) and 0->2->3 (cost 4); cheap path capacity 2.""" + return _graph( + 4, + [ + (0, 1, 1, 2.0), + (1, 3, 1, 2.0), + (0, 2, 2, 10.0), + (2, 3, 2, 10.0), + ], + ) + + +def _edges_used(fg, g, idx): + flows = fg.get_flow_edges(ngc.FlowIndex(*idx)) + src = np.asarray(g.edge_src_view()) + dst = np.asarray(g.edge_dst_view()) + return {(int(src[e]), int(dst[e])) for e, amt in flows if amt > 1e-12} + + +class TestMemoInvalidation: + def test_replacement_sees_residuals_changed_by_another_policy(self, algs): + """A cached bundle must not survive residual changes made between calls. + + Policy A places on the cheap path and is removed (restoring residuals, + which reproduces the memo's key). Policy B then saturates the cheap + path through the same FlowGraph. When A places again, its SPF runs + against the *new* residuals and must route via the expensive path; a + stale memo hit would re-serve the cheap-path DAG. + """ + g = _two_path_graph() + gh = algs.build_graph(g) + fg = ngc.FlowGraph(g) + + pa = _eb_policy(algs, gh, max_flow_count=1, min_flow_count=1) + placed, _ = pa.place_demand(fg, 0, 3, 0, 1.0) + assert placed == pytest.approx(1.0) + (idx,) = pa.flows.keys() + assert _edges_used(fg, g, idx) == {(0, 1), (1, 3)} + pa.remove_demand(fg) # residual content is now back to the initial bytes + + pb = _eb_policy(algs, gh, max_flow_count=1, min_flow_count=1) + placed_b, _ = pb.place_demand(fg, 0, 3, 1, 2.0) + assert placed_b == pytest.approx(2.0) # saturates 0->1->3 + + placed2, _ = pa.place_demand(fg, 0, 3, 0, 1.0) + assert placed2 == pytest.approx(1.0) + (idx2,) = pa.flows.keys() + assert _edges_used(fg, g, idx2) == {(0, 2), (2, 3)}, ( + "placement after residual change reused a stale cached path" + ) + + def test_alternating_flow_graphs_stay_isolated(self, algs): + """One policy asked about two FlowGraphs must answer for the right one. + + fg_busy has the cheap path saturated; fg_idle does not. Alternating + get-path queries (via place/remove cycles) between the two must route + differently every time, regardless of which answer the memo last held. + """ + g = _two_path_graph() + gh = algs.build_graph(g) + fg_idle = ngc.FlowGraph(g) + fg_busy = ngc.FlowGraph(g) + + blocker = _eb_policy(algs, gh, max_flow_count=1, min_flow_count=1) + placed, _ = blocker.place_demand(fg_busy, 0, 3, 9, 2.0) + assert placed == pytest.approx(2.0) + + prober = _eb_policy(algs, gh, max_flow_count=1, min_flow_count=1) + for _ in range(3): + placed, _ = prober.place_demand(fg_idle, 0, 3, 0, 1.0) + assert placed == pytest.approx(1.0) + (idx,) = prober.flows.keys() + assert _edges_used(fg_idle, g, idx) == {(0, 1), (1, 3)} + prober.remove_demand(fg_idle) + + placed, _ = prober.place_demand(fg_busy, 0, 3, 0, 1.0) + assert placed == pytest.approx(1.0) + (idx,) = prober.flows.keys() + assert _edges_used(fg_busy, g, idx) == {(0, 2), (2, 3)} + prober.remove_demand(fg_busy) + + def test_eb_rebalance_churn_matches_fresh_policy(self, algs): + """Heavy place/rebalance churn (the memoized path) must land exactly + where a fresh policy with no memo history lands.""" + rng = np.random.default_rng(3) + n = 24 + edges = [] + for _ in range(70): + u, v = rng.integers(0, n, size=2) + if u != v: + edges.append( + ( + int(u), + int(v), + int(rng.integers(1, 21)), + float(rng.integers(1, 6)), + ) + ) + g = _graph(n, edges) + gh = algs.build_graph(g) + + def run(cycles): + p = _eb_policy(algs, gh, max_flow_count=4, min_flow_count=4) + fg = ngc.FlowGraph(g) + for _ in range(cycles): + p.place_demand(fg, 0, n - 1, 0, 8.0) + p.rebalance_demand(fg, 0, n - 1, 0, 2.0) + return np.asarray(fg.edge_flow_view()).copy() + + churned = run(cycles=4) # memo-heavy history + fresh = run(cycles=4) # brand-new policy, identical inputs + np.testing.assert_array_equal(churned, fresh)