From 4c188d5aba5ac92ec475b599bccc219ba6a840d7 Mon Sep 17 00:00:00 2001 From: Andrey Golovanov Date: Mon, 24 Aug 2026 01:03:37 +0100 Subject: [PATCH 1/2] Support pinning demands to explicit routes (static paths) A demand can name the routes its traffic must follow instead of letting the flow policy choose them, modelling MPLS-style LSPs: static_paths: - ["A", "B", "C"] # node names - links: ["A|D|0", "D|C|0"] # or link ids, for a specific parallel link One flow is created per route, in the order listed. A route broken by a failure carries nothing rather than rerouting, which is what distinguishes a pinned route from ordinary routing. Built on netgraph-core 0.8.0's FlowPolicy.set_static_paths and PredDAG.from_edges, so the dependency floor moves to 0.8.0. That release also adds a max-flow completion phase, so max_flow can return more than it did on 0.7.x; the design reference is updated to describe reverse residual arcs returning placed flow rather than serving only min-cut reachability. Because routes run between two concrete nodes, a pinned demand must use mode: pairwise with selectors matching exactly one source and one target. Combine mode routes through pseudo endpoints that no operator-supplied route can start from, so it is rejected with that explanation. Resolution details: - A node hop takes the cheapest enabled link between the pair, ties broken by link id, so the choice is stable across identical scenario builds. Disabled links are never chosen, and naming one explicitly is an error: a route pinned to a disabled link could never carry traffic. - Hops resolve through the graph's adjacency rows rather than a scan of every edge, so cost is proportional to node degree rather than graph size. - Bundles depend only on the static graph, not on per-iteration masks, so they are resolved once per analysis context and reused across Monte Carlo iterations and MSD probes. Known limitation: two demands pinned between the same source, target and priority are rejected, because netgraph-core assigns flow ids per policy starting at zero and they would collide. List every route on one demand, or separate the demands by priority. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 3 +- docs/reference/api-full.md | 86 +++++- docs/reference/design.md | 4 +- docs/reference/dsl.md | 56 ++++ ngraph/analysis/demand.py | 39 ++- ngraph/analysis/functions.py | 36 ++- ngraph/analysis/placement.py | 39 ++- ngraph/analysis/static_paths.py | 213 +++++++++++++++ ngraph/model/demand/builder.py | 54 +++- ngraph/model/demand/spec.py | 48 +++- ngraph/model/flow/policy_config.py | 38 ++- ngraph/schemas/scenario.json | 53 +++- pyproject.toml | 2 +- tests/analysis/test_static_paths.py | 407 ++++++++++++++++++++++++++++ 14 files changed, 1034 insertions(+), 44 deletions(-) create mode 100644 ngraph/analysis/static_paths.py create mode 100644 tests/analysis/test_static_paths.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c4eff13..6d60f26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -79,13 +79,14 @@ Other fixes. ### Added +- Demands can be pinned to explicit routes with `static_paths`, modelling MPLS-style LSPs: one flow per route, and a route broken by a failure carries nothing instead of rerouting. A route is a list of node names or link ids (link ids pick a specific one of several parallel links). Requires `mode: pairwise` with selectors matching exactly one source and one target - `AnalysisContext.sensitivity_with_flow`: max flow and edge sensitivity per group pair in a single pass, used by `sensitivity_analysis` and the sensitivity Monte Carlo path (results unchanged) - `AnalysisContext.build_node_mask`/`build_edge_mask` as public methods, for analysis functions calling Core primitives directly - `Scenario.run(step_hook=...)`: a callable returning a context manager entered around each step, used by the CLI `--profile` path - `FailurePolicy.apply_failures_typed`, returning scope-typed failure sets (`apply_failures` remains as a merged-list wrapper), plus `prepare_weights`, `build_risk_group_index`, and the `prepared_rg_index` parameter on `apply_failures` for reuse across Monte Carlo iterations - `build_demand_placement_inputs`, exported from `ngraph.analysis`; `demand_placement_analysis` accepts precomputed `expansion=` and `resolved_ids=` - `TrafficDemand.to_dict()`, `Mode.from_string` in `ngraph.types`, and `link_path_key` in `ngraph.model.selectors` (the canonical "source|target" key used when path-matching links) -- Minimum `netgraph-core` raised to 0.7.0, the API family this release targets; the previous 0.3.0 floor predated APIs now in use +- Minimum `netgraph-core` raised to 0.8.0, which introduces the `FlowPolicy.set_static_paths` and `PredDAG.from_edges` APIs that pinned routes are built on, and whose max-flow completion phase makes `max_flow` return a true maximum (values can increase against 0.7.x) ### Removed diff --git a/docs/reference/api-full.md b/docs/reference/api-full.md index e08a475..c928fad 100644 --- a/docs/reference/api-full.md +++ b/docs/reference/api-full.md @@ -12,9 +12,9 @@ Quick links: - [CLI Reference](cli.md) - [DSL Reference](dsl.md) -Generated from source code on: August 21, 2026 at 00:49 UTC +Generated from source code on: August 24, 2026 at 00:43 UTC -Modules auto-discovered: 53 +Modules auto-discovered: 54 --- @@ -488,7 +488,33 @@ Attributes: Traffic demand specification. Defines `TrafficDemand`, a user-facing specification used by demand expansion -and placement. Routing behavior is selected via an optional `FlowPolicyPreset`. +and placement. Routing behavior is selected via an optional `FlowPolicyPreset`, +or pinned to explicit routes with `StaticPath`. + +### StaticPath + +One explicit route a demand can be pinned to (an MPLS-style LSP). + +Give exactly one of `nodes` or `links`: + +- `nodes`: the node names the route visits, source first and target last. + + Each consecutive pair must be adjacent. When several parallel links + connect a pair, the cheapest is used (ties broken by link id); name the + link explicitly to choose a different one. + +- `links`: the link ids the route traverses, in order. Unambiguous when + + parallel links exist. A link may be traversed in either direction. + +Attributes: + nodes: Node names along the route, or empty when `links` is given. + links: Link ids along the route, or empty when `nodes` is given. + +**Attributes:** + +- `nodes` (Tuple) = () +- `links` (Tuple) = () ### TrafficDemand @@ -503,6 +529,10 @@ Attributes: group_mode: How grouped nodes produce demands ("flatten", "per_group", "group_pairwise"). flow_policy: Policy preset for routing. + static_paths: Explicit routes to pin this demand to. When set, the + demand is placed only on these routes: one flow per route, and a + route broken by a failure carries nothing rather than rerouting. + Requires selectors matching exactly one source and one target. attrs: Arbitrary user metadata. id: Unique identifier. Auto-generated if empty. @@ -515,6 +545,7 @@ Attributes: - `mode` (str) = combine - `group_mode` (str) = flatten - `flow_policy` (Union) +- `static_paths` (Tuple) = () - `attrs` (Dict) = {} - `id` (str) @@ -888,7 +919,7 @@ Enumerates common flow policy presets for traffic routing. These presets map to specific combinations of path algorithms, flow placement strategies, and edge selection modes provided by NetGraph-Core. -### create_flow_policy(algorithms: 'netgraph_core.Algorithms', graph: 'netgraph_core.Graph', preset: 'FlowPolicyPreset', node_mask=None, edge_mask=None) -> 'netgraph_core.FlowPolicy' +### create_flow_policy(algorithms: 'netgraph_core.Algorithms', graph: 'netgraph_core.Graph', preset: 'FlowPolicyPreset', node_mask=None, edge_mask=None, static_path_count: 'Optional[int]' = None) -> 'netgraph_core.FlowPolicy' Create a FlowPolicy instance from a preset configuration. @@ -899,6 +930,9 @@ Args: flow-count bounds to apply. node_mask: Optional numpy bool array for node exclusions (True = include). edge_mask: Optional numpy bool array for edge exclusions (True = include). + static_path_count: Number of routes the caller will pin with + `FlowPolicy.set_static_paths`. Sets the flow count to match, since + a pinned policy creates one flow per route and never grows. Returns: netgraph_core.FlowPolicy: Configured policy instance. @@ -2922,6 +2956,8 @@ Attributes: volume: Traffic volume to place. priority: Priority class (lower is higher priority). policy_preset: FlowPolicy configuration preset. + static_paths: Routes this demand is pinned to, empty when it is + routed by the policy. **Attributes:** @@ -2930,6 +2966,7 @@ Attributes: - `volume` (float) - `priority` (int) - `policy_preset` (FlowPolicyPreset) +- `static_paths` (Tuple[StaticPath, ...]) = () ### expand_demands(network: 'Network', traffic_demands: 'List[TrafficDemand]', default_policy_preset: 'FlowPolicyPreset' = ) -> 'DemandExpansion' @@ -2957,9 +2994,11 @@ Returns: DemandExpansion with demands and augmentations. Raises: - ValueError: If no demands could be expanded, or if two demands share - an id (pseudo node names embed the id, so duplicates would merge - distinct demands' attachment edges into one endpoint). + ValueError: If no demands could be expanded, if two demands share an + id (pseudo node names embed the id, so duplicates would merge + distinct demands' attachment edges into one endpoint), or if a + demand with `static_paths` does not resolve to exactly one + source/target pair. --- @@ -3262,6 +3301,39 @@ Raises: --- +## ngraph.analysis.static_paths + +Resolution of explicit routes into Core path bundles. + +Turns the `StaticPath` entries on a demand into the `PredDAG` bundles that +`FlowPolicy.set_static_paths` pins traffic to. A bundle is a single simple +path: one edge per hop, so a route that names adjacent nodes with parallel +links between them picks one of those links (see `StaticPath`). + +### build_static_path_bundles(ctx: "'AnalysisContext'", paths: 'Sequence[StaticPath]', src_name: 'str', dst_name: 'str') -> 'List[netgraph_core.PredDAG]' + +Build the Core path bundles a demand is pinned to. + +Args: + ctx: Context holding the built graph; routes resolve against it. + paths: Routes to pin, in the order flows should be created. + src_name: Node every route must start at. + dst_name: Node every route must end at. + +Returns: + One `PredDAG` per route, in the given order. Results are cached per + context, so repeated calls during a Monte Carlo run resolve once. + +Raises: + ValueError: If an endpoint is absent from the context graph, if a + route names an unknown node or a disabled/unknown link, has a hop + whose nodes are joined only by disabled links, has a hop with + no link, traverses a link that does not leave the node it has + reached, does not run from `src_name` to `dst_name`, or revisits + a node (a pinned route must be a simple path). + +--- + ## ngraph.lib.nx NetworkX graph conversion utilities. diff --git a/docs/reference/design.md b/docs/reference/design.md index ba3b5ef..12c58d7 100644 --- a/docs/reference/design.md +++ b/docs/reference/design.md @@ -491,11 +491,11 @@ See "Routing Semantics: IP/IGP vs SDN/TE" section for detailed explanation. The residual network is maintained via `FlowState`, which tracks per-edge flow and computes residual capacities on demand. For each edge u→v: - Forward residual capacity: `capacity(u,v) - flow(u,v)` -- Reverse residual capacity: `flow(u,v)` (used for residual reachability when computing the min-cut and reachable set) +- Reverse residual capacity: `flow(u,v)` (traversed to return previously placed flow during the completion phase, and for residual reachability when computing the min-cut and reachable set) SPF operates over the residual graph by requesting edges with `require_capacity=true`, which filters to edges with positive residual capacity. The `FlowState` provides a residual capacity view without graph mutation. -Note: Reverse residual arcs are distinct from physical reverse edges added via `add_reverse=True` during graph construction. Physical reverse edges model bidirectional links with independent capacity; reverse residual arcs are bookkeeping over a single edge's flow. The augmenting SPF search traverses forward residual edges only — placed flow is never cancelled across tiers; reverse residual arcs are traversed only for reachability when computing the min-cut and reachable set, while Dinic-style reverse edges allow redistribution within a single tier's placement. +Note: Reverse residual arcs are distinct from physical reverse edges added via `add_reverse=True` during graph construction. Physical reverse edges model bidirectional links with independent capacity; reverse residual arcs are bookkeeping over a single edge's flow. The cost-tier SPF loop traverses forward residual edges only, so it cannot cancel an earlier placement and may stop below the true maximum. A completion phase then runs BFS augmentation over the full residual graph, traversing arcs backwards to return previously placed flow, which makes the result a true maximum flow whose min-cut matches it. The completion phase applies only to max-flow semantics — `PROPORTIONAL` placement with `require_capacity=True` and `shortest_path=False`; `EQUAL_BALANCED` (ECMP admission) and `require_capacity=False` (fixed-cost IP routing) are placement models rather than max-flow computations and keep the tier-loop result. Dinic-style reverse edges additionally allow redistribution within a single tier's placement. The core loop finds augmenting paths using the cost-aware SPF described above: diff --git a/docs/reference/dsl.md b/docs/reference/dsl.md index 19fb5b4..6f61579 100644 --- a/docs/reference/dsl.md +++ b/docs/reference/dsl.md @@ -1249,11 +1249,67 @@ demands: | `mode` | string | Node pairing mode: `combine` or `pairwise` (default: `combine`) | | `group_mode` | string | How grouped nodes produce demands (default: `flatten`) | | `flow_policy` | string or integer | Routing policy preset name (case-insensitive, or its integer value); inline policy mappings fail schema validation at scenario load | +| `static_paths` | array | Explicit routes to pin the demand to; see below | | `attrs` | object | Arbitrary metadata | | `expand` | object | Variable expansion block | Each demand receives an auto-generated unique `id`; an explicit `id` key is not accepted in scenario YAML. Duplicate ids can arise only when demands are constructed programmatically, and demand expansion rejects them with `ValueError`. +### Static Paths + +`static_paths` pins a demand to routes you choose, instead of letting the flow +policy pick them. This models MPLS-style LSPs: one flow per route, and a route +broken by a failure carries nothing rather than rerouting. + +```yaml +demands: + default: + - source: "^A$" + target: "^C$" + volume: 10 + mode: pairwise + flow_policy: SHORTEST_PATHS_WCMP + static_paths: + - ["A", "B", "C"] # node names, source first + - links: ["A|D|0", "D|C|0"] # or link ids, in traversal order +``` + +Each route is either a list of node names or a mapping with `nodes` or +`links`. Name nodes for readability; name links when parallel links connect +the same pair and you need a specific one. + +A route is a strict explicit route: every hop is one link. Where parallel +links connect a pair, a node hop takes the cheapest enabled one (ties broken +by link id), so the route carries that single link's capacity and fails when +that link fails — not when the whole bundle does. To model an LSP per parallel +link, list one route per link using the `links` form. Disabled links are never +chosen for a node hop, and naming one in the `links` form is an error. + +Semantics: + +- One flow per route, created in the order listed. How volume divides depends + on the preset: proportional presets such as `SHORTEST_PATHS_WCMP` fill the + routes in the order you list them, so a volume smaller than the first + route's bottleneck never reaches the second; equal-balanced presets give + every surviving route the same share, so total placement is limited by the + smallest surviving route. Route order is therefore significant under + proportional presets. +- A route whose nodes or links are excluded carries nothing. Traffic does not + move to another route, which is what makes this different from ordinary + routing. +- The policy neither adds routes nor reoptimizes, so preset cost ceilings and + LSP counts do not apply. + +Two demands pinned between the same source, target, and priority are rejected, +because their flows would collide. Put every route on one demand, or separate +the demands by priority. + +Because routes run between two concrete nodes, a demand using `static_paths` +must set `mode: pairwise` and use selectors matching exactly one source and +one target; anything else raises `ValueError` during expansion. Every route +must start at the source and end at the target, visit adjacent nodes, and not +revisit a node. + ### Selector Fields The `source` and `target` fields accept either: diff --git a/ngraph/analysis/demand.py b/ngraph/analysis/demand.py index 3fe2702..9666133 100644 --- a/ngraph/analysis/demand.py +++ b/ngraph/analysis/demand.py @@ -8,11 +8,11 @@ from __future__ import annotations from dataclasses import dataclass, replace -from typing import Dict, List +from typing import Dict, List, Tuple from ngraph.analysis.context import LARGE_CAPACITY, AugmentationEdge from ngraph.dsl.selectors import normalize_selector -from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.demand.spec import StaticPath, TrafficDemand from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.model.network import Network, Node from ngraph.model.selectors import select_nodes @@ -31,6 +31,8 @@ class ExpandedDemand: volume: Traffic volume to place. priority: Priority class (lower is higher priority). policy_preset: FlowPolicy configuration preset. + static_paths: Routes this demand is pinned to, empty when it is + routed by the policy. """ src_name: str @@ -38,6 +40,7 @@ class ExpandedDemand: volume: float priority: int policy_preset: FlowPolicyPreset + static_paths: Tuple[StaticPath, ...] = () @dataclass @@ -144,6 +147,7 @@ def _expand_pairwise( volume=volume_per_pair, priority=td.priority, policy_preset=policy_preset, + static_paths=td.static_paths, ) for src, dst in pairs ] @@ -294,9 +298,11 @@ def expand_demands( DemandExpansion with demands and augmentations. Raises: - ValueError: If no demands could be expanded, or if two demands share - an id (pseudo node names embed the id, so duplicates would merge - distinct demands' attachment edges into one endpoint). + ValueError: If no demands could be expanded, if two demands share an + id (pseudo node names embed the id, so duplicates would merge + distinct demands' attachment edges into one endpoint), or if a + demand with `static_paths` does not resolve to exactly one + source/target pair. """ seen_ids: set[str] = set() for td in traffic_demands: @@ -320,6 +326,11 @@ def expand_demands( dst_groups = select_nodes(network, tgt_sel, default_active_only=True) if not src_groups or not dst_groups: + if td.static_paths: + raise ValueError( + f"Demand '{td.id}' sets static_paths but its selectors match " + "no active source or target node" + ) continue policy_preset = td.flow_policy or default_policy_preset @@ -329,6 +340,24 @@ def expand_demands( td, src_groups, dst_groups, policy_preset ) + if td.static_paths: + # Routes are pinned between two concrete nodes, so the demand has + # to name exactly one pair. Combine mode routes through pseudo + # endpoints, which no operator-supplied route can start from. + if td.mode == "combine": + raise ValueError( + f"Demand '{td.id}' sets static_paths, which pins traffic to " + "routes between two nodes; use mode 'pairwise' instead of " + "'combine'" + ) + if len(demands) != 1: + raise ValueError( + f"Demand '{td.id}' sets static_paths but its selectors " + f"expand to {len(demands)} source/target pairs; static " + "paths require selectors matching exactly one source and " + "one target" + ) + all_demands.extend(demands) all_augmentations.extend(augmentations) diff --git a/ngraph/analysis/functions.py b/ngraph/analysis/functions.py index 87dea1b..eea1055 100644 --- a/ngraph/analysis/functions.py +++ b/ngraph/analysis/functions.py @@ -26,12 +26,43 @@ from ngraph.analysis.demand import DemandExpansion, expand_demands from ngraph.analysis.placement import place_demands from ngraph.model.demand.builder import coerce_flow_policy -from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.demand.spec import StaticPath, TrafficDemand from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.results.flow import FlowEntry, FlowIterationResult, FlowSummary from ngraph.types.base import FlowPlacement, Mode +def _static_paths_from_config(raw: Any) -> tuple[StaticPath, ...]: + """Rebuild pinned routes from their serialized form. + + Accepts every form the scenario DSL accepts, plus `StaticPath` objects + unchanged, so demand configs built in Python behave like ones produced by + `TrafficDemand.to_dict()`. + + Raises: + ValueError: If an entry is not a StaticPath, a list of node names, or + a mapping with exactly one of 'nodes' or 'links'. + """ + if not raw: + return () + paths = [] + for entry in raw: + if isinstance(entry, StaticPath): + paths.append(entry) + elif isinstance(entry, (list, tuple)): + paths.append(StaticPath(nodes=tuple(entry))) + elif isinstance(entry, dict) and set(entry) == {"nodes"}: + paths.append(StaticPath(nodes=tuple(entry["nodes"]))) + elif isinstance(entry, dict) and set(entry) == {"links"}: + paths.append(StaticPath(links=tuple(entry["links"]))) + else: + raise ValueError( + f"Invalid static path {entry!r}: expected a list of node names " + "or a mapping with exactly one of 'nodes' or 'links'" + ) + return tuple(paths) + + def _reconstruct_traffic_demands( demands_config: list[dict[str, Any]], ) -> list[TrafficDemand]: @@ -49,7 +80,7 @@ def _reconstruct_traffic_demands( Args: demands_config: List of demand configurations with fields: source, target, volume, mode, group_mode, flow_policy, - priority, attrs. + priority, static_paths, attrs. Returns: List of TrafficDemand objects with stable IDs. @@ -67,6 +98,7 @@ def _reconstruct_traffic_demands( group_mode=config.get("group_mode", "flatten"), flow_policy=coerce_flow_policy(config.get("flow_policy")), priority=config.get("priority", 0), + static_paths=_static_paths_from_config(config.get("static_paths")), attrs=config.get("attrs") or {}, ) ) diff --git a/ngraph/analysis/placement.py b/ngraph/analysis/placement.py index 81f1f8e..16f1b89 100644 --- a/ngraph/analysis/placement.py +++ b/ngraph/analysis/placement.py @@ -8,6 +8,8 @@ import netgraph_core import numpy as np +from ngraph.analysis.static_paths import build_static_path_bundles +from ngraph.model.demand.spec import StaticPath from ngraph.model.flow.policy_config import FlowPolicyPreset, create_flow_policy if TYPE_CHECKING: @@ -189,7 +191,7 @@ def place_demands( ): total_demand += volume - if demand.policy_preset in CACHEABLE_PRESETS: + if demand.policy_preset in CACHEABLE_PRESETS and not demand.static_paths: placed, cost_dist, used_edges, flow_idx_counter = _place_cached( src_id, dst_id, @@ -208,11 +210,21 @@ def place_demands( else: triple = (src_id, dst_id, demand.priority) if triple in policy_triples: + same_pair = ( + f"source '{demand.src_name}', destination " + f"'{demand.dst_name}', priority {demand.priority}" + ) + if demand.static_paths: + raise ValueError( + f"Two demands pinned to static paths share {same_pair}. " + "Their flow ids would collide and corrupt placement. " + "List every route on a single demand, or give the " + "demands distinct priorities." + ) raise ValueError( - f"Duplicate policy-based demand for source '{demand.src_name}', " - f"destination '{demand.dst_name}', priority {demand.priority}: " - "flow ids would collide and corrupt placement. Merge the " - "demand volumes or use distinct priorities." + f"Duplicate policy-based demand for {same_pair}: flow ids " + "would collide and corrupt placement. Merge the demand " + "volumes or use distinct priorities." ) policy_triples.add(triple) placed, cost_dist, used_edges = _place_with_policy( @@ -227,6 +239,9 @@ def place_demands( edge_mask, include_cost_distribution, include_used_edges, + static_paths=demand.static_paths, + src_name=demand.src_name, + dst_name=demand.dst_name, ) total_placed += placed @@ -382,15 +397,27 @@ def _place_with_policy( edge_mask: np.ndarray, include_cost_distribution: bool, include_used_edges: bool, + static_paths: Sequence[StaticPath] = (), + src_name: str = "", + dst_name: str = "", ) -> tuple[float, dict[float, float], set[str]]: - """Place single demand using FlowPolicy (for non-cacheable presets).""" + """Place a single demand using FlowPolicy. + + Used for non-cacheable presets and for any demand pinned to explicit + routes. With `static_paths` the policy is pinned to those routes: one flow + per route, and a route broken by the masks carries nothing. + """ policy = create_flow_policy( ctx.algorithms, ctx.handle, preset, node_mask=node_mask, edge_mask=edge_mask, + static_path_count=len(static_paths) or None, ) + if static_paths: + bundles = build_static_path_bundles(ctx, static_paths, src_name, dst_name) + policy.set_static_paths(src_id, dst_id, bundles) placed, _ = policy.place_demand(flow_graph, src_id, dst_id, priority, volume) cost_dist: dict[float, float] = {} diff --git a/ngraph/analysis/static_paths.py b/ngraph/analysis/static_paths.py new file mode 100644 index 0000000..b27db7f --- /dev/null +++ b/ngraph/analysis/static_paths.py @@ -0,0 +1,213 @@ +"""Resolution of explicit routes into Core path bundles. + +Turns the `StaticPath` entries on a demand into the `PredDAG` bundles that +`FlowPolicy.set_static_paths` pins traffic to. A bundle is a single simple +path: one edge per hop, so a route that names adjacent nodes with parallel +links between them picks one of those links (see `StaticPath`). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple + +import netgraph_core + +from ngraph.model.demand.spec import StaticPath + +if TYPE_CHECKING: + from ngraph.analysis.context import AnalysisContext + +__all__ = ["build_static_path_bundles"] + +# Bundles depend only on the static graph, not on the per-iteration masks +# (Core prunes them against the masks itself), so they are resolved once per +# context and reused across Monte Carlo iterations and MSD probes. The cache +# lives on the context so it dies with it. +_CACHE_ATTR = "_static_path_bundle_cache" + + +def _disabled_edge_ids(ctx: "AnalysisContext") -> frozenset: + """Core edge ids belonging to administratively disabled links.""" + link_edges = ctx.link_id_to_edge_indices + return frozenset( + int(edge_id) + for link_id in ctx.disabled_link_ids + for edge_id in link_edges.get(link_id, ()) + ) + + +def _hop_edge( + src_id: int, dst_id: int, adjacency, cost, ext_ids, disabled_edges +) -> int: + """Return the edge id to use for one node-to-node hop. + + Picks the cheapest enabled edge from src_id to dst_id, breaking ties by + external edge id so the choice is stable across identical scenario + builds. Disabled links are skipped: pinning a route to one would take the + route down for the whole run even when an enabled parallel link exists. + Walks the source's adjacency row, so cost is proportional to that node's + degree rather than to the size of the graph. + """ + row, col, adj_edge = adjacency + best = -1 + best_key: Tuple[int, int] = (0, 0) + for i in range(int(row[src_id]), int(row[src_id + 1])): + if int(col[i]) != dst_id: + continue + edge_id = int(adj_edge[i]) + if edge_id in disabled_edges: + continue + key = (int(cost[edge_id]), int(ext_ids[edge_id])) + if best < 0 or key < best_key: + best, best_key = edge_id, key + return best + + +def _edges_from_nodes(ctx: "AnalysisContext", path: StaticPath) -> List[int]: + """Resolve a node-sequence route to one Core edge per hop.""" + graph = ctx.multidigraph + adjacency = ( + graph.row_offsets_view(), + graph.col_indices_view(), + graph.adj_edge_index_view(), + ) + cost = graph.cost_view() + ext_ids = graph.ext_edge_ids_view() + disabled_edges = _disabled_edge_ids(ctx) + + edges: List[int] = [] + for u, v in zip(path.nodes, path.nodes[1:], strict=False): + for name in (u, v): + if name not in ctx.node_mapper.node_id_of: + raise ValueError( + f"Static path names unknown node {name!r}; " + f"route was {list(path.nodes)}" + ) + u_id = ctx.node_mapper.to_id(u) + v_id = ctx.node_mapper.to_id(v) + edge_id = _hop_edge(u_id, v_id, adjacency, cost, ext_ids, disabled_edges) + if edge_id < 0: + raise ValueError( + f"Static path hop {u!r} -> {v!r} has no enabled link; " + f"route was {list(path.nodes)}" + ) + edges.append(edge_id) + return edges + + +def _edges_from_links( + ctx: "AnalysisContext", path: StaticPath, src_id: int +) -> List[int]: + """Resolve a link-id route to Core edges, following the traversal order.""" + graph = ctx.multidigraph + edge_src = graph.edge_src_view() + edge_dst = graph.edge_dst_view() + link_edges = ctx.link_id_to_edge_indices + + current = src_id + edges: List[int] = [] + for link_id in path.links: + candidates = link_edges.get(link_id) + if not candidates: + raise ValueError( + f"Static path names unknown link {link_id!r}; " + f"route was {list(path.links)}" + ) + # A link has a forward and (for bidirectional links) a reverse edge; + # pick whichever leaves the node the route has reached. + if link_id in ctx.disabled_link_ids: + raise ValueError( + f"Static path names disabled link {link_id!r}; a route pinned to " + "a disabled link can never carry traffic" + ) + chosen = next( + (int(e) for e in candidates if int(edge_src[int(e)]) == current), None + ) + if chosen is None: + reached = ctx.node_mapper.to_name(current) + raise ValueError( + f"Static path link {link_id!r} does not leave node {reached!r}; " + f"route was {list(path.links)}" + ) + edges.append(chosen) + current = int(edge_dst[chosen]) + return edges + + +def build_static_path_bundles( + ctx: "AnalysisContext", + paths: Sequence[StaticPath], + src_name: str, + dst_name: str, +) -> List[netgraph_core.PredDAG]: + """Build the Core path bundles a demand is pinned to. + + Args: + ctx: Context holding the built graph; routes resolve against it. + paths: Routes to pin, in the order flows should be created. + src_name: Node every route must start at. + dst_name: Node every route must end at. + + Returns: + One `PredDAG` per route, in the given order. Results are cached per + context, so repeated calls during a Monte Carlo run resolve once. + + Raises: + ValueError: If an endpoint is absent from the context graph, if a + route names an unknown node or a disabled/unknown link, has a hop + whose nodes are joined only by disabled links, has a hop with + no link, traverses a link that does not leave the node it has + reached, does not run from `src_name` to `dst_name`, or revisits + a node (a pinned route must be a simple path). + """ + cache_key = (src_name, dst_name, tuple(paths)) + per_ctx: Optional[Dict[tuple, List[netgraph_core.PredDAG]]] = getattr( + ctx, _CACHE_ATTR, None + ) + if per_ctx is None: + per_ctx = {} + setattr(ctx, _CACHE_ATTR, per_ctx) + cached = per_ctx.get(cache_key) + if cached is not None: + return cached + + graph = ctx.multidigraph + edge_src = graph.edge_src_view() + edge_dst = graph.edge_dst_view() + for name in (src_name, dst_name): + if name not in ctx.node_mapper.node_id_of: + raise ValueError( + f"Demand endpoint {name!r} is not in the analysis context graph, " + "so its static paths cannot be resolved" + ) + src_id = ctx.node_mapper.to_id(src_name) + dst_id = ctx.node_mapper.to_id(dst_name) + + bundles: List[netgraph_core.PredDAG] = [] + for path in paths: + if path.nodes: + if path.nodes[0] != src_name or path.nodes[-1] != dst_name: + raise ValueError( + f"Static path must run from {src_name!r} to {dst_name!r}, " + f"but runs from {path.nodes[0]!r} to {path.nodes[-1]!r}" + ) + edges = _edges_from_nodes(ctx, path) + else: + edges = _edges_from_links(ctx, path, src_id) + + if int(edge_src[edges[0]]) != src_id or int(edge_dst[edges[-1]]) != dst_id: + start = ctx.node_mapper.to_name(int(edge_src[edges[0]])) + end = ctx.node_mapper.to_name(int(edge_dst[edges[-1]])) + raise ValueError( + f"Static path must run from {src_name!r} to {dst_name!r}, " + f"but runs from {start!r} to {end!r}" + ) + + try: + bundles.append(netgraph_core.PredDAG.from_edges(graph, edges)) + except ValueError as exc: + route = list(path.nodes or path.links) + raise ValueError(f"Invalid static path {route}: {exc}") from exc + + per_ctx[cache_key] = bundles + return bundles diff --git a/ngraph/model/demand/builder.py b/ngraph/model/demand/builder.py index 2a9594c..f2e2c02 100644 --- a/ngraph/model/demand/builder.py +++ b/ngraph/model/demand/builder.py @@ -5,11 +5,11 @@ from __future__ import annotations -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from ngraph.dsl.expansion import ExpansionSpec, expand_block from ngraph.model.demand.matrix import DemandSet -from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.demand.spec import StaticPath, TrafficDemand from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.utils.yaml_utils import normalize_yaml_dict_keys @@ -92,9 +92,59 @@ def _build_demand(d: Dict[str, Any], set_name: str) -> TrafficDemand: if "flow_policy" in d: td_kwargs["flow_policy"] = coerce_flow_policy(d["flow_policy"]) + if "static_paths" in d: + td_kwargs["static_paths"] = _build_static_paths(d["static_paths"], set_name) + return TrafficDemand(**td_kwargs) +def _build_static_paths(raw: Any, set_name: str) -> Tuple[StaticPath, ...]: + """Build the pinned routes of one demand from its YAML form. + + Each entry is either a list of node names or a mapping with `nodes` or + `links`. + + Raises: + ValueError: If the block is not a non-empty list, or an entry is not + one of the accepted forms. + """ + if not isinstance(raw, list) or not raw: + raise ValueError( + f"'static_paths' in set '{set_name}' must be a non-empty list of routes" + ) + + def _names(values: Any, kind: str) -> Tuple[str, ...]: + if not isinstance(values, (list, tuple)) or not all( + isinstance(v, str) for v in values + ): + # Variable expansion substitutes native types, so a ${var} bound to + # a list reaches here despite the schema constraining route entries. + raise ValueError( + f"'static_paths' in set '{set_name}': {kind} must all be " + f"strings, got {values!r}" + ) + return tuple(values) + + paths: list[StaticPath] = [] + for entry in raw: + if isinstance(entry, list): + paths.append(StaticPath(nodes=_names(entry, "node names"))) + continue + if isinstance(entry, dict): + keys = set(entry) + if keys == {"nodes"}: + paths.append(StaticPath(nodes=_names(entry["nodes"], "node names"))) + continue + if keys == {"links"}: + paths.append(StaticPath(links=_names(entry["links"], "link ids"))) + continue + raise ValueError( + f"Each entry of 'static_paths' in set '{set_name}' must be a list " + "of node names, or a mapping with exactly one of 'nodes' or 'links'" + ) + return tuple(paths) + + def coerce_flow_policy(value: Any) -> Optional[FlowPolicyPreset]: """Return a FlowPolicyPreset from various user-friendly forms. diff --git a/ngraph/model/demand/spec.py b/ngraph/model/demand/spec.py index 246ccaa..2f3d5a5 100644 --- a/ngraph/model/demand/spec.py +++ b/ngraph/model/demand/spec.py @@ -1,11 +1,12 @@ """Traffic demand specification. Defines `TrafficDemand`, a user-facing specification used by demand expansion -and placement. Routing behavior is selected via an optional `FlowPolicyPreset`. +and placement. Routing behavior is selected via an optional `FlowPolicyPreset`, +or pinned to explicit routes with `StaticPath`. """ from dataclasses import dataclass, field -from typing import Any, Dict, Optional, Union +from typing import Any, Dict, Optional, Tuple, Union from ngraph.model.flow.policy_config import FlowPolicyPreset, serialize_policy_preset from ngraph.types.base import Mode @@ -16,6 +17,34 @@ _VALID_GROUP_MODES = ("flatten", "per_group", "group_pairwise") +@dataclass(frozen=True) +class StaticPath: + """One explicit route a demand can be pinned to (an MPLS-style LSP). + + Give exactly one of `nodes` or `links`: + + - `nodes`: the node names the route visits, source first and target last. + Each consecutive pair must be adjacent. When several parallel links + connect a pair, the cheapest is used (ties broken by link id); name the + link explicitly to choose a different one. + - `links`: the link ids the route traverses, in order. Unambiguous when + parallel links exist. A link may be traversed in either direction. + + Attributes: + nodes: Node names along the route, or empty when `links` is given. + links: Link ids along the route, or empty when `nodes` is given. + """ + + nodes: Tuple[str, ...] = () + links: Tuple[str, ...] = () + + def __post_init__(self) -> None: + if bool(self.nodes) == bool(self.links): + raise ValueError("StaticPath requires exactly one of 'nodes' or 'links'") + if self.nodes and len(self.nodes) < 2: + raise ValueError("StaticPath 'nodes' needs at least a source and a target") + + @dataclass class TrafficDemand: """Traffic demand specification using unified selectors. @@ -29,6 +58,10 @@ class TrafficDemand: group_mode: How grouped nodes produce demands ("flatten", "per_group", "group_pairwise"). flow_policy: Policy preset for routing. + static_paths: Explicit routes to pin this demand to. When set, the + demand is placed only on these routes: one flow per route, and a + route broken by a failure carries nothing rather than rerouting. + Requires selectors matching exactly one source and one target. attrs: Arbitrary user metadata. id: Unique identifier. Auto-generated if empty. """ @@ -40,6 +73,7 @@ class TrafficDemand: mode: str = "combine" group_mode: str = "flatten" flow_policy: Optional[FlowPolicyPreset] = None + static_paths: Tuple[StaticPath, ...] = () attrs: Dict[str, Any] = field(default_factory=dict) id: str = "" @@ -55,6 +89,12 @@ def __post_init__(self) -> None: f"Unknown demand group_mode '{self.group_mode}'. " f"Expected one of: {', '.join(_VALID_GROUP_MODES)}" ) + for path in self.static_paths: + if not isinstance(path, StaticPath): + raise ValueError( + f"static_paths entries must be StaticPath objects, got " + f"{type(path).__name__}" + ) if not self.id: # Build a stable identifier from source/target src_key = self.source if isinstance(self.source, str) else str(self.source) @@ -77,5 +117,9 @@ def to_dict(self) -> Dict[str, Any]: "mode": self.mode, "group_mode": self.group_mode, "flow_policy": serialize_policy_preset(self.flow_policy), + "static_paths": [ + {"nodes": list(p.nodes)} if p.nodes else {"links": list(p.links)} + for p in self.static_paths + ], "attrs": dict(self.attrs), } diff --git a/ngraph/model/flow/policy_config.py b/ngraph/model/flow/policy_config.py index d1d98e3..a32069d 100644 --- a/ngraph/model/flow/policy_config.py +++ b/ngraph/model/flow/policy_config.py @@ -82,6 +82,7 @@ def create_flow_policy( preset: FlowPolicyPreset, node_mask=None, edge_mask=None, + static_path_count: Optional[int] = None, ) -> netgraph_core.FlowPolicy: """Create a FlowPolicy instance from a preset configuration. @@ -92,6 +93,9 @@ def create_flow_policy( flow-count bounds to apply. node_mask: Optional numpy bool array for node exclusions (True = include). edge_mask: Optional numpy bool array for edge exclusions (True = include). + static_path_count: Number of routes the caller will pin with + `FlowPolicy.set_static_paths`. Sets the flow count to match, since + a pinned policy creates one flow per route and never grows. Returns: netgraph_core.FlowPolicy: Configured policy instance. @@ -105,6 +109,20 @@ def create_flow_policy( >>> graph = algs.build_graph(strict_multidigraph) >>> policy = create_flow_policy(algs, graph, FlowPolicyPreset.SHORTEST_PATHS_ECMP) """ + + def _build(config: netgraph_core.FlowPolicyConfig) -> netgraph_core.FlowPolicy: + if static_path_count is not None: + # A pinned policy creates exactly one flow per route, so the flow + # bounds must match; Core rejects a mismatch. Cost ceilings and + # reoptimization are inert once paths are pinned. + config.min_flow_count = 1 + config.max_flow_count = static_path_count + config.reoptimize_flows_on_each_placement = False + config.shortest_path = False + return netgraph_core.FlowPolicy( + algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask + ) + if preset == FlowPolicyPreset.SHORTEST_PATHS_ECMP: # Hop-by-hop equal-cost balanced routing (similar to IP forwarding with ECMP) config = netgraph_core.FlowPolicyConfig() @@ -117,9 +135,7 @@ def create_flow_policy( ) config.min_flow_count = 1 config.max_flow_count = 1 - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.SHORTEST_PATHS_WCMP: # Hop-by-hop weighted ECMP (WCMP) over equal-cost paths (proportional split) @@ -133,9 +149,7 @@ def create_flow_policy( ) config.min_flow_count = 1 config.max_flow_count = 1 - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.TE_WCMP_UNLIM: # Traffic engineering with WCMP (proportional split) and capacity-aware selection @@ -149,9 +163,7 @@ def create_flow_policy( ) config.min_flow_count = 1 # max_flow_count defaults to None (unlimited) - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP: # TE with up to 256 LSPs using ECMP flow placement @@ -168,9 +180,7 @@ def create_flow_policy( config.min_flow_count = 1 config.max_flow_count = 256 config.reoptimize_flows_on_each_placement = True - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.TE_ECMP_16_LSP: # TE with exactly 16 LSPs using ECMP flow placement @@ -187,9 +197,7 @@ def create_flow_policy( config.min_flow_count = 16 config.max_flow_count = 16 config.reoptimize_flows_on_each_placement = True - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) else: raise ValueError(f"Unknown flow policy preset: {preset}") diff --git a/ngraph/schemas/scenario.json b/ngraph/schemas/scenario.json index 5ff8a62..3ce5515 100644 --- a/ngraph/schemas/scenario.json +++ b/ngraph/schemas/scenario.json @@ -31,7 +31,9 @@ } } }, - "required": ["name"], + "required": [ + "name" + ], "additionalProperties": false } ] @@ -674,6 +676,55 @@ "attrs": { "type": "object", "description": "Additional demand attributes" + }, + "static_paths": { + "type": "array", + "description": "Explicit routes to pin this demand to (one flow per route, no reroute on failure). Requires selectors matching exactly one source and one target.", + "minItems": 1, + "items": { + "oneOf": [ + { + "type": "array", + "description": "Node names along the route, source first and target last", + "minItems": 2, + "items": { + "type": "string" + } + }, + { + "type": "object", + "properties": { + "nodes": { + "type": "array", + "minItems": 2, + "items": { + "type": "string" + } + } + }, + "required": [ + "nodes" + ], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "links": { + "type": "array", + "minItems": 1, + "items": { + "type": "string" + } + } + }, + "required": [ + "links" + ], + "additionalProperties": false + } + ] + } } }, "required": [ diff --git a/pyproject.toml b/pyproject.toml index 71faf55..0999c0a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ dependencies = [ "pyyaml>=6.0", "pandas>=2.0", "jsonschema>=4.0", - "netgraph-core>=0.7.0", + "netgraph-core>=0.8.0", ] [project.urls] diff --git a/tests/analysis/test_static_paths.py b/tests/analysis/test_static_paths.py new file mode 100644 index 0000000..90838ca --- /dev/null +++ b/tests/analysis/test_static_paths.py @@ -0,0 +1,407 @@ +"""Demands pinned to explicit routes (MPLS-style LSPs). + +Covers the property that distinguishes a pinned route from ordinary routing: +a route broken by a failure carries nothing instead of rerouting. +""" + +from __future__ import annotations + +import netgraph_core +import pytest + +from ngraph.analysis import analyze +from ngraph.analysis.demand import expand_demands +from ngraph.analysis.placement import place_demands +from ngraph.analysis.static_paths import build_static_path_bundles +from ngraph.model.demand.spec import StaticPath, TrafficDemand +from ngraph.model.flow.policy_config import FlowPolicyPreset +from ngraph.model.network import Link, Network, Node + +SHORT = StaticPath(nodes=("A", "B", "C")) +LONG = StaticPath(nodes=("A", "D", "C")) + + +def _network() -> Network: + """A->B->C carries 6 at cost 1/hop; A->D->C carries 4 at cost 9/hop.""" + net = Network() + for name in ("A", "B", "D", "C"): + net.add_node(Node(name)) + net.add_link(Link("A", "B", capacity=6.0, cost=1)) + net.add_link(Link("B", "C", capacity=6.0, cost=1)) + net.add_link(Link("A", "D", capacity=4.0, cost=9)) + net.add_link(Link("D", "C", capacity=4.0, cost=9)) + return net + + +def _place( + paths, + *, + excluded_links=None, + volume: float = 10.0, + preset: FlowPolicyPreset = FlowPolicyPreset.SHORTEST_PATHS_WCMP, + net: Network | None = None, +): + net = net or _network() + demand = TrafficDemand( + source="^A$", + target="^C$", + volume=volume, + mode="pairwise", + flow_policy=preset, + static_paths=tuple(paths), + id="d", + ) + expansion = expand_demands(net, [demand]) + ctx = analyze(net) + flow_graph = netgraph_core.FlowGraph(ctx.multidigraph) + return place_demands( + expansion.demands, + [d.volume for d in expansion.demands], + flow_graph, + ctx, + ctx.build_node_mask(None), + ctx.build_edge_mask(excluded_links), + collect_entries=True, + include_used_edges=True, + ) + + +class TestPlacement: + def test_uses_every_pinned_route(self): + result = _place([SHORT, LONG]) + assert result.summary.total_placed == pytest.approx(10.0) + assert result.entries[0].used_edges == { + "A|B|0:fwd", + "B|C|0:fwd", + "A|D|0:fwd", + "D|C|0:fwd", + } + + def test_pins_traffic_off_the_cheapest_route(self): + """Ordinary routing would send everything down the cost-2 path.""" + result = _place([LONG]) + assert result.summary.total_placed == pytest.approx(4.0) + assert result.entries[0].used_edges == {"A|D|0:fwd", "D|C|0:fwd"} + + def test_broken_route_carries_nothing_and_does_not_reroute(self): + result = _place([SHORT], excluded_links={"B|C|0"}) + assert result.summary.total_placed == pytest.approx(0.0) + + def test_surviving_routes_still_carry_their_share(self): + result = _place([SHORT, LONG], excluded_links={"B|C|0"}) + assert result.summary.total_placed == pytest.approx(4.0) + + def test_link_form_selects_a_specific_parallel_link(self): + net = _network() + net.add_link(Link("A", "B", capacity=1.0, cost=50)) # A|B|1 + cheap = _place([StaticPath(links=("A|B|0", "B|C|0"))], net=net) + pricey = _place([StaticPath(links=("A|B|1", "B|C|0"))], net=_rebuild(net)) + assert cheap.summary.total_placed == pytest.approx(6.0) + assert pricey.summary.total_placed == pytest.approx(1.0) + + def test_node_form_prefers_the_cheapest_parallel_link(self): + net = _network() + net.add_link(Link("A", "B", capacity=1.0, cost=50)) + result = _place([SHORT], net=net) + assert result.summary.total_placed == pytest.approx(6.0) + + +def _rebuild(source: Network) -> Network: + """Copy a network so each placement starts from clean capacity.""" + net = Network() + for name in source.nodes: + net.add_node(Node(name)) + for link in source.links.values(): + net.add_link( + Link(link.source, link.target, capacity=link.capacity, cost=link.cost) + ) + return net + + +class TestResolution: + def _ctx(self): + return analyze(_network()) + + def test_rejects_non_adjacent_hop(self): + with pytest.raises(ValueError, match="has no enabled link"): + build_static_path_bundles( + self._ctx(), [StaticPath(nodes=("A", "C"))], "A", "C" + ) + + def test_rejects_unknown_node(self): + with pytest.raises(ValueError, match="unknown node"): + build_static_path_bundles( + self._ctx(), [StaticPath(nodes=("A", "B", "Z", "C"))], "A", "C" + ) + + def test_rejects_unknown_link(self): + with pytest.raises(ValueError, match="unknown link"): + build_static_path_bundles( + self._ctx(), [StaticPath(links=("nope",))], "A", "C" + ) + + def test_rejects_link_not_leaving_the_reached_node(self): + with pytest.raises(ValueError, match="does not leave node"): + build_static_path_bundles( + self._ctx(), [StaticPath(links=("B|C|0",))], "A", "C" + ) + + def test_rejects_route_with_wrong_endpoints(self): + with pytest.raises(ValueError, match="must run from"): + build_static_path_bundles( + self._ctx(), [StaticPath(nodes=("A", "B"))], "A", "C" + ) + + def test_rejects_route_revisiting_a_node(self): + net = _network() + net.add_link(Link("B", "A", capacity=1.0, cost=1)) + ctx = analyze(net) + with pytest.raises(ValueError, match="simple path"): + build_static_path_bundles( + ctx, [StaticPath(nodes=("A", "B", "A", "B", "C"))], "A", "C" + ) + + +class TestSpec: + def test_requires_exactly_one_form(self): + with pytest.raises(ValueError, match="exactly one"): + StaticPath() + with pytest.raises(ValueError, match="exactly one"): + StaticPath(nodes=("A", "B"), links=("A|B|0",)) + + def test_requires_two_nodes(self): + with pytest.raises(ValueError, match="source and a target"): + StaticPath(nodes=("A",)) + + def test_round_trips_through_to_dict(self): + demand = TrafficDemand( + source="^A$", + target="^C$", + static_paths=(SHORT, StaticPath(links=("A|B|0",))), + ) + assert demand.to_dict()["static_paths"] == [ + {"nodes": ["A", "B", "C"]}, + {"links": ["A|B|0"]}, + ] + + +class TestExpansionConstraints: + def test_rejects_combine_mode(self): + net = _network() + demand = TrafficDemand( + source="^A$", + target="^C$", + volume=1.0, + mode="combine", + static_paths=(SHORT,), + id="d", + ) + with pytest.raises(ValueError, match="use mode 'pairwise'"): + expand_demands(net, [demand]) + + def test_rejects_selectors_matching_several_pairs(self): + net = _network() + net.add_node(Node("A2")) + net.add_link(Link("A2", "B", capacity=1.0, cost=1)) + demand = TrafficDemand( + source="^A", + target="^C$", + volume=1.0, + mode="pairwise", + static_paths=(SHORT,), + id="d", + ) + with pytest.raises(ValueError, match="exactly one source and one target"): + expand_demands(net, [demand]) + + +class TestScenarioYaml: + """The YAML surface: both spellings, and schema rejection of bad forms.""" + + _NETWORK = """ +network: + nodes: + A: {} + B: {} + D: {} + C: {} + links: + - source: A + target: B + capacity: 6 + cost: 1 + - source: B + target: C + capacity: 6 + cost: 1 + - source: A + target: D + capacity: 4 + cost: 9 + - source: D + target: C + capacity: 4 + cost: 9 +""" + + def _scenario(self, static_paths_block: str) -> str: + return ( + self._NETWORK + + """ +demands: + default: + - source: "^A$" + target: "^C$" + volume: 10 + mode: pairwise + flow_policy: SHORTEST_PATHS_WCMP + static_paths: +""" + + static_paths_block + ) + + def test_node_and_link_forms_both_load_and_place(self): + from ngraph.scenario import Scenario + + scenario = Scenario.from_yaml( + self._scenario( + ' - ["A", "B", "C"]\n - links: ["A|D|0", "D|C|0"]\n' + ) + ) + demand = scenario.demand_set.get_set("default")[0] + assert [p.nodes or p.links for p in demand.static_paths] == [ + ("A", "B", "C"), + ("A|D|0", "D|C|0"), + ] + + def test_mapping_form_with_nodes_key(self): + from ngraph.scenario import Scenario + + scenario = Scenario.from_yaml( + self._scenario(' - nodes: ["A", "B", "C"]\n') + ) + assert scenario.demand_set.get_set("default")[0].static_paths[0].nodes == ( + "A", + "B", + "C", + ) + + def test_schema_rejects_empty_and_malformed_routes(self): + import jsonschema + + from ngraph.scenario import Scenario + + with pytest.raises(jsonschema.ValidationError): + Scenario.from_yaml(self._scenario(" []\n")) + with pytest.raises(jsonschema.ValidationError): + Scenario.from_yaml( + self._scenario(' - {nodes: ["A", "B"], links: ["A|B|0"]}\n') + ) + with pytest.raises(jsonschema.ValidationError): + Scenario.from_yaml(self._scenario(' - ["A"]\n')) + + +class TestDisabledLinks: + """A route must never be pinned to an administratively disabled link.""" + + def _network_with_disabled_cheap_link(self) -> Network: + net = Network() + for name in ("A", "B", "C"): + net.add_node(Node(name)) + net.add_link(Link("A", "B", capacity=6.0, cost=1)) # cheapest, disabled + net.add_link(Link("A", "B", capacity=7.0, cost=2)) # enabled alternative + net.add_link(Link("B", "C", capacity=100.0, cost=1)) + net.links["A|B|0"].disabled = True + return net + + def test_node_hop_skips_disabled_link(self): + result = _place( + [StaticPath(nodes=("A", "B", "C"))], + net=self._network_with_disabled_cheap_link(), + ) + assert result.summary.total_placed == pytest.approx(7.0) + assert "A|B|1:fwd" in result.entries[0].used_edges + + def test_naming_a_disabled_link_is_an_error(self): + ctx = analyze(self._network_with_disabled_cheap_link()) + with pytest.raises(ValueError, match="disabled link"): + build_static_path_bundles( + ctx, [StaticPath(links=("A|B|0", "B|C|0"))], "A", "C" + ) + + def test_hop_with_only_disabled_links_is_an_error(self): + net = self._network_with_disabled_cheap_link() + net.links["A|B|1"].disabled = True + with pytest.raises(ValueError, match="has no enabled link"): + build_static_path_bundles( + analyze(net), [StaticPath(nodes=("A", "B", "C"))], "A", "C" + ) + + +class TestInputForms: + """Every route form the DSL accepts must work programmatically too.""" + + def test_config_round_trip_accepts_list_and_mapping_forms(self): + from ngraph.analysis.functions import _static_paths_from_config + + assert _static_paths_from_config([["A", "B", "C"]]) == ( + StaticPath(nodes=("A", "B", "C")), + ) + assert _static_paths_from_config([{"links": ["A|B|0"]}]) == ( + StaticPath(links=("A|B|0",)), + ) + assert _static_paths_from_config([StaticPath(nodes=("A", "B"))]) == ( + StaticPath(nodes=("A", "B")), + ) + + @pytest.mark.parametrize( + "bad", + [[{"path": ["A", "B"]}], ["not-a-route"], [{"nodes": ["A"], "links": ["x"]}]], + ) + def test_config_round_trip_rejects_other_shapes(self, bad): + from ngraph.analysis.functions import _static_paths_from_config + + with pytest.raises(ValueError, match="Invalid static path"): + _static_paths_from_config(bad) + + def test_traffic_demand_rejects_non_staticpath_entries(self): + with pytest.raises(ValueError, match="must be StaticPath objects"): + TrafficDemand( + source="^A$", target="^B$", static_paths=[{"nodes": ["A", "B"]}] + ) + + def test_yaml_rejects_non_string_hops(self): + from ngraph.model.demand.builder import _build_static_paths + + with pytest.raises(ValueError, match="must all be strings"): + _build_static_paths([["A", ["B", "B"], "C"]], "default") + + +class TestPinnedDemandIsNeverSilentlyDropped: + def test_selector_matching_nothing_raises(self): + net = _network() + demand = TrafficDemand( + source="^A$", + target="^NOPE$", + volume=5.0, + mode="pairwise", + static_paths=(SHORT,), + id="d", + ) + with pytest.raises(ValueError, match="match no active source or target"): + expand_demands(net, [demand]) + + +class TestBundleCaching: + """Bundles are resolved once per context but must still track the masks.""" + + def test_repeated_placements_track_changing_failures(self): + outcomes = [] + for excluded in (None, {"B|C|0"}, None, {"A|D|0"}, {"B|C|0", "A|D|0"}, None): + outcomes.append( + round( + _place([SHORT, LONG], excluded_links=excluded).summary.total_placed, + 3, + ) + ) + assert outcomes == [10.0, 4.0, 10.0, 6.0, 0.0, 10.0] From c0231b00da550c4cfa166cc4ecb12572090328aa Mon Sep 17 00:00:00 2001 From: Andrey Golovanov Date: Mon, 24 Aug 2026 01:59:25 +0100 Subject: [PATCH 2/2] Scope the max-flow docs to what the 0.8.0 engine actually does Publishing netgraph-core 0.8.0 made several statements in this release inaccurate. An audit that ran ngraph against 0.7.2 and 0.8.0 side by side, cross-checked against networkx.maximum_flow_value on thousands of random topologies, found 0.7.2 understates max flow on cost-asymmetric graphs and 0.8.0 is exact -- so the completion phase is a correctness fix, and the docs should describe it rather than the tier loop alone. The min-cut duality guarantee announced under BREAKING was unqualified. It holds for the default max-flow configuration (PROPORTIONAL, require_capacity, not shortest_path) -- the same gate the C++ completion phase uses -- and not for the placement models. Scoped in the changelog and in api.md. Also corrected, all wrong on both this branch and its parent: - cost_distribution keys: completion-phase entries are marginal costs (forward edge costs minus the cancelled flow's cost), so a key need not match any traversable path. design.md and the MaxFlowResult docstring said "path cost tier", which is only true of the tier loop. - The MAX_FLOW pseudocode ended at the tier loop, so a reader implementing it would reproduce 0.7.x's smaller answer. Added the completion phase. - The complexity bound was justified by "placed flow is never removed from an edge", which the completion phase does. The bound itself still holds -- Edmonds-Karp at O(VE^2) is dominated -- so only the justification changed. - "Does not re-route previously placed flow" now says which phase does not. No behavior change; docs and changelog only. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 2 +- docs/reference/api-full.md | 6 ++++-- docs/reference/api.md | 2 +- docs/reference/design.md | 22 +++++++++++++++++++--- ngraph/types/dto.py | 4 +++- 5 files changed, 28 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d60f26..6c594a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,7 @@ Analyses that previously returned wrong numbers without any error. If you rely o - Seeded Monte Carlo runs were not reproducible across identical scenario rebuilds: link ids carried a random uuid suffix whose sort order changed per build, remapping seeded draws onto different parallel links. Link ids are now a deterministic per-(source, target) sequence (`A|B|0`), and `add_link` raises instead of silently overwriting when an id collides (possible when node names contain `|`) or when the same link is added twice - A risk group sharing its name with a node or link excluded the wrong entity: the node was excluded and the group's members were left up. Failed entities are now classified by rule scope - `expand_groups` expanded differently depending on whether a failure came from an entity rule or a `risk_group` rule, and never reached members of nested groups. Expansion is now transitive and identical for both rule kinds -- **BREAKING**: `MaxFlowResult.min_cut` (`max_flow_detailed` with `include_min_cut=True`) returns a true minimum cut whose capacity equals the max flow, instead of all saturated edges. Saturated-edge analysis remains available via `sensitivity()` +- **BREAKING**: `MaxFlowResult.min_cut` (`max_flow_detailed` with `include_min_cut=True`) returns a true minimum cut whose capacity equals the max flow, instead of all saturated edges. The capacities match for the default max-flow configuration (`PROPORTIONAL` placement, `require_capacity=True`, `shortest_path=False`); the other configurations are placement models rather than max-flow computations. Saturated-edge analysis remains available via `sensitivity()` - Monte Carlo with `seed=None` now falls back to the failure policy's own seed; a seeded policy previously produced one identical failure pattern on every iteration - `group_mode: per_group` now matches its documented semantics: `combine` creates one demand per source group, `pairwise` pairs nodes within each same-label group, and volume splits evenly across groups. A skipped group's share is not redistributed - `k_shortest_paths` between multi-node groups merges results across all source/sink pairs instead of returning paths for the single best pair, and breaks ties structurally so results no longer depend on `PYTHONHASHSEED`; `shortest_paths` ordering is likewise deterministic diff --git a/docs/reference/api-full.md b/docs/reference/api-full.md index c928fad..33ce544 100644 --- a/docs/reference/api-full.md +++ b/docs/reference/api-full.md @@ -12,7 +12,7 @@ Quick links: - [CLI Reference](cli.md) - [DSL Reference](dsl.md) -Generated from source code on: August 24, 2026 at 00:43 UTC +Generated from source code on: August 24, 2026 at 01:58 UTC Modules auto-discovered: 54 @@ -2614,7 +2614,9 @@ Captures total flow, cost distribution, and optionally min-cut edges. Attributes: total_flow: Maximum flow value achieved. - cost_distribution: Mapping of path cost to flow volume placed at that cost. + cost_distribution: Mapping of cost to flow volume placed at that cost. + Completion-phase entries are marginal costs, so a key need not match + any single path's cost. min_cut: Edges forming a minimum cut (None if not computed). **Attributes:** diff --git a/docs/reference/api.md b/docs/reference/api.md index 3149710..ce9943e 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -229,7 +229,7 @@ print(summary.cost_distribution) # Dict[float, float] mapping cost to flow volu - `analyze(network, *, source=None, sink=None, mode=Mode.COMBINE)` - Create analysis context - `ctx.max_flow(source, sink, *, mode, shortest_path, require_capacity, flow_placement, excluded_nodes, excluded_links)` - Maximum flow -- `ctx.max_flow_detailed(..., include_min_cut=False)` - Maximum flow with cost distribution and optional min-cut; the min-cut is a true minimum cut (its capacity equals the max flow), not the set of saturated edges +- `ctx.max_flow_detailed(..., include_min_cut=False)` - Maximum flow with cost distribution and optional min-cut; the min-cut is a true minimum cut (its capacity equals the max flow under the default `PROPORTIONAL` placement with `require_capacity=True` and `shortest_path=False`), not the set of saturated edges - `ctx.sensitivity(...)` - Identify critical edges and their impact on flow - `ctx.sensitivity_with_flow(...)` - Compute max flow and edge sensitivity together per group pair in a single pass (used by the sensitivity Monte Carlo hot path) - `ctx.shortest_path_cost(source, sink, *, mode, edge_select=ALL_MIN_COST, excluded_nodes, excluded_links)` - Shortest path cost diff --git a/docs/reference/design.md b/docs/reference/design.md index 12c58d7..9e560ba 100644 --- a/docs/reference/design.md +++ b/docs/reference/design.md @@ -521,7 +521,7 @@ After the loop, the C++ algorithm computes a FlowSummary which includes: - min_cut: the list of edges that are saturated and go from reachable to non-reachable (these form the minimum cut) -- cost_distribution: flow volume placed at each path cost tier. Core returns parallel arrays (`costs`, `flows`); AnalysisContext converts these to the `Dict[Cost, Flow]` mapping in `MaxFlowResult.cost_distribution`. +- cost_distribution: flow volume keyed by cost. Tier-loop entries are the cost of the shortest-path DAG the flow was placed on; completion-phase entries are marginal costs (the augmenting path's forward edge costs minus the cost of the flow it cancels), so a key need not correspond to any traversable path. Core returns parallel arrays (`costs`, `flows`); AnalysisContext converts these to the `Dict[Cost, Flow]` mapping in `MaxFlowResult.cost_distribution`. The summary is returned along with the total flow value. @@ -688,6 +688,22 @@ function MAX_FLOW(graph, S, T, placement=PROPORTIONAL, require_capacity=True, if shortest_path: # Single augmentation pass (IP/IGP mode) break + # Completion phase: max-flow semantics only. The tier loop above walks + # forward residual edges, so it can stop below the true maximum. + if placement == PROPORTIONAL and require_capacity and not shortest_path: + while True: + # BFS over the full residual graph, including reverse arcs that + # return previously placed flow + path = BFS_AUGMENTING_PATH(flow_state.residual_view(), S, T) + if path is None: + break + placed = flow_state.augment(path) + if placed < kMinFlow: + break + total_flow += placed + # Marginal cost: forward edge costs minus the cost of cancelled flow + cost_distribution[marginal_cost(path)] += placed + # Compute min-cut, reachability, cost distribution min_cut = flow_state.compute_min_cut(S, node_mask, edge_mask) @@ -703,7 +719,7 @@ function MAX_FLOW(graph, S, T, placement=PROPORTIONAL, require_capacity=True, The flow tolerance constant `kMinFlow` (1/4096 ≈ 2.4e-4) determines when flow placement is considered negligible and iteration terminates. -Each augmentation phase performs one SPF \(O((V+E) \log V)\) and one placement pass over the tier's predecessor DAG. For EQUAL_BALANCED the placement is a single topological pass \(O(V+E)\); for PROPORTIONAL it is a complete Dinic max-flow over the tier DAG (repeated BFS level construction, level-restricted blocking-flow DFS, and a group rebuild from the updated residual), worst case \(O(V^2 E)\). Placed flow is never removed from an edge, so each phase permanently saturates at least one edge before the next SPF runs, bounding the number of phases by \(O(E)\); with PROPORTIONAL placement the tier's path cost also strictly increases between phases, so phases are further bounded by the number of distinct path-cost values. The resulting loose worst-case bound is \(O(E \cdot (V^2 E + (V+E) \log V))\). +Each augmentation phase performs one SPF \(O((V+E) \log V)\) and one placement pass over the tier's predecessor DAG. For EQUAL_BALANCED the placement is a single topological pass \(O(V+E)\); for PROPORTIONAL it is a complete Dinic max-flow over the tier DAG (repeated BFS level construction, level-restricted blocking-flow DFS, and a group rebuild from the updated residual), worst case \(O(V^2 E)\). The tier loop never removes placed flow, so each phase permanently saturates at least one edge before the next SPF runs, bounding the number of phases by \(O(E)\); with PROPORTIONAL placement the tier's path cost also strictly increases between phases, so phases are further bounded by the number of distinct path-cost values. The resulting loose worst-case bound is \(O(E \cdot (V^2 E + (V+E) \log V))\). The completion phase that follows is Edmonds-Karp at \(O(V E^2)\), which this bound dominates. Practical performance is significantly better than these worst-case bounds: iteration stops as soon as the residual network disconnects source from sink, the phase count in practice equals the small number of cost tiers actually used, and the `kMinFlow` threshold additionally caps the number of phases at \(F / k_{MinFlow}\) for total flow \(F\). @@ -775,7 +791,7 @@ NetGraph's design includes several features that differentiate it from tradition - Configurable flow placement: Proportional (WCMP-style, capacity-weighted) and Equal-Balanced (ECMP-style, uniform) splitting across parallel equal-cost edges -- Cost-aware augmentation: Prefer cheapest capacity first via successive shortest paths. Does not re-route previously placed flow. +- Cost-aware augmentation: Prefer cheapest capacity first via successive shortest paths. The cost-tier loop does not re-route previously placed flow; the max-flow completion phase may cancel earlier placements to reach the true maximum. - Deterministic simulation with seeding: Random aspects (e.g., failure sampling) are controlled by explicit seeds that propagate through steps. Runs are reproducible given the same scenario and seed. diff --git a/ngraph/types/dto.py b/ngraph/types/dto.py index 25f9f49..1ee38a0 100644 --- a/ngraph/types/dto.py +++ b/ngraph/types/dto.py @@ -38,7 +38,9 @@ class MaxFlowResult: Attributes: total_flow: Maximum flow value achieved. - cost_distribution: Mapping of path cost to flow volume placed at that cost. + cost_distribution: Mapping of cost to flow volume placed at that cost. + Completion-phase entries are marginal costs, so a key need not match + any single path's cost. min_cut: Edges forming a minimum cut (None if not computed). """