From 480e7977fc8390e8c03dd6a23bbc3ce9323569ac Mon Sep 17 00:00:00 2001 From: Andrey Golovanov Date: Sun, 13 Sep 2026 00:01:55 +0100 Subject: [PATCH] Add fixed and lossy equal-balanced placement and reverse SPF Two equal-balanced placement modes for hop-by-hop ECMP with a load-blind forwarding table: EQUAL_BALANCED_FIXED admits losslessly over the topology's next-hop set (a member filled since the DAG was computed drives the scale to 0), EQUAL_BALANCED_LOSSY forwards best-effort and reports the dropped volume per edge via FlowGraph.place_with_drops. FlowPolicy treats both as equal-balanced but skips the equalizing rebalance for the lossy mode, so pinned routes carry what fits. shortest_paths_to / Algorithms.spf_to runs Dijkstra over the in-adjacency and returns distances to one destination plus a forward-oriented PredDAG valid for placement from any node, with optional forced fan-out edges for an origin whose first hop is a traffic split rather than a routing choice. A cost-only FlowPolicy (require_capacity=false) no longer passes the residual to SPF when an equal-balanced per-flow target is set; a residual forces capacity-aware selection, so such policies routed around saturated edges instead of following costs. Version 0.9.0. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 9 +- bindings/python/module.cpp | 62 +++++- include/netgraph/core/algorithms.hpp | 5 + include/netgraph/core/backend.hpp | 6 + include/netgraph/core/flow_graph.hpp | 9 +- include/netgraph/core/flow_policy.hpp | 9 +- include/netgraph/core/flow_state.hpp | 15 +- include/netgraph/core/options.hpp | 11 ++ include/netgraph/core/shortest_paths.hpp | 36 ++++ include/netgraph/core/types.hpp | 23 ++- pyproject.toml | 2 +- python/netgraph_core/_docs.py | 89 ++++++++- src/cpu_backend.cpp | 14 ++ src/flow_graph.cpp | 5 +- src/flow_policy.cpp | 36 +++- src/flow_state.cpp | 126 ++++++++++-- src/shortest_paths.cpp | 231 ++++++++++++++++++++++ tests/cpp/flow_state_tests.cpp | 122 ++++++++++++ tests/cpp/shortest_paths_tests.cpp | 110 +++++++++++ tests/py/test_equal_balanced_modes.py | 237 +++++++++++++++++++++++ tests/py/test_spf_to.py | 114 +++++++++++ 21 files changed, 1234 insertions(+), 37 deletions(-) create mode 100644 tests/py/test_equal_balanced_modes.py create mode 100644 tests/py/test_spf_to.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ede1dd..ed073b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,17 @@ 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] +## [0.9.0] - 2026-09-13 + +### Added + +- **Shortest Paths**: `shortest_paths_to(dst, ...)` / `Algorithms.spf_to` runs Dijkstra over the in-adjacency and returns distances *to* one destination plus a forward-oriented `PredDAG` valid for placement from any node toward it (the union of the per-source shortest-path DAGs). `fanout_edges` adds edges regardless of cost, for an origin whose first hop is a traffic split rather than a routing decision, such as a pseudo source whose demand originates evenly at every attached real source; a fan-out edge must leave a node with no incoming DAG entry, which keeps the result acyclic. +- **Flow Placement**: two equal-balanced modes for hop-by-hop ECMP with a load-blind forwarding table. `EQUAL_BALANCED_FIXED` takes the split set from the DAG edges with *capacity* rather than residual and keeps the single global admission scale, so a member filled since the DAG was computed drives the scale to 0 (lossless hash-ECMP admission; the existing `EQUAL_BALANCED` drops such members from the split, which is progressive/TE behaviour). `EQUAL_BALANCED_LOSSY` splits over the same set without a scale: each edge carries `min(share, residual)`, the excess is dropped, deficits propagate downstream, and the placed amount is what reaches the destination. `FlowGraph.place_with_drops` returns the dropped volume per edge; `FlowState::place_on_dag` gained an optional `drops` collector. On a 100/10 parallel pair offered 100 units, `EQUAL_BALANCED` and `EQUAL_BALANCED_FIXED` admit 20 and `EQUAL_BALANCED_LOSSY` delivers 60. `FlowPolicy` treats the new modes as equal-balanced except that the lossy mode skips the equalizing rebalance, so with pinned routes each LSP carries what fits and `placed` is the delivered total. ### Changed +- **Flow Policy**: a cost-only policy (`require_capacity=false`) no longer passes the residual to SPF when an equal-balanced per-flow target is set. A residual forces capacity-aware edge selection, so such policies silently routed around saturated edges instead of following costs; they now route on cost alone, and the per-flow target still bounds how much each flow requests. + - **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 diff --git a/bindings/python/module.cpp b/bindings/python/module.cpp index 1cde885..a55a71d 100644 --- a/bindings/python/module.cpp +++ b/bindings/python/module.cpp @@ -110,7 +110,9 @@ PYBIND11_MODULE(_netgraph_core, m, py::mod_gil_not_used()) { py::enum_(m, "FlowPlacement") .value("PROPORTIONAL", FlowPlacement::Proportional) - .value("EQUAL_BALANCED", FlowPlacement::EqualBalanced); + .value("EQUAL_BALANCED", FlowPlacement::EqualBalanced) + .value("EQUAL_BALANCED_FIXED", FlowPlacement::EqualBalancedFixed) + .value("EQUAL_BALANCED_LOSSY", FlowPlacement::EqualBalancedLossy); py::class_(m, "StrictMultiDiGraph") .def_static( @@ -218,6 +220,56 @@ PYBIND11_MODULE(_netgraph_core, m, py::mod_gil_not_used()) { throw py::value_error("dtype must be 'float64' or 'int64'"); } }, py::arg("graph"), py::arg("src"), py::arg("dst") = py::none(), py::kw_only(), py::arg("selection") = py::none(), py::arg("residual") = py::none(), py::arg("node_mask") = py::none(), py::arg("edge_mask") = py::none(), py::arg("multipath") = true, py::arg("dtype") = "float64") + .def("spf_to", [](const Algorithms& algs, const PyGraph& pg, std::int32_t dst, + py::object selection_obj, py::object residual_obj, + py::object node_mask, py::object edge_mask, bool multipath, + py::object fanout_obj, std::string dtype) -> py::tuple { + if (dst < 0 || dst >= pg.num_nodes) throw py::value_error("dst out of range"); + SpfToOptions opts; if (!selection_obj.is_none()) opts.selection = py::cast(selection_obj); + std::vector residual_vec; + if (!residual_obj.is_none()) { + auto arr = py::cast(residual_obj); + if (!(arr.flags() & py::array::c_style)) throw py::type_error("residual must be C-contiguous (np.ascontiguousarray)"); + auto buf = arr.request(); + if (buf.ndim != 1 || buf.format != py::format_descriptor::format()) throw py::type_error("residual must be 1-D float64"); + if (static_cast(buf.shape[0]) != pg.num_edges) { + throw py::type_error("residual length must equal " + std::to_string(pg.num_edges)); + } + residual_vec.resize(static_cast(buf.shape[0])); + std::memcpy(residual_vec.data(), buf.ptr, residual_vec.size()*sizeof(double)); + opts.residual = std::span(residual_vec.data(), residual_vec.size()); + } + std::vector fanout_vec; + if (!fanout_obj.is_none()) { + for (auto item : py::iterable(fanout_obj)) { + auto e = py::cast(item); + if (e < 0 || e >= pg.num_edges) throw py::value_error("fanout edge id out of range"); + fanout_vec.push_back(static_cast(e)); + } + opts.fanout_edges = std::span(fanout_vec.data(), fanout_vec.size()); + } + auto node_bs = to_bool_span_from_numpy(node_mask, static_cast(pg.num_nodes), "node_mask"); + auto edge_bs = to_bool_span_from_numpy(edge_mask, static_cast(pg.num_edges), "edge_mask"); + opts.node_mask = node_bs.view; + opts.edge_mask = edge_bs.view; + opts.multipath = multipath; + py::gil_scoped_release rel; auto res = algs.spf_to(pg.handle, dst, opts); py::gil_scoped_acquire acq; + auto maxc = std::numeric_limits::max(); + if (dtype == "int64") { + py::array_t dist_arr(res.first.size()); + auto* out = dist_arr.mutable_data(); + for (std::size_t i=0;i(maxc) : static_cast(res.first[i]); + return py::make_tuple(std::move(dist_arr), res.second); + } else if (dtype == "float64") { + py::array_t dist_arr(res.first.size()); + auto* out = dist_arr.mutable_data(); + for (std::size_t i=0;i::infinity() : static_cast(res.first[i]); + return py::make_tuple(std::move(dist_arr), res.second); + } else { + throw py::value_error("dtype must be 'float64' or 'int64'"); + } + }, py::arg("graph"), py::arg("dst"), py::kw_only(), py::arg("selection") = py::none(), py::arg("residual") = py::none(), py::arg("node_mask") = py::none(), py::arg("edge_mask") = py::none(), py::arg("multipath") = true, py::arg("fanout_edges") = py::none(), py::arg("dtype") = "float64", + "Shortest paths from every node to dst (reverse SPF). Returns (distances_to_dst, dag); the DAG is forward-oriented and valid for placement from any node toward dst. fanout_edges are added to the DAG regardless of cost (see shortest_paths.hpp).") .def("ksp", [](const Algorithms& algs, const PyGraph& pg, std::int32_t src, std::int32_t dst, int k, py::object max_cost_factor, bool unique, py::object node_mask, py::object edge_mask, std::string dtype){ if (src < 0 || src >= pg.num_nodes || dst < 0 || dst >= pg.num_nodes) throw py::value_error("src/dst out of range"); @@ -489,6 +541,14 @@ PYBIND11_MODULE(_netgraph_core, m, py::mod_gil_not_used()) { }) .def_property_readonly("graph", [](const FlowGraph& fg){ return &fg.graph(); }, py::return_value_policy::reference_internal) .def("place", [](FlowGraph& fg, const FlowIndex& idx, std::int32_t src, std::int32_t dst, const PredDAG& dag, double amount, FlowPlacement placement){ py::gil_scoped_release rel; auto placed = fg.place(idx, src, dst, dag, amount, placement); py::gil_scoped_acquire acq; return placed; }, py::arg("index"), py::arg("src"), py::arg("dst"), py::arg("dag"), py::arg("amount"), py::arg("flow_placement") = FlowPlacement::Proportional) + .def("place_with_drops", [](FlowGraph& fg, const FlowIndex& idx, std::int32_t src, std::int32_t dst, const PredDAG& dag, double amount, FlowPlacement placement){ + std::vector> drops; + double placed; + { py::gil_scoped_release rel; placed = fg.place(idx, src, dst, dag, amount, placement, &drops); } + py::list out; for (auto const& pr : drops) out.append(py::make_tuple(pr.first, pr.second)); + return py::make_tuple(placed, out); + }, py::arg("index"), py::arg("src"), py::arg("dst"), py::arg("dag"), py::arg("amount"), py::arg("flow_placement") = FlowPlacement::EqualBalancedLossy, + "Like place(), but also returns the per-edge dropped volume as a list of (edge_id, dropped) pairs. Only EQUAL_BALANCED_LOSSY drops flow; other placements return an empty list.") .def("remove", [](FlowGraph& fg, const FlowIndex& idx){ py::gil_scoped_release rel; fg.remove(idx); py::gil_scoped_acquire acq; }) .def("remove_by_class", [](FlowGraph& fg, std::int32_t cls){ py::gil_scoped_release rel; fg.remove_by_class(cls); py::gil_scoped_acquire acq; }) .def("reset", [](FlowGraph& fg){ py::gil_scoped_release rel; fg.reset(); py::gil_scoped_acquire acq; }) diff --git a/include/netgraph/core/algorithms.hpp b/include/netgraph/core/algorithms.hpp index c5ca458..0f22bfd 100644 --- a/include/netgraph/core/algorithms.hpp +++ b/include/netgraph/core/algorithms.hpp @@ -21,6 +21,11 @@ class Algorithms { return backend_->spf(gh, src, opts); } + [[nodiscard]] std::pair, PredDAG> + spf_to(const GraphHandle& gh, NodeId dst, const SpfToOptions& opts) const { + return backend_->spf_to(gh, dst, opts); + } + [[nodiscard]] std::vector, PredDAG>> ksp(const GraphHandle& gh, NodeId src, NodeId dst, const KspOptions& opts) const { return backend_->ksp(gh, src, dst, opts); diff --git a/include/netgraph/core/backend.hpp b/include/netgraph/core/backend.hpp index 1c2ba73..724de2b 100644 --- a/include/netgraph/core/backend.hpp +++ b/include/netgraph/core/backend.hpp @@ -55,6 +55,12 @@ class Backend { [[nodiscard]] virtual std::pair, PredDAG> spf( const GraphHandle& gh, NodeId src, const SpfOptions& opts) = 0; + // Computes shortest paths from every node to dst (reverse SPF); see + // shortest_paths_to() in shortest_paths.hpp. Returns distances to dst and a + // forward-oriented predecessor DAG valid for placement from any node toward dst. + [[nodiscard]] virtual std::pair, PredDAG> spf_to( + const GraphHandle& gh, NodeId dst, const SpfToOptions& opts) = 0; + // Computes maximum flow between a source and destination node. // // Arguments: diff --git a/include/netgraph/core/flow_graph.hpp b/include/netgraph/core/flow_graph.hpp index f17a518..8687d36 100644 --- a/include/netgraph/core/flow_graph.hpp +++ b/include/netgraph/core/flow_graph.hpp @@ -40,10 +40,15 @@ class FlowGraph { // Access underlying graph (const) [[nodiscard]] const StrictMultiDiGraph& graph() const noexcept { return *g_; } -// Apply placement and record per-edge allocations for this flow. Returns placed amount. + // Apply placement and record per-edge allocations for this flow. Returns placed + // amount. `drops`, when given, receives the per-edge dropped volume of an + // EqualBalancedLossy placement (see FlowState::place_on_dag); other placements + // leave it untouched. Dropped volume is not part of the ledger: it never + // occupied an edge, so removing the flow reverts only what was carried. [[nodiscard]] Flow place(const FlowIndex& idx, NodeId src, NodeId dst, const PredDAG& dag, Flow amount, - FlowPlacement placement); + FlowPlacement placement, + std::vector>* drops = nullptr); // Remove a specific flow, reverting its edge allocations from the ledger. void remove(const FlowIndex& idx); diff --git a/include/netgraph/core/flow_policy.hpp b/include/netgraph/core/flow_policy.hpp index 7079d81..1c0b455 100644 --- a/include/netgraph/core/flow_policy.hpp +++ b/include/netgraph/core/flow_policy.hpp @@ -146,9 +146,12 @@ class FlowPolicy { // // With static paths configured the policy neither creates additional flows nor // reoptimizes: max_path_cost, max_path_cost_factor, min_flow_count and - // reoptimize_flows_on_each_placement are inert. EqualBalanced spreads over the - // usable (up) bundles only. flow_count() reports the usable count U; the - // supplied count N is the caller's, so down LSPs = N - flow_count(). + // reoptimize_flows_on_each_placement are inert. EqualBalanced and + // EqualBalancedFixed spread over the usable (up) bundles only and equalize the + // carried volume per bundle; EqualBalancedLossy offers each up bundle an equal + // share and lets it carry what fits (no equalizing rebalance), so placed is the + // delivered total. flow_count() reports the usable count U; the supplied count + // N is the caller's, so down LSPs = N - flow_count(). // // Throws std::invalid_argument if bundles is empty, the policy already holds // flows (remove_demand() first), shortest_path=true is configured (single- diff --git a/include/netgraph/core/flow_state.hpp b/include/netgraph/core/flow_state.hpp index bdc327e..a9f746c 100644 --- a/include/netgraph/core/flow_state.hpp +++ b/include/netgraph/core/flow_state.hpp @@ -53,12 +53,25 @@ class FlowState { // saturates. Re-invoking this on the updated residuals changes the effective // next-hop set (progressive traffic-engineering behavior) and is outside // "single-pass ECMP admission". + // + // EqualBalancedFixed uses the same global scale but takes the split set from + // the DAG edges with capacity rather than with residual, so a member that has + // been saturated since the DAG was built yields scale 0 and nothing is placed. + // + // EqualBalancedLossy splits over the same capacity-based set without scaling: + // each edge carries min(share, residual), the excess is dropped, and the + // returned amount is what reaches dst. When `drops` is given it receives one + // (edge, dropped) entry per edge that dropped flow. Drops are recorded only + // for finite requested_flow; an infinite request fills every source edge and + // records no drops. Other placements never write to `drops`. [[nodiscard]] Flow place_on_dag(NodeId src, NodeId dst, const PredDAG& dag, Flow requested_flow, FlowPlacement placement, // Optional trace collector to record per-edge allocations applied by this call - std::vector>* trace = nullptr); + std::vector>* trace = nullptr, + // Optional collector for per-edge dropped volume (EqualBalancedLossy only) + std::vector>* drops = nullptr); // Convenience: run repeated placements until exhaustion (or single tier when // shortest_path=true). Returns total placed flow. Uses internal residual. diff --git a/include/netgraph/core/options.hpp b/include/netgraph/core/options.hpp index d82dcce..c084e4b 100644 --- a/include/netgraph/core/options.hpp +++ b/include/netgraph/core/options.hpp @@ -17,6 +17,17 @@ struct SpfOptions { std::span edge_mask {}; }; +// Options for shortest_paths_to (reverse SPF toward one destination). See +// shortest_paths.hpp for the semantics of fanout_edges. +struct SpfToOptions { + bool multipath { true }; + EdgeSelection selection {}; + std::span residual {}; + std::span node_mask {}; + std::span edge_mask {}; + std::span fanout_edges {}; +}; + struct KspOptions { int k { 1 }; std::optional max_cost_factor {}; diff --git a/include/netgraph/core/shortest_paths.hpp b/include/netgraph/core/shortest_paths.hpp index 6c91d3d..e22699d 100644 --- a/include/netgraph/core/shortest_paths.hpp +++ b/include/netgraph/core/shortest_paths.hpp @@ -93,3 +93,39 @@ resolve_to_paths(const PredDAG& dag, NodeId src, NodeId dst, std::optional max_paths = std::nullopt); } // namespace netgraph::core + +namespace netgraph::core { + +// Compute shortest paths *to* dst from every node (Dijkstra over the in-adjacency). +// Returns (distances, dag) where distances[u] is the cost of a shortest u -> dst walk +// (INT64_MAX if none) and dag is a forward-oriented PredDAG: parents[v] holds (u, e) +// for every edge e = u -> v that lies on a shortest u -> dst walk. Unlike a DAG from +// shortest_paths(), which is rooted at one source, this DAG is valid for placement +// from *any* node toward dst, and every walk it contains from a node u to dst has +// cost distances[u]. It is the union of the per-source shortest-path DAGs toward dst. +// +// Parameters mirror shortest_paths(): multipath keeps every equal-cost successor per +// node (false keeps one, preferring higher bottleneck capacity toward dst); +// selection, residual, node_mask and edge_mask apply as there. Zero-cost edges get +// the same acyclicity guard (a successor is recorded only while its node is +// unsettled). +// +// fanout_edges: edges to add to the DAG regardless of cost, modelling an origin +// whose first hop is decided by a traffic split rather than by routing (e.g. a +// pseudo source whose demand originates evenly at every attached real source). +// Each edge u -> v is added as a parent entry of v when it passes the masks and +// capacity gate and v has a finite distance; entries the SPF already recorded are +// not duplicated, and distances[u] is set to cost(e) + distances[v] if u was +// unreachable. Every fanout edge must leave a node that has no incoming DAG entry, +// so the result stays acyclic; a violating edge throws std::invalid_argument, as +// does an out-of-range edge id. +[[nodiscard]] std::pair, PredDAG> +shortest_paths_to(const StrictMultiDiGraph& g, NodeId dst, + bool multipath, + const EdgeSelection& selection, + std::span residual = {}, + std::span node_mask = {}, + std::span edge_mask = {}, + std::span fanout_edges = {}); + +} // namespace netgraph::core diff --git a/include/netgraph/core/types.hpp b/include/netgraph/core/types.hpp index 0d709d9..64aa6d2 100644 --- a/include/netgraph/core/types.hpp +++ b/include/netgraph/core/types.hpp @@ -56,9 +56,28 @@ struct FlowIndexHash { // out of scope for EqualBalanced. // // - Proportional may be used iteratively (e.g., for max-flow). +// +// - EqualBalanced builds its split set from the DAG edges that currently have +// residual, so re-invoking it on updated residuals shrinks the next-hop set +// (progressive behavior, used by place_max_flow and FlowPolicy fills). +// +// - EqualBalancedFixed builds its split set from the DAG edges that have +// *capacity* (the topology's next-hop set), and admits with the same single +// global scale computed from residual headroom. A member saturated since the +// DAG was built therefore blocks admission entirely (scale 0): this is +// lossless hash-ECMP admission with a forwarding table that does not react +// to load. +// +// - EqualBalancedLossy also splits over the topology's next-hop set, but does +// not scale: every edge carries min(share, residual) and the excess is +// dropped, deficits propagate downstream, and the placed amount is the volume +// that reaches dst. This is best-effort hash-ECMP forwarding; the dropped +// share per edge is available through the drop trace of place_on_dag. enum class FlowPlacement { - Proportional = 1, // Distribute flow proportionally to residual capacity (like ECMP with weights) - EqualBalanced = 2 // Split equally per parallel edge on a fixed DAG (single-pass ECMP admission) + Proportional = 1, // Distribute flow proportionally to residual capacity (like ECMP with weights) + EqualBalanced = 2, // Split equally per parallel edge on a fixed DAG (single-pass ECMP admission) + EqualBalancedFixed = 3, // Equal split over the topology next-hop set; a saturated member blocks admission + EqualBalancedLossy = 4 // Equal split over the topology next-hop set; excess over residual is dropped }; // Tie-breaking rule when multiple equal-cost edges exist between the same (u,v) pair. diff --git a/pyproject.toml b/pyproject.toml index 5ab88d5..214b299 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "scikit_build_core.build" [project] name = "netgraph-core" -version = "0.8.0" +version = "0.9.0" description = "C++ implementation of graph algorithms for network flow analysis and traffic engineering with Python bindings" readme = "README.md" requires-python = ">=3.11" diff --git a/python/netgraph_core/_docs.py b/python/netgraph_core/_docs.py index c873f0e..a69f652 100644 --- a/python/netgraph_core/_docs.py +++ b/python/netgraph_core/_docs.py @@ -56,13 +56,28 @@ class FlowPlacement: EQUAL_BALANCED (ECMP): Single-pass admission on a fixed shortest-path DAG (Dijkstra). Computes one global scale so no edge is oversubscribed under equal - per-edge splits, places once, and stops. Re-invoking on updated - residuals changes the next-hop set (progressive traffic-engineering behavior). + per-edge splits, places once, and stops. The split set is the DAG's + edges that still have residual, so re-invoking on updated residuals + changes the next-hop set (progressive traffic-engineering behavior). ECMP = Equal-Cost Multi-Path; WCMP = Weighted-Cost Multi-Path. + + EQUAL_BALANCED_FIXED: The same single-pass admission, but the split set is + the DAG's edges with capacity (the topology's next-hop set), so a + member saturated since the DAG was computed drives the scale to 0 and + nothing is admitted. Models lossless hash-ECMP with a forwarding table + that does not react to load. + + EQUAL_BALANCED_LOSSY: Equal split over the same capacity-based set with no + scaling: each edge carries min(share, residual) and drops the excess, + deficits propagate downstream, and the placed amount is what reaches + the destination. Models best-effort hash-ECMP forwarding; per-edge + drops are available from FlowGraph.place_with_drops. """ PROPORTIONAL: ClassVar[FlowPlacement] EQUAL_BALANCED: ClassVar[FlowPlacement] + EQUAL_BALANCED_FIXED: ClassVar[FlowPlacement] + EQUAL_BALANCED_LOSSY: ClassVar[FlowPlacement] __members__: ClassVar[dict[str, FlowPlacement]] def __init__(self, value: int) -> None: ... @@ -176,6 +191,24 @@ def place( flow_placement: FlowPlacement = ..., ) -> float: ... + def place_with_drops( + self, + index: "FlowIndex", + src: int, + dst: int, + dag: "PredDAG", + amount: float, + flow_placement: FlowPlacement = ..., + ) -> tuple[float, list[tuple[int, float]]]: + """Like place(), returning (placed, drops). + + `drops` lists (edge_id, dropped_volume) for every edge that dropped + flow under EQUAL_BALANCED_LOSSY; other placements return an empty + list. Dropped volume never enters the ledger, so remove() reverts only + what was carried. + """ + ... + def remove(self, index: "FlowIndex") -> None: ... def remove_by_class(self, cls: int) -> None: ... def reset(self) -> None: ... @@ -249,7 +282,9 @@ def place_on_dag( EqualBalanced is **single-pass ECMP admission** on the provided DAG: we compute one global scale so no edge is oversubscribed under equal per-edge splits, apply it once, and return. Re-invoking on updated residuals changes - the next-hop set (progressive behavior). + the next-hop set (progressive behavior). EQUAL_BALANCED_FIXED keeps the + topology's next-hop set instead (a saturated member yields scale 0), and + EQUAL_BALANCED_LOSSY forwards best-effort over it (see FlowPlacement). Returns: Amount of flow actually placed (may be less than requested). @@ -667,6 +702,54 @@ def spf( """ ... + def spf_to( + self, + graph: "Graph", + dst: int, + *, + selection: Optional[EdgeSelection] = None, + residual: Optional["np.ndarray"] = None, + node_mask: Optional["np.ndarray"] = None, + edge_mask: Optional["np.ndarray"] = None, + multipath: bool = True, + fanout_edges: Optional[Sequence[int]] = None, + dtype: str = "float64", + ) -> tuple["np.ndarray", "PredDAG"]: + """Shortest paths from every node *to* dst (reverse SPF). + + Returns ``(distances_to_dst, dag)``. ``distances_to_dst[u]`` is the cost + of a shortest ``u -> dst`` walk (inf / int64 max if none). The DAG is + forward-oriented, so it works anywhere a PredDAG is accepted, and is + valid for placement from *any* node toward ``dst``: every walk it + holds from a node ``u`` to ``dst`` costs ``distances_to_dst[u]``. It is + the union of the per-source shortest-path DAGs toward ``dst``. + + Args: + graph: Graph handle + dst: Destination node + selection: Edge selection policy (as for spf) + residual: Optional 1-D float64 array of residuals (copied); forces + capacity gating as for spf + node_mask: Optional 1-D bool mask (length num_nodes), copied + edge_mask: Optional 1-D bool mask (length num_edges), copied + multipath: Keep every equal-cost successor per node; False keeps + one, preferring higher bottleneck capacity toward dst + fanout_edges: Edge ids added to the DAG regardless of cost, for an + origin whose first hop is a traffic split rather than a + routing decision (a pseudo source fanning out over every + attached real source). Each edge is added when it passes the + masks and capacity gate and its head has a finite distance; + its tail must have no incoming DAG entry + dtype: "float64" (inf for unreachable) or "int64" (max for unreachable) + + Raises: + TypeError: If arrays have wrong dtype, ndim, or length. + ValueError: If dst or a fanout edge id is out of range, or a + fanout edge leaves a node that already has an incoming DAG + entry. + """ + ... + def ksp( self, graph: "Graph", diff --git a/src/cpu_backend.cpp b/src/cpu_backend.cpp index 2f64537..b173d7d 100644 --- a/src/cpu_backend.cpp +++ b/src/cpu_backend.cpp @@ -37,6 +37,20 @@ class CpuBackend final : public Backend { opts.residual, opts.node_mask, opts.edge_mask); } + std::pair, PredDAG> spf_to( + const GraphHandle& gh, NodeId dst, const SpfToOptions& opts) override { + const StrictMultiDiGraph& g = *gh.graph; + if (!opts.node_mask.empty() && opts.node_mask.size() != static_cast(g.num_nodes())) { + throw std::invalid_argument("CpuBackend::spf_to: node_mask length mismatch"); + } + if (!opts.edge_mask.empty() && opts.edge_mask.size() != static_cast(g.num_edges())) { + throw std::invalid_argument("CpuBackend::spf_to: edge_mask length mismatch"); + } + return netgraph::core::shortest_paths_to(g, dst, opts.multipath, opts.selection, + opts.residual, opts.node_mask, opts.edge_mask, + opts.fanout_edges); + } + std::pair max_flow( const GraphHandle& gh, NodeId src, NodeId dst, const MaxFlowOptions& opts) override { const StrictMultiDiGraph& g = *gh.graph; diff --git a/src/flow_graph.cpp b/src/flow_graph.cpp index c073e10..44ac4ae 100644 --- a/src/flow_graph.cpp +++ b/src/flow_graph.cpp @@ -62,7 +62,8 @@ FlowGraph& FlowGraph::operator=(FlowGraph&& other) noexcept { Flow FlowGraph::place(const FlowIndex& idx, NodeId src, NodeId dst, const PredDAG& dag, Flow amount, - FlowPlacement placement) { + FlowPlacement placement, + std::vector>* drops) { if (amount <= 0.0) return 0.0; ++version_; // residuals may change from here on @@ -73,7 +74,7 @@ Flow FlowGraph::place(const FlowIndex& idx, NodeId src, NodeId dst, // Delegate placement to FlowState, which returns the actual placed flow and // populates bucket with per-edge allocations (EdgeId, Flow) pairs. - Flow placed = fs_.place_on_dag(src, dst, dag, amount, placement, &bucket); + Flow placed = fs_.place_on_dag(src, dst, dag, amount, placement, &bucket, drops); // Coalesce and filter: merge duplicate EdgeIds. Keep any positive totals // (do not drop sub-kMinFlow amounts) to preserve exact reversibility. diff --git a/src/flow_policy.cpp b/src/flow_policy.cpp index def6672..3578af4 100644 --- a/src/flow_policy.cpp +++ b/src/flow_policy.cpp @@ -26,6 +26,16 @@ namespace netgraph::core { +namespace { +// The equal-balanced placements share FlowPolicy's per-flow target and DAG +// refresh logic (and, except for the lossy mode, the equalizing rebalance in +// place_demand); they differ only inside FlowState::place_on_dag. +constexpr bool is_equal_balanced(FlowPlacement p) noexcept { + return p == FlowPlacement::EqualBalanced || p == FlowPlacement::EqualBalancedFixed || + p == FlowPlacement::EqualBalancedLossy; +} +} // namespace + /* Reject uses that would silently produce a wrong answer: - a FlowGraph wrapping a different graph than the policy routes on (SPF would run on one topology while flow is placed on another); @@ -76,7 +86,7 @@ std::optional> FlowPolicy::get_path_bundle(const FlowGr // - Hash-ECMP with EqualBalanced: use all equal-cost edges to maximize fanout if (!multipath_) { sel.multi_edge = false; - } else if (flow_placement_ == FlowPlacement::EqualBalanced) { + } else if (is_equal_balanced(flow_placement_)) { sel.multi_edge = true; } @@ -86,8 +96,10 @@ std::optional> FlowPolicy::get_path_bundle(const FlowGr // Residual awareness is controlled by require_capacity_: // - require_capacity=true: Require edges to have capacity, routes adapt to residuals (SDN/TE behavior) // - require_capacity=false: Routes based on costs only (IP/IGP behavior) - // Additionally, for EqualBalanced mode with minimum flow threshold, we use residuals. - const bool require_residual = (require_capacity_ || (flow_placement_ == FlowPlacement::EqualBalanced && min_flow.has_value())); + // Passing a residual to SPF forces capacity-aware selection, so a cost-only + // policy must not pass one even when an equal-balanced per-flow target is + // set; the target only shapes how much is requested per flow, not the route. + const bool require_residual = require_capacity_; const auto residual = fg.residual_view(); // Edge mask: combine user-provided mask with minimum residual capacity threshold. @@ -99,7 +111,7 @@ std::optional> FlowPolicy::get_path_bundle(const FlowGr std::unique_ptr combined_edge_mask; std::span final_edge_mask; - if (require_residual && min_flow.has_value() && flow_placement_ != FlowPlacement::EqualBalanced) { + if (require_residual && min_flow.has_value() && !is_equal_balanced(flow_placement_)) { // Need to filter by min_flow threshold for Proportional mode combined_edge_mask.reset(new bool[residual.size()]); double thr = *min_flow; @@ -134,7 +146,7 @@ std::optional> FlowPolicy::get_path_bundle(const FlowGr PredDAG dag; Cost dst_cost; - const bool use_memo = (flow_placement_ == FlowPlacement::EqualBalanced); + const bool use_memo = (is_equal_balanced(flow_placement_)); const auto stamp = fg.state_stamp(); const bool with_residual = !opts.residual.empty(); std::size_t hit_idx = spf_memo_.size(); @@ -311,7 +323,11 @@ std::pair FlowPolicy::place_demand(FlowGraph& fg, // currently placed volume at the equal-share target), and the recursion's return // value telescoped to (placed_demand(), pre-rebalance leftover + volume lost in // rebalancing), which is reproduced after the loop. - if (flow_placement_ == FlowPlacement::EqualBalanced && !flows_.empty()) { + // Lossy placement is best-effort: each flow is offered its share and carries + // what fits, so equalizing what was *carried* would contradict the model + // (it would throttle the healthy flows down to the congested one). + if (is_equal_balanced(flow_placement_) && + flow_placement_ != FlowPlacement::EqualBalancedLossy && !flows_.empty()) { // Restore the reoptimize flag even if a round throws (bad_alloc is the only // realistic thrower here); otherwise the policy would stay permanently // non-reoptimizing. @@ -365,7 +381,7 @@ std::pair FlowPolicy::place_demand_body(FlowGraph& fg, // pinned policy that is the number of USABLE bundles (a head-end hashes over up // LSPs only); dynamically it is the configured max_flow_count. int eb_divisor = 0; - if (flow_placement_ == FlowPlacement::EqualBalanced) { + if (is_equal_balanced(flow_placement_)) { if (is_static) eb_divisor = static_cast(static_bundles_.size()); else if (max_flow_count_.has_value()) eb_divisor = *max_flow_count_; } @@ -413,7 +429,7 @@ std::pair FlowPolicy::place_demand_body(FlowGraph& fg, if (max_flow_count_.has_value()) { initial = std::min(initial, *max_flow_count_); } - auto min_req = (flow_placement_ == FlowPlacement::EqualBalanced && max_flow_count_.has_value()) + auto min_req = (is_equal_balanced(flow_placement_) && max_flow_count_.has_value()) ? std::optional(per_target) : min_flow; // Seeding places no flow, so residuals do not change between iterations and @@ -465,7 +481,7 @@ std::pair FlowPolicy::place_demand_body(FlowGraph& fg, // For multipath flows, this tracks saturated edges within the DAG. // For tunnel flows, this allows different tunnels to discover different paths // as residuals change, enabling natural fan-out across equal-cost paths. - if (flow_placement_ == FlowPlacement::EqualBalanced && !is_static) { + if (is_equal_balanced(flow_placement_) && !is_static) { if (auto pb = get_path_bundle(fg, f->src, f->dst, std::optional(per_target))) { f->dag = std::move(pb->first); f->cost = pb->second; @@ -515,7 +531,7 @@ std::pair FlowPolicy::place_demand_body(FlowGraph& fg, // A pinned policy neither grows its flow set nor reoptimizes: the pinned-ness // guard is explicit, never inferred from flow-count arithmetic. if (!is_static) { - if (flow_placement_ == FlowPlacement::EqualBalanced) { + if (is_equal_balanced(flow_placement_)) { if (max_flow_count_.has_value()) { // Bounded EB: add flows up to configured maximum. if (static_cast(flows_.size()) < *max_flow_count_) { diff --git a/src/flow_state.cpp b/src/flow_state.cpp index 3789853..08b2700 100644 --- a/src/flow_state.cpp +++ b/src/flow_state.cpp @@ -2,7 +2,7 @@ FlowState — residual capacities and placement over a fixed graph. Maintains per-edge residual capacity and cumulative edge flows. Supports - two placement strategies when pushing flow along an SPF DAG: + four placement strategies when pushing flow along an SPF DAG: - Proportional: distribute flow proportionally to residual capacity, processing nodes in topological order from source to destination. - EqualBalanced: distribute flow equally across available parallel edges, @@ -10,6 +10,15 @@ scale so no edge is oversubscribed, then return. Re-running on updated residuals intentionally changes the allowed next-hop set (progressive/TE); use place_max_flow() if you want that behavior. + - EqualBalancedFixed: the same single-pass admission, but the split set is + the DAG's edges with capacity (the topology's next-hop set) rather than + its edges with residual. A member saturated since the DAG was built + yields scale 0: lossless hash-ECMP admission with a load-blind + forwarding table. + - EqualBalancedLossy: equal split over the same capacity-based set with no + scaling; each edge carries min(share, residual) and drops the excess, + deficits propagate downstream, and the placed amount is what reaches + dst (best-effort hash-ECMP forwarding). */ #include "netgraph/core/flow_state.hpp" #include "netgraph/core/shortest_paths.hpp" @@ -122,12 +131,19 @@ struct GroupSet { } }; -// Build grouped edges by (parent u, child v) that can reach destination t, -// using the current residual snapshot. +// Build grouped edges by (parent u, child v) that can reach destination t. +// Membership is decided by residual (edges that can still carry flow) unless +// members_by_capacity is set, in which case every DAG edge with capacity is a +// member and sum_cap/min_cap still reflect the current residual -- so a member +// saturated since the DAG was built contributes min_cap = 0. The fixed and +// lossy equal-balanced placements use the latter to model a forwarding table +// that does not react to load. static void build_groups_residual(const StrictMultiDiGraph& g, const PredDAG& dag, NodeId t, const std::vector& residual, + bool members_by_capacity, GroupSet& gs) { + const auto capacity = g.capacity_view(); gs.groups.clear(); gs.eids.clear(); const auto& offsets = dag.parent_offsets; @@ -178,7 +194,8 @@ static void build_groups_residual(const StrictMultiDiGraph& g, if (parents[i] != u) continue; const auto eid0 = via[i]; const Cap c = residual[static_cast(eid0)]; - if (c >= kMinCap) { + const Cap gate = members_by_capacity ? capacity[static_cast(eid0)] : c; + if (gate >= kMinCap) { gs.eids.push_back(eid0); gr.sum_cap += c; gr.min_cap = std::min(gr.min_cap, c); @@ -241,15 +258,22 @@ void FlowState::reset(std::span residual_init) { Flow FlowState::place_on_dag(NodeId src, NodeId dst, const PredDAG& dag, Flow requested_flow, FlowPlacement placement, - std::vector>* trace) { + std::vector>* trace, + std::vector>* drops) { NGRAPH_PROFILE_SCOPE("place_on_dag"); const auto N = g_->num_nodes(); if (src < 0 || src >= N || dst < 0 || dst >= N || src == dst) return 0.0; + // The fixed and lossy equal-balanced placements model a forwarding table + // that does not react to load: their split set is every DAG edge with + // capacity, saturated or not. + const bool fixed_members = (placement == FlowPlacement::EqualBalancedFixed || + placement == FlowPlacement::EqualBalancedLossy); + // Build groups using current residual. `gs` is reused across rebuilds so its // buffers keep their capacity for the whole call. GroupSet gs; - build_groups_residual(*g_, dag, dst, residual_, gs); + build_groups_residual(*g_, dag, dst, residual_, fixed_members, gs); const auto& groups = gs.groups; Flow placed = static_cast(0.0); @@ -294,12 +318,92 @@ Flow FlowState::place_on_dag(NodeId src, NodeId dst, const PredDAG& dag, } } // Rebuild groups for next tier using updated residual - build_groups_residual(*g_, dag, dst, residual_, gs); + build_groups_residual(*g_, dag, dst, residual_, /*members_by_capacity=*/false, gs); build_reversed_residual(ws, N, groups); } + } else if (placement == FlowPlacement::EqualBalancedLossy) { + // EqualBalancedLossy: best-effort hash-ECMP forwarding. Every node splits + // what it received equally over its outgoing member edges; each edge carries + // min(share, residual) and drops the rest, so a deficit propagates downstream + // and the placed amount is what arrives at dst. No global scale. + const bool record_drops = (drops != nullptr) && std::isfinite(static_cast(requested_flow)); + + std::vector> succ(static_cast(N)); + for (std::size_t gi = 0; gi < groups.size(); ++gi) { + const auto& gr = groups[gi]; + if (gr.eid_count == 0) continue; + succ[static_cast(gr.to)].push_back(gi); // u -> v (group index) + } + + // Reachability from src over the member graph; nodes outside it never see flow. + std::vector reach(static_cast(N), 0); + { + std::queue q; q.push(src); reach[static_cast(src)] = 1; + while (!q.empty()) { + auto u = q.front(); q.pop(); + for (auto gi : succ[static_cast(u)]) { + auto v = groups[gi].from; + if (!reach[static_cast(v)]) { reach[static_cast(v)] = 1; q.push(v); } + } + } + } + if (!reach[static_cast(dst)]) return static_cast(0.0); + + // Per-node fan-out (number of member edges) for the equal per-edge split. + std::vector node_split(static_cast(N), 0); + std::vector indeg(static_cast(N), 0); + for (std::size_t u = 0; u < succ.size(); ++u) { + if (!reach[u]) continue; + int s = 0; + for (auto gi : succ[u]) { + s += static_cast(groups[gi].eid_count); + auto v = static_cast(groups[gi].from); + if (reach[v]) indeg[v] += 1; + } + node_split[u] = s; + } + + // Kahn's algorithm carrying actual volumes. Nodes are released once every + // parent has forwarded, so inflow[u] is final when u is processed. + std::queue q; + std::vector inflow(static_cast(N), 0.0); + q.push(src); + inflow[static_cast(src)] = static_cast(requested_flow); + while (!q.empty()) { + auto u = q.front(); q.pop(); + const double f_in = inflow[static_cast(u)]; + const int split = node_split[static_cast(u)]; + // Even a node that received nothing must release its children, or a + // reconvergent child stays blocked behind an unreachable parent. + const double per_edge = (split > 0 && f_in >= kEpsilon) ? f_in / static_cast(split) : 0.0; + for (auto gi : succ[static_cast(u)]) { + const auto& gr = groups[gi]; + if (per_edge > 0.0) { + for (auto eid : gs.edges_of(gr)) { + const auto ei = static_cast(eid); + const double res = static_cast(residual_[ei]); + const double carried = std::max(0.0, std::min(per_edge, res)); + if (carried > 0.0) { + edge_flow_[ei] += static_cast(carried); + residual_[ei] = static_cast(std::max(0.0, res - carried)); + inflow[static_cast(gr.from)] += carried; + if (trace) trace->emplace_back(eid, static_cast(carried)); + } + const double dropped = per_edge - carried; + if (record_drops && dropped >= kEpsilon) drops->emplace_back(eid, static_cast(dropped)); + } + } + auto v = static_cast(gr.from); + if (reach[v] && --indeg[v] == 0) q.push(static_cast(v)); + } + } + placed = static_cast(inflow[static_cast(dst)]); } else { - // EqualBalanced placement: split flow equally across parallel edges, with - // topological accumulation to correctly handle reconvergent DAGs. + // EqualBalanced / EqualBalancedFixed placement: split flow equally across + // parallel edges, with topological accumulation to correctly handle + // reconvergent DAGs. The two differ only in the split set: EqualBalanced + // keeps the groups that still have headroom, EqualBalancedFixed keeps every + // group with capacity so that a saturated member drives the scale to 0. // Build forward adjacency from parent u to child v for each group and // compute aggregated reverse capacities per group. @@ -310,9 +414,9 @@ Flow FlowState::place_on_dag(NodeId src, NodeId dst, const PredDAG& dag, if (gr.eid_count == 0) continue; // EB: group admissible total = min_edge_residual * |edges| const double cap_rev = static_cast(gr.min_cap) * static_cast(gr.eid_count); - if (cap_rev >= kMinCap) { + if (cap_rev >= kMinCap || fixed_members) { succ[static_cast(gr.to)].push_back(gi); // u -> v (group index) - rev_cap[gi] = cap_rev; + rev_cap[gi] = cap_rev >= kMinCap ? cap_rev : 0.0; } } diff --git a/src/shortest_paths.cpp b/src/shortest_paths.cpp index cd31d8c..73bdedb 100644 --- a/src/shortest_paths.cpp +++ b/src/shortest_paths.cpp @@ -498,3 +498,234 @@ shortest_paths(const StrictMultiDiGraph& g, NodeId src, } } // namespace netgraph::core + +/* + Reverse Dijkstra: shortest paths from every node *to* one destination. + + Mirrors shortest_paths_core over the in-adjacency (in_row_offsets / + in_col_indices / in_adj_edge_index): the priority queue settles nodes by their + distance to dst, edge selection runs per (u -> v) parallel group exactly as in + the forward variant, and node-level tie-breaking in single-path mode prefers + the higher bottleneck capacity toward dst. Successor entries (v, e) are kept + per tail node u while u is unsettled, then bucketed by v into the forward + PredDAG layout that placement consumes. +*/ +namespace netgraph::core { + +namespace { +static std::pair, PredDAG> +shortest_paths_to_core(const StrictMultiDiGraph& g, NodeId target, + bool multipath, + const EdgeSelection& selection, + std::span residual, + std::span node_mask, + std::span edge_mask, + std::span fanout_edges) { + NGRAPH_PROFILE_SCOPE("shortest_paths_to_core"); + const auto N = g.num_nodes(); + const auto E = g.num_edges(); + const auto irow = g.in_row_offsets_view(); + const auto icol = g.in_col_indices_view(); + const auto iaei = g.in_adj_edge_index_view(); + const auto esrc = g.edge_src_view(); + const auto edst = g.edge_dst_view(); + const auto cost = g.cost_view(); + const auto cap = g.capacity_view(); + + std::vector dist(static_cast(N), std::numeric_limits::max()); + std::vector min_residual_to_dst(static_cast(N), static_cast(0)); + + const bool use_node_mask = (node_mask.size() == static_cast(N)); + const bool use_edge_mask = (edge_mask.size() == static_cast(E)); + const bool has_residual = (residual.size() == static_cast(E)); + const bool require_cap = selection.require_capacity || has_residual; + const bool target_allowed = (target >= 0 && target < N && + (!use_node_mask || node_mask[static_cast(target)])); + + // Successor lists per tail node u as a flat intrusive list of (v, e) entries. + std::vector succ_head(static_cast(N), -1); + std::vector succ_tail(static_cast(N), -1); + std::vector ent_node; ent_node.reserve(static_cast(E)); + std::vector ent_edge; ent_edge.reserve(static_cast(E)); + std::vector ent_next; ent_next.reserve(static_cast(E)); + // has_in_entry[v]: some DAG entry u -> v exists; a fanout edge may not leave such a node. + std::vector has_in_entry(static_cast(N), 0); + auto succ_clear = [&](std::size_t u){ succ_head[u] = -1; succ_tail[u] = -1; }; + auto succ_append = [&](std::size_t u, NodeId v, EdgeId e){ + const auto idx = static_cast(ent_node.size()); + ent_node.push_back(v); ent_edge.push_back(e); ent_next.push_back(-1); + if (succ_head[u] < 0) { succ_head[u] = idx; } + else { ent_next[static_cast(succ_tail[u])] = idx; } + succ_tail[u] = idx; + }; + + if (target_allowed) { + dist[static_cast(target)] = static_cast(0); + min_residual_to_dst[static_cast(target)] = std::numeric_limits::max(); + + using QItem = std::tuple; + auto cmp = [](const QItem& a, const QItem& b) { return a > b; }; + std::priority_queue, decltype(cmp)> pq(cmp); + pq.emplace(static_cast(0), -std::numeric_limits::max(), target); + std::vector settled(static_cast(N), 0); + std::vector sel_buf; sel_buf.reserve(16); + + while (!pq.empty()) { + auto [d_v, neg_res_v, v] = pq.top(); pq.pop(); + if (v < 0 || v >= N) continue; + if (d_v > dist[static_cast(v)]) continue; + if (!multipath && d_v == dist[static_cast(v)] && + -neg_res_v < min_residual_to_dst[static_cast(v)] - kEpsilon) continue; + settled[static_cast(v)] = 1; + + // In-edges of v, clustered by tail node u (edges are sorted by (src, dst)). + auto start = static_cast(irow[static_cast(v)]); + auto end = static_cast(irow[static_cast(v)+1]); + std::size_t i = start; + while (i < end) { + NodeId u = icol[i]; + if (use_node_mask && !node_mask[static_cast(u)]) { + std::size_t j_skip = i; while (j_skip < end && icol[j_skip] == u) ++j_skip; i = j_skip; continue; + } + Cost min_edge_cost = std::numeric_limits::max(); + std::vector& selected_edges = sel_buf; selected_edges.clear(); + double best_rem_for_min_cost = -1.0; + std::size_t j = i; + int best_edge_id = -1; + for (; j < end && icol[j] == u; ++j) { + auto e = static_cast(iaei[j]); + if (use_edge_mask && !edge_mask[e]) continue; + const Cap rem = has_residual ? residual[e] : cap[e]; + if (require_cap && rem < kMinCap) continue; + const Cost ecost = static_cast(cost[e]); + if (ecost < min_edge_cost) { + min_edge_cost = ecost; + selected_edges.clear(); + if (selection.multi_edge) { + selected_edges.push_back(static_cast(iaei[j])); + } else { + best_edge_id = static_cast(e); + best_rem_for_min_cost = static_cast(rem); + } + } else if (ecost == min_edge_cost) { + if (selection.multi_edge) { + selected_edges.push_back(static_cast(iaei[j])); + } else if (selection.tie_break == EdgeTieBreak::PreferHigherResidual) { + if (static_cast(rem) > best_rem_for_min_cost + kEpsilon) { + best_edge_id = static_cast(e); + best_rem_for_min_cost = static_cast(rem); + } else if (std::abs(static_cast(rem) - best_rem_for_min_cost) <= kEpsilon) { + if (best_edge_id < 0 || static_cast(e) < best_edge_id) best_edge_id = static_cast(e); + } + } else { + if (best_edge_id < 0 || static_cast(e) < best_edge_id) best_edge_id = static_cast(e); + } + } + } + if (!selection.multi_edge && best_edge_id >= 0) { + selected_edges.clear(); + selected_edges.push_back(static_cast(best_edge_id)); + } + if (!selected_edges.empty()) { + const Cost new_cost = static_cast(d_v + min_edge_cost); + const auto u_idx = static_cast(u); + Cap max_edge_residual = static_cast(0); + for (auto edge_id : selected_edges) { + const Cap rem = has_residual ? residual[static_cast(edge_id)] + : cap[static_cast(edge_id)]; + if (rem > max_edge_residual) max_edge_residual = rem; + } + const Cap path_residual = std::min(min_residual_to_dst[static_cast(v)], max_edge_residual); + if (new_cost < dist[u_idx] || + (!multipath && new_cost == dist[u_idx] && !settled[u_idx] && + path_residual > min_residual_to_dst[u_idx] + kEpsilon)) { + dist[u_idx] = new_cost; + min_residual_to_dst[u_idx] = path_residual; + succ_clear(u_idx); + for (auto sel_e : selected_edges) succ_append(u_idx, v, sel_e); + has_in_entry[static_cast(v)] = 1; + pq.emplace(new_cost, -path_residual, u); + } else if (multipath && new_cost == dist[u_idx] && !settled[u_idx]) { + for (auto sel_e : selected_edges) succ_append(u_idx, v, sel_e); + has_in_entry[static_cast(v)] = 1; + } + } + i = j; + } + } + } + + // Forced fan-out entries (see the header). + for (auto e_raw : fanout_edges) { + if (e_raw < 0 || e_raw >= E) { + throw std::invalid_argument("shortest_paths_to: fanout edge id out of range"); + } + const auto e = static_cast(e_raw); + const NodeId u = esrc[e]; + const NodeId v = edst[e]; + if (has_in_entry[static_cast(u)]) { + throw std::invalid_argument( + "shortest_paths_to: a fanout edge must leave a node with no incoming DAG entry " + "(otherwise the DAG could contain a cycle)"); + } + if (use_edge_mask && !edge_mask[e]) continue; + if (use_node_mask && (!node_mask[static_cast(u)] || !node_mask[static_cast(v)])) continue; + const Cap rem = has_residual ? residual[e] : cap[e]; + if (require_cap && rem < kMinCap) continue; + if (dist[static_cast(v)] == std::numeric_limits::max()) continue; + bool present = false; + for (std::int32_t i = succ_head[static_cast(u)]; i >= 0; i = ent_next[static_cast(i)]) { + if (ent_edge[static_cast(i)] == e_raw) { present = true; break; } + } + if (present) continue; + succ_append(static_cast(u), v, e_raw); + if (dist[static_cast(u)] == std::numeric_limits::max()) { + dist[static_cast(u)] = static_cast(cost[e]) + dist[static_cast(v)]; + } + } + + // Bucket successor entries (u -> v via e) by v into the forward PredDAG layout. + PredDAG dag; + dag.parent_offsets.assign(static_cast(N+1), 0); + for (std::int32_t u = 0; u < N; ++u) { + for (std::int32_t i = succ_head[static_cast(u)]; i >= 0; i = ent_next[static_cast(i)]) { + dag.parent_offsets[static_cast(ent_node[static_cast(i)]) + 1] += 1; + } + } + for (std::size_t k = 1; k < dag.parent_offsets.size(); ++k) dag.parent_offsets[k] += dag.parent_offsets[k-1]; + dag.parents.resize(static_cast(dag.parent_offsets.back())); + dag.via_edges.resize(static_cast(dag.parent_offsets.back())); + std::vector cursor(dag.parent_offsets.begin(), dag.parent_offsets.end() - 1); + for (std::int32_t u = 0; u < N; ++u) { + for (std::int32_t i = succ_head[static_cast(u)]; i >= 0; i = ent_next[static_cast(i)]) { + const auto v = static_cast(ent_node[static_cast(i)]); + const auto pos = static_cast(cursor[v]++); + dag.parents[pos] = u; + dag.via_edges[pos] = ent_edge[static_cast(i)]; + } + } + return {std::move(dist), std::move(dag)}; +} +} // namespace + +std::pair, PredDAG> +shortest_paths_to(const StrictMultiDiGraph& g, NodeId dst, + bool multipath, + const EdgeSelection& selection, + std::span residual, + std::span node_mask, + std::span edge_mask, + std::span fanout_edges) { + if (!node_mask.empty() && node_mask.size() != static_cast(g.num_nodes())) { + throw std::invalid_argument("shortest_paths_to: node_mask length mismatch"); + } + if (!edge_mask.empty() && edge_mask.size() != static_cast(g.num_edges())) { + throw std::invalid_argument("shortest_paths_to: edge_mask length mismatch"); + } + if (!residual.empty() && residual.size() != static_cast(g.num_edges())) { + throw std::invalid_argument("shortest_paths_to: residual length mismatch"); + } + return shortest_paths_to_core(g, dst, multipath, selection, residual, node_mask, edge_mask, fanout_edges); +} + +} // namespace netgraph::core diff --git a/tests/cpp/flow_state_tests.cpp b/tests/cpp/flow_state_tests.cpp index 68141dd..10b206f 100644 --- a/tests/cpp/flow_state_tests.cpp +++ b/tests/cpp/flow_state_tests.cpp @@ -469,3 +469,125 @@ TEST(FlowState, RepeatedPlacementAfterResetAndIndependentInstancesStayStable) { EXPECT_NEAR(fs1.edge_flow_view()[i], first_flows[i], 1e-9); } } + +// --------------------------------------------------------------------------- +// EqualBalancedFixed / EqualBalancedLossy: load-blind hash-ECMP models. +// --------------------------------------------------------------------------- +#include "netgraph/core/flow_graph.hpp" + +namespace { +// 0 -> 1 over two parallel equal-cost edges: edge 0 cap 10, edge 1 cap 100. +StrictMultiDiGraph make_unbalanced_pair() { + std::int32_t src[2] = {0, 0}; + std::int32_t dst[2] = {1, 1}; + double cap[2] = {10.0, 100.0}; + std::int64_t cost[2] = {1, 1}; + return StrictMultiDiGraph::from_arrays(2, + std::span(src, 2), std::span(dst, 2), std::span(cap, 2), std::span(cost, 2)); +} +PredDAG cost_only_dag(const StrictMultiDiGraph& g, NodeId s, NodeId t) { + EdgeSelection sel; sel.multi_edge = true; sel.require_capacity = false; sel.tie_break = EdgeTieBreak::Deterministic; + auto [dist, dag] = shortest_paths(g, s, t, /*multipath=*/true, sel, {}, {}, {}); + return dag; +} +} // namespace + +TEST(FlowState, EqualBalancedFixed_SaturatedMemberBlocksAdmission) { + auto g = make_unbalanced_pair(); + auto dag = cost_only_dag(g, 0, 1); + + FlowState fixed(g); + EXPECT_NEAR(fixed.place_on_dag(0, 1, dag, 20.0, FlowPlacement::EqualBalancedFixed), 20.0, 1e-9); + EXPECT_NEAR(fixed.edge_flow_view()[0], 10.0, 1e-9); + EXPECT_NEAR(fixed.edge_flow_view()[1], 10.0, 1e-9); + // The 10-unit member is saturated; equal hashing of any further demand would + // lose 1/2 of it, so nothing more is admitted losslessly. + EXPECT_NEAR(fixed.place_on_dag(0, 1, dag, 10.0, FlowPlacement::EqualBalancedFixed), 0.0, 1e-9); + EXPECT_NEAR(fixed.edge_flow_view()[1], 10.0, 1e-9) << "no flow may leak onto the remaining member"; + + // The progressive mode drops the saturated member from the split instead. + FlowState progressive(g); + EXPECT_NEAR(progressive.place_on_dag(0, 1, dag, 20.0, FlowPlacement::EqualBalanced), 20.0, 1e-9); + EXPECT_NEAR(progressive.place_on_dag(0, 1, dag, 10.0, FlowPlacement::EqualBalanced), 10.0, 1e-9); + EXPECT_NEAR(progressive.edge_flow_view()[1], 20.0, 1e-9); +} + +TEST(FlowState, EqualBalancedFixed_SinglePassMatchesEqualBalancedOnFreshState) { + auto g = make_unbalanced_pair(); + auto dag = cost_only_dag(g, 0, 1); + FlowState a(g), b(g); + EXPECT_NEAR(a.place_on_dag(0, 1, dag, 100.0, FlowPlacement::EqualBalanced), 20.0, 1e-9); + EXPECT_NEAR(b.place_on_dag(0, 1, dag, 100.0, FlowPlacement::EqualBalancedFixed), 20.0, 1e-9); + for (std::size_t i = 0; i < 2; ++i) EXPECT_NEAR(a.edge_flow_view()[i], b.edge_flow_view()[i], 1e-12); +} + +TEST(FlowState, EqualBalancedLossy_FillAndDropReportsDrops) { + auto g = make_unbalanced_pair(); + auto dag = cost_only_dag(g, 0, 1); + FlowState fs(g); + std::vector> trace, drops; + Flow placed = fs.place_on_dag(0, 1, dag, 100.0, FlowPlacement::EqualBalancedLossy, &trace, &drops); + EXPECT_NEAR(placed, 60.0, 1e-9) << "50 on the 100-unit edge, 10 on the 10-unit edge"; + EXPECT_NEAR(fs.edge_flow_view()[0], 10.0, 1e-9); + EXPECT_NEAR(fs.edge_flow_view()[1], 50.0, 1e-9); + ASSERT_EQ(drops.size(), 1u); + EXPECT_EQ(drops[0].first, 0); + EXPECT_NEAR(drops[0].second, 40.0, 1e-9); + double traced = 0.0; for (auto const& pr : trace) traced += pr.second; + EXPECT_NEAR(traced, 60.0, 1e-9) << "the trace records carried volume only"; + + // A later demand still hashes half onto the saturated member and loses it. + drops.clear(); + EXPECT_NEAR(fs.place_on_dag(0, 1, dag, 10.0, FlowPlacement::EqualBalancedLossy, nullptr, &drops), 5.0, 1e-9); + ASSERT_EQ(drops.size(), 1u); + EXPECT_NEAR(drops[0].second, 5.0, 1e-9); +} + +TEST(FlowState, EqualBalancedLossy_DeficitPropagatesDownstream) { + // 0 -> 1 over edges cap 10 / 100 (cost 1 each), then 1 -> 2 cap 30 (cost 1). + std::int32_t src[3] = {0, 0, 1}; + std::int32_t dst[3] = {1, 1, 2}; + double cap[3] = {10.0, 100.0, 30.0}; + std::int64_t cost[3] = {1, 1, 1}; + auto g = StrictMultiDiGraph::from_arrays(3, + std::span(src, 3), std::span(dst, 3), std::span(cap, 3), std::span(cost, 3)); + auto dag = cost_only_dag(g, 0, 2); + FlowState fs(g); + std::vector> drops; + Flow placed = fs.place_on_dag(0, 2, dag, 100.0, FlowPlacement::EqualBalancedLossy, nullptr, &drops); + // Node 1 receives 60 (40 dropped upstream) and forwards 30 (30 dropped on 1->2). + EXPECT_NEAR(placed, 30.0, 1e-9); + EXPECT_NEAR(fs.edge_flow_view()[2], 30.0, 1e-9); + double total_dropped = 0.0; for (auto const& pr : drops) total_dropped += pr.second; + EXPECT_NEAR(total_dropped, 70.0, 1e-9); +} + +TEST(FlowState, EqualBalancedLossy_InfiniteRequestFillsWithoutDrops) { + auto g = make_unbalanced_pair(); + auto dag = cost_only_dag(g, 0, 1); + FlowState fs(g); + std::vector> drops; + Flow placed = fs.place_on_dag(0, 1, dag, std::numeric_limits::infinity(), + FlowPlacement::EqualBalancedLossy, nullptr, &drops); + EXPECT_NEAR(placed, 110.0, 1e-9); + EXPECT_TRUE(drops.empty()) << "drops are undefined for an infinite offered load"; +} + +TEST(FlowGraph, LossyPlacementIsReversibleAndDropsStayOutOfLedger) { + auto g = make_unbalanced_pair(); + auto dag = cost_only_dag(g, 0, 1); + FlowGraph fg(g); + FlowIndex idx{0, 1, 0, 7}; + std::vector> drops; + EXPECT_NEAR(fg.place(idx, 0, 1, dag, 100.0, FlowPlacement::EqualBalancedLossy, &drops), 60.0, 1e-9); + ASSERT_EQ(drops.size(), 1u); + double ledger = 0.0; for (auto const& pr : fg.get_flow_edges(idx)) ledger += pr.second; + EXPECT_NEAR(ledger, 60.0, 1e-9) << "only carried volume is in the ledger"; + fg.remove(idx); + EXPECT_NEAR(fg.residual_view()[0], 10.0, 1e-9); + EXPECT_NEAR(fg.residual_view()[1], 100.0, 1e-9); + // Non-lossy placements never touch the collector. + drops.clear(); + EXPECT_NEAR(fg.place(idx, 0, 1, dag, 20.0, FlowPlacement::EqualBalanced, &drops), 20.0, 1e-9); + EXPECT_TRUE(drops.empty()); +} diff --git a/tests/cpp/shortest_paths_tests.cpp b/tests/cpp/shortest_paths_tests.cpp index 261f964..5809fb1 100644 --- a/tests/cpp/shortest_paths_tests.cpp +++ b/tests/cpp/shortest_paths_tests.cpp @@ -341,3 +341,113 @@ TEST(ShortestPaths, ZeroCostEdges_PredDAGIsAcyclic) { ASSERT_EQ(paths.size(), 1u); EXPECT_EQ(paths[0].size(), 4u); // 0 -> 1 -> 2 -> 3 } + +// --------------------------------------------------------------------------- +// shortest_paths_to: reverse SPF toward one destination, with forced fan-out. +// --------------------------------------------------------------------------- +#include "netgraph/core/flow_state.hpp" + +namespace { +// Nodes: 0=P (pseudo source), 1=S1, 2=S2, 3=M, 4=T. +// Edges: 0: P->S1 cost 0 cap 1e6; 1: P->S2 cost 0 cap 1e6; +// 2: S1->T cost 1 cap 100; 3: S2->M cost 1 cap 20; 4: M->T cost 1 cap 60. +// S1 is one hop from T, S2 two hops, so an SPF from P keeps only S1. +StrictMultiDiGraph make_fanout_graph() { + std::int32_t src[5] = {0, 0, 1, 2, 3}; + std::int32_t dst[5] = {1, 2, 4, 3, 4}; + double cap[5] = {1e6, 1e6, 100.0, 20.0, 60.0}; + std::int64_t cost[5] = {0, 0, 1, 1, 1}; + return StrictMultiDiGraph::from_arrays(5, + std::span(src, 5), std::span(dst, 5), std::span(cap, 5), std::span(cost, 5)); +} +EdgeSelection cost_only_sel() { + EdgeSelection sel; sel.multi_edge = true; sel.require_capacity = false; sel.tie_break = EdgeTieBreak::Deterministic; + return sel; +} +} // namespace + +TEST(ShortestPathsTo, DistancesMatchForwardDistancesToDst) { + auto g = make_grid_graph(3, 4); + const NodeId t = g.num_nodes() - 1; + auto [dist_to, dag] = shortest_paths_to(g, t, /*multipath=*/true, cost_only_sel()); + expect_pred_dag_valid(dag, g.num_nodes()); + for (NodeId s = 0; s < g.num_nodes(); ++s) { + auto [dist_from_s, fwd] = shortest_paths(g, s, t, /*multipath=*/true, cost_only_sel()); + EXPECT_EQ(dist_to[static_cast(s)], dist_from_s[static_cast(t)]) << "node " << s; + } + // Every DAG entry u -> v via e lies on a shortest u -> t walk. + const auto esrc = g.edge_src_view(); const auto edst = g.edge_dst_view(); const auto cost = g.cost_view(); + for (NodeId v = 0; v < g.num_nodes(); ++v) { + for (auto i = dag.parent_offsets[static_cast(v)]; i < dag.parent_offsets[static_cast(v)+1]; ++i) { + const auto e = static_cast(dag.via_edges[static_cast(i)]); + const NodeId u = dag.parents[static_cast(i)]; + EXPECT_EQ(esrc[e], u); EXPECT_EQ(edst[e], v); + EXPECT_EQ(dist_to[static_cast(u)], cost[e] + dist_to[static_cast(v)]); + } + } +} + +TEST(ShortestPathsTo, ForwardSpfFromPseudoSourceKeepsOnlyNearestSource) { + auto g = make_fanout_graph(); + auto [dist, dag] = shortest_paths(g, 0, 4, /*multipath=*/true, cost_only_sel()); + // T (node 4) is reached only via S1 -> T (edge 2): S2's branch costs 2 and is + // not on a shortest P -> T path, so a placement from P never uses S2. + ASSERT_EQ(dag.parent_offsets[5] - dag.parent_offsets[4], 1); + EXPECT_EQ(dag.via_edges[static_cast(dag.parent_offsets[4])], 2); +} + +TEST(ShortestPathsTo, FanoutEdgesForceEverySourceIntoTheDag) { + auto g = make_fanout_graph(); + EdgeId fan[2] = {0, 1}; + auto [dist, dag] = shortest_paths_to(g, 4, /*multipath=*/true, cost_only_sel(), {}, {}, {}, std::span(fan, 2)); + expect_pred_dag_valid(dag, g.num_nodes()); + EXPECT_EQ(dist[1], 1); EXPECT_EQ(dist[2], 2); EXPECT_EQ(dist[0], 1) << "pseudo source keeps the SPF distance via S1"; + // Both S1 and S2 now have P as parent via their attachment edge. + ASSERT_EQ(dag.parent_offsets[2] - dag.parent_offsets[1], 1); EXPECT_EQ(dag.via_edges[static_cast(dag.parent_offsets[1])], 0); + ASSERT_EQ(dag.parent_offsets[3] - dag.parent_offsets[2], 1); EXPECT_EQ(dag.via_edges[static_cast(dag.parent_offsets[2])], 1); + // Entry the SPF already recorded (P->S1) is not duplicated. + int p_entries = 0; + for (auto v : dag.parents) if (v == 0) ++p_entries; + EXPECT_EQ(p_entries, 2); + + // Lossless equal-balanced admission over the fan-out: shares 50/50 of 100; + // S2's branch admits 20 of 50 (link 3) so the whole demand scales to 0.4. + FlowState fs(g); + EXPECT_NEAR(fs.place_on_dag(0, 4, dag, 100.0, FlowPlacement::EqualBalancedFixed), 40.0, 1e-9); + EXPECT_NEAR(fs.edge_flow_view()[2], 20.0, 1e-9); + EXPECT_NEAR(fs.edge_flow_view()[3], 20.0, 1e-9); +} + +TEST(ShortestPathsTo, FanoutSkipsUnreachableAndMaskedHeads) { + auto g = make_fanout_graph(); + EdgeId fan[2] = {0, 1}; + auto edge_mask = make_bool_mask(static_cast(g.num_edges()), true); + edge_mask[3] = false; // S2 -> M down: S2 cannot reach T + auto [dist, dag] = shortest_paths_to(g, 4, true, cost_only_sel(), {}, {}, + std::span(edge_mask.get(), static_cast(g.num_edges())), + std::span(fan, 2)); + EXPECT_EQ(dist[2], std::numeric_limits::max()); + EXPECT_EQ(dag.parent_offsets[3] - dag.parent_offsets[2], 0) << "no fan-out entry to an unreachable source"; + EXPECT_EQ(dag.parent_offsets[2] - dag.parent_offsets[1], 1); +} + +TEST(ShortestPathsTo, FanoutFromInteriorNodeIsRejected) { + auto g = make_fanout_graph(); + EdgeId fan[1] = {4}; // M -> T, but M already has the incoming entry S2 -> M + EXPECT_THROW((void)shortest_paths_to(g, 4, true, cost_only_sel(), {}, {}, {}, std::span(fan, 1)), + std::invalid_argument); + EdgeId bad[1] = {99}; + EXPECT_THROW((void)shortest_paths_to(g, 4, true, cost_only_sel(), {}, {}, {}, std::span(bad, 1)), + std::invalid_argument); +} + +TEST(ShortestPathsTo, SinglePathModeKeepsOneSuccessorPerNode) { + auto g = make_n_disjoint_paths(3, 10.0); + const NodeId t = g.num_nodes() - 1; + auto [dist, dag] = shortest_paths_to(g, t, /*multipath=*/false, cost_only_sel()); + expect_pred_dag_valid(dag, g.num_nodes()); + // Count DAG entries leaving node 0: exactly one successor. + int leaving_src = 0; + for (auto p : dag.parents) if (p == 0) ++leaving_src; + EXPECT_EQ(leaving_src, 1); +} diff --git a/tests/py/test_equal_balanced_modes.py b/tests/py/test_equal_balanced_modes.py new file mode 100644 index 0000000..6eff362 --- /dev/null +++ b/tests/py/test_equal_balanced_modes.py @@ -0,0 +1,237 @@ +"""EQUAL_BALANCED_FIXED and EQUAL_BALANCED_LOSSY placement semantics. + +Both model a hop-by-hop forwarding table that does not react to load: the +split set is every shortest-path edge with capacity. FIXED admits losslessly +(a saturated member blocks admission), LOSSY forwards best-effort (each member +carries what it can and the rest is dropped). +""" + +from __future__ import annotations + +import numpy as np +import pytest + +import netgraph_core as ngc + + +def _graph(src, dst, cap, cost, n): + return ngc.StrictMultiDiGraph.from_arrays( + num_nodes=n, + src=np.array(src, dtype=np.int32), + dst=np.array(dst, dtype=np.int32), + capacity=np.array(cap, dtype=np.float64), + cost=np.array(cost, dtype=np.int64), + ext_edge_ids=np.arange(len(src), dtype=np.int64), + ) + + +@pytest.fixture +def pair(): + """0 -> 1 over two equal-cost parallel edges: edge 0 cap 10, edge 1 cap 100.""" + g = _graph([0, 0], [1, 1], [10.0, 100.0], [1, 1], 2) + algs = ngc.Algorithms(ngc.Backend.cpu()) + handle = algs.build_graph(g) + sel = ngc.EdgeSelection( + multi_edge=True, + require_capacity=False, + tie_break=ngc.EdgeTieBreak.DETERMINISTIC, + ) + _, dag = algs.spf( + handle, src=0, dst=None, selection=sel, multipath=True, dtype="float64" + ) + return g, dag + + +def test_enum_exposes_new_modes(): + assert ngc.FlowPlacement.EQUAL_BALANCED_FIXED != ngc.FlowPlacement.EQUAL_BALANCED + assert ( + ngc.FlowPlacement.EQUAL_BALANCED_LOSSY != ngc.FlowPlacement.EQUAL_BALANCED_FIXED + ) + assert set(ngc.FlowPlacement.__members__) >= { + "PROPORTIONAL", + "EQUAL_BALANCED", + "EQUAL_BALANCED_FIXED", + "EQUAL_BALANCED_LOSSY", + } + + +def test_fixed_blocks_after_member_saturates(pair): + g, dag = pair + fg = ngc.FlowGraph(g) + first = fg.place( + ngc.FlowIndex(0, 1, 0, 0), + 0, + 1, + dag, + 20.0, + ngc.FlowPlacement.EQUAL_BALANCED_FIXED, + ) + assert first == pytest.approx(20.0) + second = fg.place( + ngc.FlowIndex(0, 1, 0, 1), + 0, + 1, + dag, + 10.0, + ngc.FlowPlacement.EQUAL_BALANCED_FIXED, + ) + assert second == pytest.approx(0.0) + assert fg.edge_flow_view()[1] == pytest.approx(10.0) + + +def test_equal_balanced_still_progressive(pair): + g, dag = pair + fg = ngc.FlowGraph(g) + assert fg.place( + ngc.FlowIndex(0, 1, 0, 0), 0, 1, dag, 20.0, ngc.FlowPlacement.EQUAL_BALANCED + ) == pytest.approx(20.0) + assert fg.place( + ngc.FlowIndex(0, 1, 0, 1), 0, 1, dag, 10.0, ngc.FlowPlacement.EQUAL_BALANCED + ) == pytest.approx(10.0) + + +def test_lossy_delivers_and_reports_drops(pair): + g, dag = pair + fg = ngc.FlowGraph(g) + placed, drops = fg.place_with_drops( + ngc.FlowIndex(0, 1, 0, 0), + 0, + 1, + dag, + 100.0, + ngc.FlowPlacement.EQUAL_BALANCED_LOSSY, + ) + assert placed == pytest.approx(60.0) + assert drops == [(0, pytest.approx(40.0))] + assert fg.edge_flow_view().tolist() == pytest.approx([10.0, 50.0]) + # A later demand still hashes half onto the saturated member. + placed2, drops2 = fg.place_with_drops( + ngc.FlowIndex(0, 1, 0, 1), + 0, + 1, + dag, + 10.0, + ngc.FlowPlacement.EQUAL_BALANCED_LOSSY, + ) + assert placed2 == pytest.approx(5.0) + assert drops2 == [(0, pytest.approx(5.0))] + + +def test_place_with_drops_is_empty_for_other_modes(pair): + g, dag = pair + fg = ngc.FlowGraph(g) + placed, drops = fg.place_with_drops( + ngc.FlowIndex(0, 1, 0, 0), 0, 1, dag, 100.0, ngc.FlowPlacement.EQUAL_BALANCED + ) + assert placed == pytest.approx(20.0) + assert drops == [] + + +def test_lossy_ledger_holds_carried_volume_only(pair): + g, dag = pair + fg = ngc.FlowGraph(g) + idx = ngc.FlowIndex(0, 1, 0, 3) + placed, _ = fg.place_with_drops( + idx, 0, 1, dag, 100.0, ngc.FlowPlacement.EQUAL_BALANCED_LOSSY + ) + assert sum(a for _, a in fg.get_flow_edges(idx)) == pytest.approx(placed) + fg.remove(idx) + assert fg.residual_view().tolist() == pytest.approx([10.0, 100.0]) + + +def test_cost_only_flow_policy_does_not_reroute_around_saturation(): + """require_capacity=False must route on cost alone, even with an EB per-flow target. + + A -> B direct (cap 10, cost 1); A -> C -> B (cap 100, cost 5 + 5). + """ + g = _graph([0, 0, 2], [1, 2, 1], [10.0, 100.0, 100.0], [1, 5, 5], 3) + algs = ngc.Algorithms(ngc.Backend.cpu()) + handle = algs.build_graph(g) + + cfg = ngc.FlowPolicyConfig() + cfg.path_alg = ngc.PathAlg.SPF + cfg.flow_placement = ngc.FlowPlacement.EQUAL_BALANCED_FIXED + cfg.selection = ngc.EdgeSelection( + multi_edge=True, + require_capacity=False, + tie_break=ngc.EdgeTieBreak.DETERMINISTIC, + ) + cfg.require_capacity = False + cfg.shortest_path = True + cfg.min_flow_count = 1 + cfg.max_flow_count = 1 + + fg = ngc.FlowGraph(g) + # Saturate the direct link with a foreign flow first. + sel = ngc.EdgeSelection( + multi_edge=True, + require_capacity=False, + tie_break=ngc.EdgeTieBreak.DETERMINISTIC, + ) + _, dag = algs.spf( + handle, src=0, dst=None, selection=sel, multipath=True, dtype="float64" + ) + assert fg.place( + ngc.FlowIndex(0, 1, 9, 0), + 0, + 1, + dag, + 10.0, + ngc.FlowPlacement.EQUAL_BALANCED_FIXED, + ) == pytest.approx(10.0) + + policy = ngc.FlowPolicy(algs, handle, cfg) + placed, remaining = policy.place_demand(fg, 0, 1, 0, 50.0) + assert placed == pytest.approx(0.0), ( + "cost-only routing must not discover the A->C->B detour" + ) + assert remaining == pytest.approx(50.0) + assert all(float(v[2]) == 1.0 for v in policy.flows.values()), ( + "the flow stays on the cost-1 path" + ) + + +def test_lossy_static_paths_carry_what_fits_without_equalizing(): + """Pinned routes under EQUAL_BALANCED_LOSSY: each LSP is offered its share + and delivers what fits; the equalizing rebalance of EQUAL_BALANCED does + not run, so placed is the delivered total. + + A -> B direct cap 10 (edge 0); A -> C -> B cap 100 (edges 1, 2). Demand 50 + over both routes: 25 offered each, 10 + 25 = 35 delivered. + """ + g = _graph([0, 0, 2], [1, 2, 1], [10.0, 100.0, 100.0], [1, 1, 1], 3) + algs = ngc.Algorithms(ngc.Backend.cpu()) + handle = algs.build_graph(g) + bundles = [ngc.PredDAG.from_edges(g, [0]), ngc.PredDAG.from_edges(g, [1, 2])] + + def policy(placement): + cfg = ngc.FlowPolicyConfig() + cfg.path_alg = ngc.PathAlg.SPF + cfg.flow_placement = placement + cfg.selection = ngc.EdgeSelection( + multi_edge=True, + require_capacity=False, + tie_break=ngc.EdgeTieBreak.DETERMINISTIC, + ) + cfg.require_capacity = False + cfg.min_flow_count = 1 + cfg.max_flow_count = 2 + p = ngc.FlowPolicy(algs, handle, cfg) + p.set_static_paths(0, 1, bundles) + return p + + fg = ngc.FlowGraph(g) + placed, remaining = policy(ngc.FlowPlacement.EQUAL_BALANCED_LOSSY).place_demand( + fg, 0, 1, 0, 50.0 + ) + assert placed == pytest.approx(35.0) + assert remaining == pytest.approx(15.0) + assert fg.edge_flow_view().tolist() == pytest.approx([10.0, 25.0, 25.0]) + + fg = ngc.FlowGraph(g) + placed, _ = policy(ngc.FlowPlacement.EQUAL_BALANCED_FIXED).place_demand( + fg, 0, 1, 0, 50.0 + ) + assert placed == pytest.approx(20.0, abs=1e-3), ( + "lossless: equal carried share, bottleneck 10" + ) diff --git a/tests/py/test_spf_to.py b/tests/py/test_spf_to.py new file mode 100644 index 0000000..4b294cd --- /dev/null +++ b/tests/py/test_spf_to.py @@ -0,0 +1,114 @@ +"""spf_to: reverse SPF toward one destination, with forced fan-out edges.""" + +from __future__ import annotations + +import numpy as np +import pytest + +import netgraph_core as ngc + + +def _graph(src, dst, cap, cost, n): + return ngc.StrictMultiDiGraph.from_arrays( + num_nodes=n, + src=np.array(src, dtype=np.int32), + dst=np.array(dst, dtype=np.int32), + capacity=np.array(cap, dtype=np.float64), + cost=np.array(cost, dtype=np.int64), + ext_edge_ids=np.arange(len(src), dtype=np.int64), + ) + + +SEL = ngc.EdgeSelection( + multi_edge=True, require_capacity=False, tie_break=ngc.EdgeTieBreak.DETERMINISTIC +) + + +@pytest.fixture +def fanout_graph(): + """P(0) -> S1(1) cost 0, P -> S2(2) cost 0; S1 -> T(4) cap 100; S2 -> M(3) cap 20; M -> T cap 60.""" + g = _graph( + [0, 0, 1, 2, 3], + [1, 2, 4, 3, 4], + [1e6, 1e6, 100.0, 20.0, 60.0], + [0, 0, 1, 1, 1], + 5, + ) + algs = ngc.Algorithms(ngc.Backend.cpu()) + return g, algs, algs.build_graph(g) + + +def test_distances_to_dst_match_forward_spf(fanout_graph): + g, algs, h = fanout_graph + dist_to, dag = algs.spf_to(h, 4, selection=SEL) + for s in range(g.num_nodes()): + d, _ = algs.spf(h, s, 4, selection=SEL) + assert dist_to[s] == d[4] + assert dag.parent_offsets.shape == (g.num_nodes() + 1,) + + +def test_fanout_forces_all_sources_and_admission_scales_globally(fanout_graph): + g, algs, h = fanout_graph + dist, dag = algs.spf_to(h, 4, selection=SEL, fanout_edges=[0, 1]) + assert dist[1] == 1.0 and dist[2] == 2.0 + # S2 has P as parent although P->S2 is not on a shortest P->T path. + assert list(dag.via_edges[dag.parent_offsets[2] : dag.parent_offsets[3]]) == [1] + fg = ngc.FlowGraph(g) + placed = fg.place( + ngc.FlowIndex(0, 4, 0, 0), + 0, + 4, + dag, + 100.0, + ngc.FlowPlacement.EQUAL_BALANCED_FIXED, + ) + # Even 50/50 split; S2's 20-unit link admits 20 of 50, so the demand scales to 0.4. + assert placed == pytest.approx(40.0) + assert fg.edge_flow_view()[2] == pytest.approx(20.0) + + +def test_fanout_lossy_delivers_what_each_branch_carries(fanout_graph): + g, algs, h = fanout_graph + _, dag = algs.spf_to(h, 4, selection=SEL, fanout_edges=[0, 1]) + fg = ngc.FlowGraph(g) + placed, drops = fg.place_with_drops( + ngc.FlowIndex(0, 4, 0, 0), + 0, + 4, + dag, + 100.0, + ngc.FlowPlacement.EQUAL_BALANCED_LOSSY, + ) + assert placed == pytest.approx(70.0) # 50 via S1 + 20 via S2 + assert drops == [(3, pytest.approx(30.0))] + + +def test_without_fanout_the_pseudo_source_reaches_only_the_nearest_source(fanout_graph): + g, algs, h = fanout_graph + _, dag = algs.spf(h, 0, 4, selection=SEL) + # T is reached only via S1 -> T (edge 2): S2's branch costs 2 and never + # carries flow placed from P. + assert list(dag.via_edges[dag.parent_offsets[4] : dag.parent_offsets[5]]) == [2] + + +def test_fanout_respects_masks_and_reachability(fanout_graph): + g, algs, h = fanout_graph + edge_mask = np.ones(g.num_edges(), dtype=bool) + edge_mask[3] = False # S2 -> M down + dist, dag = algs.spf_to( + h, 4, selection=SEL, edge_mask=edge_mask, fanout_edges=[0, 1] + ) + assert np.isinf(dist[2]) + assert dag.parent_offsets[3] - dag.parent_offsets[2] == 0 + + +def test_fanout_validation(fanout_graph): + g, algs, h = fanout_graph + with pytest.raises(ValueError): + algs.spf_to( + h, 4, selection=SEL, fanout_edges=[4] + ) # M has an incoming DAG entry + with pytest.raises(ValueError): + algs.spf_to(h, 4, selection=SEL, fanout_edges=[99]) + with pytest.raises(ValueError): + algs.spf_to(h, 42, selection=SEL)