From 716cd4d6364920140f7a7b98eb0eb92ae53699af Mon Sep 17 00:00:00 2001 From: Andrey Golovanov Date: Sun, 13 Sep 2026 00:02:16 +0100 Subject: [PATCH] Placement semantics: pooled combine mode, lossy ECMP, resolution-based MSD Demand placement fixes and the models they rest on: - Combine mode is a virtual source: with hop-by-hop presets every source that can reach a target originates an even share (fan-out DAG built with Core's reverse SPF), and under SHORTEST_PATHS_ECMP the pool is admitted as one demand at a single lossless scale. TE presets keep letting capacity decide which sources originate. - SHORTEST_PATHS_ECMP models a load-blind forwarding table (Core EQUAL_BALANCED_FIXED); new SHORTEST_PATHS_ECMP_LOSSY delivers what survives per-link drops and reports dropped_edges. - One preset_config feeds both placement engines; hop-by-hop FlowPolicies are cost-only and single-pass, matching the cached engine. - TE rerouting is bounded by the edge count instead of a literal 100. - MSD feasibility is judged at the engine's resolution (1/4096) and requires every demand to place something. - TrafficMatrixPlacement resolves parallelism "auto" to 1 unless an LSP preset is present or the interpreter is free-threaded. Documentation reviewed for plain language and current semantics; examples extended with demand placement, failure Monte Carlo, and result reading, all verified against the code. Minimum netgraph-core is 0.9.0. Version 0.23.0. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 23 + README.md | 5 +- docs/examples/basic.md | 34 + docs/examples/bundled-scenarios.md | 35 ++ docs/examples/clos-fabric.md | 51 +- docs/getting-started/tutorial.md | 53 +- docs/index.md | 6 +- docs/reference/api-full.md | 145 ++++- docs/reference/api.md | 19 +- docs/reference/cli.md | 103 +-- docs/reference/design.md | 112 ++-- docs/reference/dsl.md | 19 +- docs/reference/schemas.md | 20 +- docs/reference/workflow.md | 15 +- ngraph/analysis/demand.py | 13 + ngraph/analysis/functions.py | 51 +- ngraph/analysis/placement.py | 528 +++++++++++++--- ngraph/model/flow/policy_config.py | 227 ++++--- .../workflow/maximum_supported_demand_step.py | 38 +- .../workflow/traffic_matrix_placement_step.py | 65 +- pyproject.toml | 4 +- .../test_demand_expansion_semantics.py | 24 +- tests/analysis/test_placement.py | 24 +- tests/analysis/test_placement_models.py | 592 ++++++++++++++++++ tests/model/demand/test_builder.py | 8 + tests/model/flow/test_policy_config.py | 67 ++ tests/workflow/test_msd_resolution.py | 87 +++ tests/workflow/test_placement_parallelism.py | 64 ++ 28 files changed, 1961 insertions(+), 471 deletions(-) create mode 100644 tests/analysis/test_placement_models.py create mode 100644 tests/workflow/test_msd_resolution.py create mode 100644 tests/workflow/test_placement_parallelism.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a20842..2dcd152 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,29 @@ 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). +## [0.23.0] - 2026-09-13 + +### Fixed + +Placement results that were wrong in plausible scenarios. Re-run analyses that used these configurations. + +- Combine mode with a `SHORTEST_PATHS_*` preset routed the whole demand from the pseudo source, so only the sources nearest the targets carried traffic and the equal hash across sources let one small source scale the rest down. The pseudo source is now a pool: every source that can reach a target originates an even share, routed to its nearest targets over a DAG built by Core's new reverse SPF. Under `SHORTEST_PATHS_ECMP` the pool is admitted as one demand at a single lossless scale (two equal-cost sources of capacity 100 and 10 admit 20 of 110; `SHORTEST_PATHS_ECMP_LOSSY` delivers 65). `TE_*` presets keep letting capacity decide which sources originate. A fixed per-source matrix is `group_mode: per_group` with `group_by: name`, or `pairwise` +- `SHORTEST_PATHS_ECMP` skipped next hops filled by earlier demands, overstating lossless capacity. It now models a load-blind forwarding table: a filled next hop blocks later demands hashed onto it (Core placement `EQUAL_BALANCED_FIXED`) +- `TE_WCMP_UNLIM` stopped rerouting after 100 cost tiers and dropped the remainder silently. The loop is now bounded by the edge count +- `MaximumSupportedDemand` failed with "No feasible alpha found" for LSP presets over many small demands. Core never places less than 1/4096 on a flow, so those demands were always short by a fraction of that. Feasible now means every demand is placed to within that resolution and none placed nothing; the failure message reports the best ratio seen +- `create_flow_policy` left Core's `require_capacity` at its default for the `SHORTEST_PATHS_*` presets, so a FlowPolicy built from an IGP preset selected paths with residual awareness and the WCMP one rerouted. Both engines now read one `preset_config`, and a hop-by-hop FlowPolicy places exactly what the cached engine places + +### Changed + +- `TrafficMatrixPlacement` resolves `parallelism: auto` to 1 unless the demand set uses an LSP preset or the interpreter is free-threaded. Iterations for the other presets are Python-bound between short engine calls, and the CPU-count default made Monte Carlo about 2x slower than serial. Explicit worker counts are unchanged +- Minimum `netgraph-core` raised to 0.9.0 (new placement modes, per-link drop reporting, reverse SPF) +- Documentation describes placement as greedy and sequential: priority order, input order within a priority, order-dependent totals under contention + +### Added + +- `SHORTEST_PATHS_ECMP_LOSSY` preset: hop-by-hop ECMP forwarded best-effort. Every link carries what fits and drops the rest, `placed` is the delivered volume, and with `include_flow_details` each entry reports `dropped_edges`, the lost volume per link. On a 100/10 parallel pair offered 100 units, `SHORTEST_PATHS_ECMP` admits 20 and the lossy preset delivers 60 +- `ngraph.model.flow.policy_config.preset_config` and `HOP_BY_HOP_PRESETS`; `PlacementSummary.max_shortfall` and `unserved_demands`; `resolve_placement_parallelism` + ## [0.22.0] - 2026-08-24 ### Fixed diff --git a/README.md b/README.md index 5dfaedc..b1fb30a 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ [![Python-test](https://github.com/networmix/NetGraph/actions/workflows/python-test.yml/badge.svg?branch=main)](https://github.com/networmix/NetGraph/actions/workflows/python-test.yml) -Network modeling and analysis framework combining Python with high-performance C++ graph algorithms. +Network modeling and analysis framework: Python front end, C++ graph algorithms. ## What It Does @@ -105,6 +105,7 @@ workflow: ```bash ngraph run scenario.yml --output results/ +jq '.steps.max_demand.data.alpha_star' results/scenario.results.json ``` This scenario builds a dual-site Clos fabric from blueprints, finds the maximum supportable demand, then runs 100 Monte Carlo iterations with random link failures - exporting results to JSON. @@ -118,7 +119,7 @@ See [DSL Reference](https://networmix.github.io/NetGraph/reference/dsl/) and [Ex - **Routing modes** for IP routing (cost-based) and traffic engineering (capacity-aware) - **Flow placement** strategies for ECMP and WCMP with max-flow and capacity envelopes - **Reproducible results** via seeded randomness and stable edge IDs -- **C++ performance** with GIL released via [NetGraph-Core](https://github.com/networmix/NetGraph-Core) +- **C++ algorithms** with the GIL released, via [NetGraph-Core](https://github.com/networmix/NetGraph-Core) ## Documentation diff --git a/docs/examples/basic.md b/docs/examples/basic.md index 4a91e8d..d8da404 100644 --- a/docs/examples/basic.md +++ b/docs/examples/basic.md @@ -241,3 +241,37 @@ for pair, path_list in k_paths.items(): for i, path in enumerate(path_list, 1): print(f" {i}. Cost: {path.cost}") ``` + +## Demand Placement + +Max-flow asks how much the network could carry. Demand placement asks how much of a given volume it does carry under a routing model. The same 6 units from A to C give a different answer under each preset: + +```python +from ngraph.analysis.functions import demand_placement_analysis + +for preset in ("SHORTEST_PATHS_ECMP", "SHORTEST_PATHS_ECMP_LOSSY", + "SHORTEST_PATHS_WCMP", "TE_WCMP_UNLIM"): + result = demand_placement_analysis( + network, + excluded_nodes=set(), + excluded_links=set(), + demands_config=[{"source": "^A$", "target": "^C$", "volume": 6, + "mode": "pairwise", "flow_policy": preset}], + include_flow_details=True, + ) + entry = result.flows[0] + print(f"{preset}: placed={entry.placed:g} dropped={entry.dropped:g} " + f"by_cost={entry.cost_distribution} {entry.data}") + +# SHORTEST_PATHS_ECMP: placed=2 dropped=4 by_cost={2.0: 2.0} {} +# SHORTEST_PATHS_ECMP_LOSSY: placed=2.5 dropped=3.5 by_cost={2.0: 2.5} {'dropped_edges': {'A|B|0:fwd': 2.0, 'A|B|1:fwd': 1.0, 'B|C|0:fwd': 0.5}} +# SHORTEST_PATHS_WCMP: placed=3 dropped=3 by_cost={2.0: 3.0} {} +# TE_WCMP_UNLIM: placed=6 dropped=0 by_cost={2.0: 3.0, 4.0: 3.0} {} +``` + +- `SHORTEST_PATHS_ECMP` hashes 3 units onto each parallel link of the cost-2 path. The capacity-1 link admits only 1 without loss, so the whole demand is admitted at that scale: 2 units. +- `SHORTEST_PATHS_ECMP_LOSSY` sends the same 3 and 3, and each link carries what fits. 2.5 units arrive; `dropped_edges` says where the other 3.5 were lost. +- `SHORTEST_PATHS_WCMP` splits by capacity, so the cost-2 path carries its full 3 units. The demand does not leave the shortest path, so 3 units are unmet. +- `TE_WCMP_UNLIM` reroutes the remainder onto the cost-4 path and places everything. + +In a scenario file the same choice is the demand's `flow_policy`; see the [Tutorial](../getting-started/tutorial.md) for placement inside a workflow. diff --git a/docs/examples/bundled-scenarios.md b/docs/examples/bundled-scenarios.md index 0e32d0f..11b1525 100644 --- a/docs/examples/bundled-scenarios.md +++ b/docs/examples/bundled-scenarios.md @@ -74,6 +74,41 @@ ngraph run scenarios/nsfnet.yaml --output out ngraph run scenarios/nsfnet.yaml --keys node_to_node_capacity_matrix_1 --stdout ``` +## Reading the results + +`--stdout` prints only the JSON, so it pipes into `jq`: + +```bash +# The largest traffic multiplier that still fits (square_mesh: 1.0) +ngraph run scenarios/square_mesh.yaml --no-results --stdout --keys msd_baseline \ + | jq '.steps.msd_baseline.data.alpha_star' + +# One line per distinct failure pattern: which links failed, how many +# iterations drew it, and the fraction of demand still placed +ngraph run scenarios/square_mesh.yaml --no-results --stdout --keys tm_placement \ + | jq -c '.steps.tm_placement.data.flow_results[] + | {links: .failure_state.excluded_links, n: .occurrence_count, ratio: .summary.overall_ratio}' + +# Pairwise capacity matrix: 676 source/destination pairs in the no-failure baseline +ngraph run scenarios/nsfnet.yaml --no-results --stdout --keys node_to_node_capacity_matrix_1 \ + | jq '.steps.node_to_node_capacity_matrix_1.data.baseline.flows | length' + +# Capex and power per metro from the components library +ngraph run scenarios/backbone_clos.yml --no-results --stdout --keys cost_power \ + | jq -c '.steps.cost_power.data.levels["1"][] | {path, capex_total, power_total_watts}' +``` + +The `square_mesh` placement output looks like this (1000 iterations, six single-link patterns): + +```text +{"links":["N1|N2|0"],"n":164,"ratio":0.8333333333333334} +{"links":["N3|N4|0"],"n":180,"ratio":0.8333333333333334} +{"links":["N2|N4|0"],"n":165,"ratio":1.0} +{"links":["N2|N3|0"],"n":183,"ratio":0.8333333333333334} +{"links":["N1|N3|0"],"n":164,"ratio":1.0} +{"links":["N1|N4|0"],"n":144,"ratio":0.8333333333333334} +``` + ## Notes on results All runs emit a consistent JSON shape with `workflow`, `steps`, and `scenario` sections. Steps like `MaxFlow` and `TrafficMatrixPlacement` store a list under `data.flow_results` with one entry per unique failure pattern - patterns are deduplicated across iterations, so the list holds at most `iterations` entries and usually far fewer - alongside a single unfailed entry under `data.baseline`; with no `failure_policy`, `flow_results` is empty. Each entry carries a `summary` and per-flow `flows` entries whose `cost_distribution` is populated when `include_flow_details` is set (and `{}` otherwise), and with `include_min_cut` the min-cut edges appear under a flow entry's `data` (`edges` plus `edges_kind: "min_cut"`). See Reference -> Workflow for the exact schema. diff --git a/docs/examples/clos-fabric.md b/docs/examples/clos-fabric.md index 15dbfc2..6b4ef02 100644 --- a/docs/examples/clos-fabric.md +++ b/docs/examples/clos-fabric.md @@ -173,7 +173,7 @@ for lk in network.links.values(): (s.startswith("my_clos2/spine") and t.startswith("my_clos1/spine")): groups[(s, t)].append(lk) for i, key in enumerate(sorted(groups.keys())): - links = sorted(groups[key], key=lambda x: (x.source, x.target, id(x))) + links = sorted(groups[key], key=lambda x: x.id) caps = [4.0, 0.25, 0.25, 0.25] if i % 2 == 0 else [2.0, 1.0, 0.5, 0.25] for lk, cap in zip(links, caps): lk.capacity = cap @@ -207,6 +207,55 @@ Uneven WCMP: {('b1|b2', 'b1|b2'): 248.0} As expected, WCMP achieves higher throughput than ECMP when parallel links within equal-cost bundles have uneven capacities. ECMP is limited by the link with the lowest capacity in the equal-cost group. +## Failure Analysis + +The same ECMP-versus-WCMP question under failures, this time with `FailureManager` running a Monte Carlo over random spine failures in `my_clos1`. Each iteration fails two spines; identical failure patterns are run once and weighted by how often they were drawn. + +```python +from collections import Counter +from ngraph import FailureManager, FlowPlacement +from ngraph.model.failure.policy import FailurePolicy, FailureMode, FailureRule +from ngraph.model.failure.policy_set import FailurePolicySet + +# Restore symmetric inter-spine links for this section +for lk in network.links.values(): + if lk.source.startswith("my_clos1/spine") or lk.source.startswith("my_clos2/spine"): + lk.capacity = 1.0 + +two_spines = FailurePolicy(modes=[FailureMode(weight=1.0, rules=[ + FailureRule(scope="node", mode="choice", count=2, path="^my_clos1/spine/"), +])]) +fm = FailureManager( + network=network, + failure_policy_set=FailurePolicySet(policies={"two_spines": two_spines}), + policy_name="two_spines", +) + +for placement in (FlowPlacement.EQUAL_BALANCED, FlowPlacement.PROPORTIONAL): + mc = fm.run_max_flow_monte_carlo( + source=r"my_clos1.*(b[0-9]*)/t1", + target=r"my_clos2.*(b[0-9]*)/t1", + mode="combine", + iterations=100, + parallelism=1, + seed=1, + shortest_path=True, + flow_placement=placement, + ) + capacity = Counter() + for item in mc["results"]: + capacity[item.summary.total_placed] += item.occurrence_count + print(placement.name, "baseline", mc["baseline"].summary.total_placed, + "under failure", dict(sorted(capacity.items()))) +``` + +```text +EQUAL_BALANCED baseline 256.0 under failure {192.0: 13, 224.0: 87} +PROPORTIONAL baseline 256.0 under failure {248.0: 100} +``` + +Losing two spines removes 8 of 256 inter-spine links. WCMP loses exactly that capacity in every iteration. ECMP loses 32, or 64 when both failed spines serve the same t2 switch, because the surviving equal-cost next hops still receive equal shares and the smallest one caps the whole split. + ## Network Structure Analysis We can also analyze the network structure using the NetworkExplorer: diff --git a/docs/getting-started/tutorial.md b/docs/getting-started/tutorial.md index b42c862..f081740 100644 --- a/docs/getting-started/tutorial.md +++ b/docs/getting-started/tutorial.md @@ -17,33 +17,66 @@ ngraph run scenarios/square_mesh.yaml --keys msd_baseline --stdout See also: `scenarios/backbone_clos.yml` and `scenarios/nsfnet.yaml`. -## Programmatic: minimal example +## Programmatic: a small workflow + +A three-node network, one demand, and the two steps most analyses start with: find the largest multiplier of the traffic matrix that still fits (`MaximumSupportedDemand`), then place the matrix under random single-link failures (`TrafficMatrixPlacement`). ```python from ngraph.scenario import Scenario scenario_yaml = """ +seed: 42 + network: - nodes: - A: {} - B: {} + nodes: {A: {}, B: {}, C: {}} links: - - {source: A, target: B, capacity: 10.0, cost: 1.0} + - {source: A, target: B, capacity: 10, cost: 1} + - {source: B, target: C, capacity: 10, cost: 1} + - {source: A, target: C, capacity: 5, cost: 3} + +failures: + single_link: + modes: + - weight: 1.0 + rules: [{scope: link, mode: choice, count: 1}] + +demands: + default: + - {source: ^A$, target: ^C$, volume: 8, mode: pairwise, flow_policy: TE_WCMP_UNLIM} + workflow: - - type: NetworkStats - name: baseline_stats + - {type: MaximumSupportedDemand, name: msd, demand_set: default} + - {type: TrafficMatrixPlacement, name: placement, demand_set: default, + failure_policy: single_link, iterations: 20} """ scenario = Scenario.from_yaml(scenario_yaml) scenario.run() +steps = scenario.results.to_dict()["steps"] + +print("alpha_star:", steps["msd"]["data"]["alpha_star"]) +placement = steps["placement"]["data"] +print("baseline placed:", placement["baseline"]["summary"]["total_placed"]) +for pattern in placement["flow_results"]: + failed = pattern["failure_state"]["excluded_links"] + summary = pattern["summary"] + print(f" {failed} x{pattern['occurrence_count']}: " + f"placed {summary['total_placed']:.0f} of {summary['total_demand']:.0f}") +``` -exported = scenario.results.to_dict() -print(list(exported["steps"].keys())) +```text +alpha_star: 1.875 +baseline placed: 8.0 + ['B|C|0'] x8: placed 5 of 8 + ['A|C|0'] x7: placed 8 of 8 + ['A|B|0'] x5: placed 5 of 8 ``` +`alpha_star` is 1.875 because A can reach C with 15 units in total (10 through B plus 5 direct) and the demand is 8. The 20 failure iterations collapse into three distinct patterns; `occurrence_count` says how many iterations drew each one. Losing either link of the B path leaves only the 5-unit direct link. + ## Results structure -Results are exported with a fixed structure containing `workflow`, `steps`, and `scenario` sections. Steps such as `MaxFlow`, `TrafficMatrixPlacement`, and `MaximumSupportedDemand` write their outputs under their step name. See the Workflow Reference for field details. +Results have a fixed shape with `workflow`, `steps`, and `scenario` sections. Each step writes `metadata` and `data` under its name: `MaximumSupportedDemand` writes `data.alpha_star`, and `MaxFlow` and `TrafficMatrixPlacement` write a no-failure `data.baseline` plus `data.flow_results`, one entry per distinct failure pattern with its `occurrence_count`, per-flow `flows`, and a `summary`. See the [Workflow Reference](../reference/workflow.md) for every field. ## Next steps diff --git a/docs/index.md b/docs/index.md index 6377f87..24955f2 100644 --- a/docs/index.md +++ b/docs/index.md @@ -31,13 +31,13 @@ NetGraph uses a hybrid Python+C++ architecture, split across two layers: ### Traffic Engineering -- **Routing Modes**: Unified modeling of IP routing (static costs, oblivious to congestion) and traffic engineering (dynamic residuals, congestion-aware). +- **Routing Modes**: IP routing (cost-only, fixed paths) and traffic engineering (capacity-aware) in one model. - **Flow Placement**: Strategies for ECMP (Equal-Cost Multi-Path) and WCMP (Weighted Cost Multi-Path). -- **Capacity Analysis**: Compute max-flow envelopes and demand allocation with configurable placement policies. +- **Capacity Analysis**: Max-flow between node groups and traffic-matrix placement with selectable placement policies. ### Workflow & Integration -- **Structured Results**: Export analysis artifacts to JSON for downstream processing. +- **Structured Results**: JSON export with a fixed shape. - **CLI**: Validate, inspect, and run scenarios from the command line. - **Python API**: Programmatic access to the same modeling and solving entry points. diff --git a/docs/reference/api-full.md b/docs/reference/api-full.md index 33ce544..c665db0 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 01:58 UTC +Generated from source code on: September 12, 2026 at 22:35 UTC Modules auto-discovered: 54 @@ -909,8 +909,11 @@ Raises: Flow policy preset configurations for NetGraph. -Named routing presets and the factory that materializes them as NetGraph-Core -FlowPolicy objects built from a FlowPolicyConfig. +Named routing presets, the single mapping from a preset to a NetGraph-Core +``FlowPolicyConfig``, and the factory that materializes a preset as a Core +``FlowPolicy``. Both placement engines (the SPF-cached fast path in +``ngraph.analysis.placement`` and Core's FlowPolicy) read their edge selection +and placement mode from ``preset_config`` so the two cannot drift. ### FlowPolicyPreset @@ -919,6 +922,11 @@ 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. +The ``SHORTEST_PATHS_*`` presets model hop-by-hop IP/IGP forwarding: routes +follow link costs alone and each demand is placed in one pass on the +cost-only shortest-path DAG. The ``TE_*`` presets model a controller that +selects paths with knowledge of residual capacity. + ### 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. @@ -927,7 +935,7 @@ Args: algorithms: NetGraph-Core Algorithms instance. graph: NetGraph-Core Graph handle. preset: Preset whose path algorithm, placement, edge selection, and - flow-count bounds to apply. + flow-count bounds to apply (see ``preset_config``). 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 @@ -946,6 +954,28 @@ Example: >>> graph = algs.build_graph(strict_multidigraph) >>> policy = create_flow_policy(algs, graph, FlowPolicyPreset.SHORTEST_PATHS_ECMP) +### preset_config(preset: 'FlowPolicyPreset') -> 'netgraph_core.FlowPolicyConfig' + +Build the Core ``FlowPolicyConfig`` a preset stands for. + +This is the single source of the preset semantics. The SPF-cached +placement engine reads ``selection`` and ``flow_placement`` from it, and +``create_flow_policy`` materializes it as a Core ``FlowPolicy``. + +Hop-by-hop presets set ``require_capacity=False`` (routes follow costs +only) and ``shortest_path=True`` (one placement on the cost-only DAG), so +a FlowPolicy built from them places exactly what the cached engine +places. + +Args: + preset: Preset to describe. + +Returns: + A fresh ``FlowPolicyConfig``; callers may adjust it further. + +Raises: + ValueError: If an unknown FlowPolicyPreset value is provided. + ### serialize_policy_preset(cfg: 'Any') -> 'Optional[str]' Serialize a FlowPolicyPreset to its string name for JSON storage. @@ -1739,7 +1769,7 @@ Attributes: max_bracket_iters: Maximum iterations for bracketing phase. max_bisect_iters: Maximum iterations for bisection phase. placement_rounds: Deprecated; accepted for backward compatibility but - has no effect (placement optimization is handled by the core engine). + has no effect (each demand is placed in one deterministic pass). **Attributes:** @@ -1879,9 +1909,13 @@ Attributes: failure_policy: Failure policy name in scenario.failure_policy_set. If None, no failure policy is applied. iterations: Number of failure iterations to run; must be >= 0. - parallelism: Worker thread count, or "auto" for the CPU count. + parallelism: Worker thread count, or "auto". Auto uses the CPU count + when iterations can run concurrently (an LSP preset in the demand + set, or a free-threaded interpreter) and 1 otherwise, because + cacheable presets are Python-bound under the GIL and threads only + slow them down. See ``resolve_placement_parallelism``. placement_rounds: Deprecated; accepted for backward compatibility but - has no effect (placement optimization is handled by the core engine). + has no effect (each demand is placed in one deterministic pass). seed: Optional seed for reproducibility. store_failure_patterns: Record the failure trace on each result. Iterations are deduplicated, so a trace describes the first @@ -1916,6 +1950,30 @@ Attributes: - `execute(self, scenario: "'Scenario'") -> 'None'` - Execute the workflow step with logging and metadata storage. - `run(self, scenario: "'Scenario'") -> 'None'` - Execute the workflow step logic. +### resolve_placement_parallelism(parallelism: 'int | str', demands: 'Iterable[TrafficDemand]') -> 'int' + +Resolve the worker count for demand placement iterations. + +An explicit integer is used as given. ``"auto"`` becomes the CPU count +only when iterations can run concurrently: on a free-threaded interpreter, +or when the demand set uses a preset outside ``CACHEABLE_PRESETS`` (the +LSP presets), whose placement runs inside the core engine with the GIL +released. Iterations for cacheable presets are dominated by Python-side +work between very short engine calls, so on a GIL interpreter threads +only add contention and ``"auto"`` resolves to 1. + +Args: + parallelism: Positive worker count or ``"auto"``. + demands: Demands of the set to place; unset presets count as the + default ``SHORTEST_PATHS_ECMP``. + +Returns: + Positive worker count. + +Raises: + ValueError: If ``parallelism`` is neither a positive integer nor + ``"auto"``. + --- ## ngraph.dsl.blueprints.expand @@ -2960,6 +3018,10 @@ Attributes: policy_preset: FlowPolicy configuration preset. static_paths: Routes this demand is pinned to, empty when it is routed by the policy. + src_members: Real source node names behind a combine-mode pseudo + source, in selection order; empty for pairwise demands. Hop-by-hop + presets originate an even share of the volume at each member that + can reach a target instead of routing from the pseudo source. **Attributes:** @@ -2969,6 +3031,7 @@ Attributes: - `priority` (int) - `policy_preset` (FlowPolicyPreset) - `static_paths` (Tuple[StaticPath, ...]) = () +- `src_members` (Tuple[str, ...]) = () ### expand_demands(network: 'Network', traffic_demands: 'List[TrafficDemand]', default_policy_preset: 'FlowPolicyPreset' = ) -> 'DemandExpansion' @@ -3131,14 +3194,23 @@ Steps: pre-computed expansion) -3. Place each demand using SPF caching for cacheable policies. +3. Place each demand using SPF caching for cacheable policies, in priority - SHORTEST_PATHS_* presets admit flow onto the cost-only shortest paths - of the base topology and drop overflow (IGP semantics); TE_* presets + order and input order within a priority; nothing is revisited. + SHORTEST_PATHS_* presets place in one pass on the cost-only shortest + paths of the base topology: ``_ECMP`` admits what the equal-cost next + hops carry without loss, ``_ECMP_LOSSY`` delivers what survives + per-link drops, ``_WCMP`` splits by residual capacity. A combine-mode + demand is a virtual source: with one of these presets every source + that can reach a target originates an even share, and under + ``_ECMP`` the pool is admitted at one global scale. TE_* presets reroute remaining volume onto residual-capacity paths. 4. Fall back to FlowPolicy for presets outside CACHEABLE_PRESETS -5. Aggregate results into FlowIterationResult +5. Aggregate results into FlowIterationResult. With + + ``include_flow_details`` a lossy demand's entry carries + ``data["dropped_edges"]``, the dropped volume per ``link_id:direction``. SPF Caching Optimization: For cacheable policies (ECMP, WCMP, TE_WCMP_UNLIM), SPF results are @@ -3230,6 +3302,19 @@ Core demand placement with SPF caching. Single demand placement result. +Attributes: + src_name: Source node name (real or pseudo). + dst_name: Destination node name (real or pseudo). + priority: Priority class. + volume: Requested volume. + placed: Placed volume. For ``SHORTEST_PATHS_ECMP_LOSSY`` this is the + volume delivered to the destination. + cost_distribution: Placed volume by path cost, when requested. + used_edges: ``link_id:direction`` of every edge carrying this demand, + when requested. + dropped_edges: Dropped volume by ``link_id:direction``, when requested; + only lossy presets drop. + **Attributes:** - `src_name` (str) @@ -3239,6 +3324,7 @@ Single demand placement result. - `placed` (float) - `cost_distribution` (dict[float, float]) = {} - `used_edges` (set[str]) = set() +- `dropped_edges` (dict[str, float]) = {} ### PlacementResult @@ -3258,15 +3344,40 @@ Attributes: Aggregated placement totals. +Attributes: + total_demand: Sum of demand volumes. + total_placed: Sum of placed volumes. + max_shortfall: Largest ``volume - placed`` over all demands. + unserved_demands: Demands with positive volume that placed nothing. + **Attributes:** - `total_demand` (float) - `total_placed` (float) +- `max_shortfall` (float) = 0.0 +- `unserved_demands` (int) = 0 -### place_demands(demands: "Sequence['ExpandedDemand']", volumes: 'Sequence[float]', flow_graph: 'netgraph_core.FlowGraph', ctx: "'AnalysisContext'", node_mask: 'np.ndarray', edge_mask: 'np.ndarray', *, resolved_ids: 'Sequence[tuple[int, int]] | None' = None, collect_entries: 'bool' = False, include_cost_distribution: 'bool' = False, include_used_edges: 'bool' = False, dag_cache: 'dict[tuple[int, bool], tuple[np.ndarray, Any]] | None' = None) -> 'PlacementResult' +### place_demands(demands: "Sequence['ExpandedDemand']", volumes: 'Sequence[float]', flow_graph: 'netgraph_core.FlowGraph', ctx: "'AnalysisContext'", node_mask: 'np.ndarray', edge_mask: 'np.ndarray', *, resolved_ids: 'Sequence[tuple[int, int]] | None' = None, collect_entries: 'bool' = False, include_cost_distribution: 'bool' = False, include_used_edges: 'bool' = False, dag_cache: 'dict[tuple, tuple[np.ndarray, Any]] | None' = None) -> 'PlacementResult' Place demands on a flow graph with SPF caching. +Demands are placed one at a time in the given order (callers sort by +priority), each seeing the residual left by the ones before it. Nothing +is revisited, so within a priority class earlier demands win contended +capacity and the totals of rerouting presets depend on demand order. + +Hop-by-hop presets (``HOP_BY_HOP_PRESETS``) place each demand in one pass +on the cost-only shortest-path DAG of its source. A combine-mode demand is +a virtual source, a pool of the selected sources: with such a preset +(``ExpandedDemand.src_members`` set) every member that can reach a target +originates an even share of the volume, since hop-by-hop routing has no +controller that could choose where traffic originates, and each share is +routed to that member's nearest targets. Under lossless ECMP the pool is +admitted as one demand at a single scale. TE presets keep the aggregated +pseudo source and let capacity decide which members originate. A fixed +per-source matrix is a different question, answered by pairwise or +per-group expansion. + Args: demands: Expanded demands (policy_preset, priority, names). volumes: Volume per demand, positionally aligned with `demands`; @@ -3280,11 +3391,13 @@ Args: resolved_ids: Pre-resolved (src_id, dst_id) pairs. Computed from the demand names if None. collect_entries: If True, populate result.entries. - include_cost_distribution: Include cost distribution in entries. + include_cost_distribution: Include cost distribution and, for lossy + presets, dropped volume per link in entries. include_used_edges: Include used edges in entries. - dag_cache: Optional persistent SPF DAG cache keyed by - (src_id, uses_capacity_aware_selection). Base DAGs depend only on - the static graph and masks, so repeated calls with the same + dag_cache: Optional persistent SPF DAG cache. Base DAGs are keyed by + ``(src_id, uses_capacity_aware_selection)`` and combine-mode + fan-out DAGs by ``(dst_id, "fanout")``. All of them depend only + on the static graph and masks, so repeated calls with the same context and masks (e.g. MSD probes) can share one cache. Returns: diff --git a/docs/reference/api.md b/docs/reference/api.md index ce9943e..ac6eefc 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -8,7 +8,7 @@ Quick links: - [CLI Reference](cli.md) -- command-line tools for running scenarios - [Auto-Generated API Reference](api-full.md) -- complete class and method documentation -A curated guide to NetGraph's Python API, organized by typical usage patterns. +The Python API, organized by how it is typically used. ## 1. Programmatic Quickstart @@ -203,7 +203,7 @@ Max-flow, shortest paths, and edge sensitivity. **When to use:** Measuring capacity between source and sink groups, under a choice of flow placement policy and with nodes or links excluded to model failures. -**Performance:** Max-flow computation executes in C++ with the GIL released for concurrent execution. The algorithm uses successive shortest paths on the residual graph, pushing a blocking flow across the full ECMP/WCMP shortest-path DAG at each augmentation step until no augmenting path remains. Worst case is `O(E * (V^2 E + (V+E) log V))`; in practice the phase count equals the small number of cost tiers actually used, and the `kMinFlow` tolerance caps phases at `F / kMinFlow` for total flow `F`. See [Design](design.md) for the derivation. +Max-flow runs in C++ with the GIL released. The algorithm and its complexity bounds are described in [Design](design.md). ```python from ngraph import analyze, Mode, FlowPlacement @@ -239,7 +239,7 @@ print(summary.cost_distribution) # Dict[float, float] mapping cost to flow volu - **Mode.COMBINE:** Aggregate sources into one super-source, sinks into one super-sink; returns single total flow - **Mode.PAIRWISE:** Compute flow for each (source_group, sink_group) pair independently -- **FlowPlacement.PROPORTIONAL (WCMP):** Split flow proportional to edge capacity +- **FlowPlacement.PROPORTIONAL (WCMP):** Split flow across parallel edges in proportion to residual capacity - **FlowPlacement.EQUAL_BALANCED (ECMP):** Equal split across parallel paths - **shortest_path=True:** Restricts flow to lowest-cost paths only (IP/IGP routing semantics) - **shortest_path=False:** Uses all paths progressively (TE/SDN semantics) @@ -489,6 +489,9 @@ entry.destination # Destination label entry.demand # Requested demand entry.placed # Actually placed entry.dropped # Unmet demand +entry.cost_distribution # Dict[cost, placed volume] with include_flow_details +entry.data # Optional details: edges/edges_kind with include_used_edges, + # dropped_edges (volume lost per link) for lossy presets # FlowSummary - Aggregated statistics summary.total_demand # Sum of all demands @@ -547,12 +550,6 @@ for pair, impacts in sensitivity.items(): ## 9. Performance Notes -NetGraph uses a hybrid Python+C++ architecture: +Network, Scenario and the workflow steps are Python. Shortest paths, max-flow and k-shortest paths run in C++ (NetGraph-Core) with the GIL released. Public APIs take and return Python types; the C++ layer is only reached through `netgraph_core` when you call it yourself, as in the NetworkX section above. -- **High-level APIs** (Network, Scenario, Workflow) are pure Python -- **Core algorithms** (shortest paths, max-flow, K-shortest paths) execute in optimized C++ via NetGraph-Core -- **GIL released** during algorithm execution for parallel processing -- **Transparent integration**: You work with Python objects; Core acceleration is automatic - -All public APIs accept and return Python types (Network, Node, Link, FlowSummary, etc.). -The C++ layer is an implementation detail you generally don't interact with directly. +Threads help only when an iteration spends its time inside the C++ engine (max-flow, the LSP presets). Demand placement for the hop-by-hop presets is Python-bound between short engine calls, which is why `TrafficMatrixPlacement` resolves `parallelism: auto` to 1 for those demand sets. diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 6d6ddc2..1b6ed52 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -82,13 +82,6 @@ ngraph inspect scenarios/nsfnet.yaml --detail ngraph --verbose inspect scenarios/square_mesh.yaml ``` -**Use cases:** - -- **Scenario validation**: Verify YAML syntax and structure -- **Network debugging**: Analyze blueprint expansion and node/link creation -- **Capacity analysis**: Review network capacity distribution and connectivity -- **Workflow preview**: Examine analysis steps before execution - ### `run` Execute a NetGraph scenario file. @@ -173,19 +166,7 @@ ngraph run scenarios/backbone_clos.yml --profile --results analysis.json ngraph run scenarios/backbone_clos.yml --profile --profile-memory --keys tm_placement ``` -The profiling output includes: - -- **Summary**: Total execution time, CPU efficiency, function call statistics -- **Step timing**: Time spent in each workflow step with percentage breakdown -- **Bottlenecks**: Steps consuming >10% of total execution time -- **Function analysis**: Top CPU-consuming functions within bottlenecks -- **Recommendations**: Specific suggestions for each bottleneck - -**When to use profiling:** - -- Performance analysis during development -- Identifying bottlenecks in complex workflows -- Benchmarking before/after changes +The report lists total execution time, time per step, the steps that take more than 10% of the total, and the top CPU-consuming functions within those steps. ### Output Format @@ -209,85 +190,23 @@ The CLI outputs results as JSON with a fixed top-level shape: ## Output Behavior -NetGraph CLI generates results by default for analysis workflows: - -### Default Behavior (Results Generated) - -```bash -ngraph run scenarios/square_mesh.yaml -``` - -- Executes the scenario -- Logs execution progress to the terminal -- **Creates `.results.json` by default** -- Shows success message with file location - -### Custom Results File - -```bash -# Save to custom file -ngraph run scenarios/square_mesh.yaml --results my_analysis.json -``` - -- Creates specified JSON file instead of the default `.results.json` -- Useful for organizing multiple analysis runs +| Command | Writes | Prints JSON | +|---------|--------|-------------| +| `ngraph run scenario.yaml` | `.results.json` | no | +| `ngraph run scenario.yaml --results out.json` | `out.json` | no | +| `ngraph run scenario.yaml --stdout` | `.results.json` | yes | +| `ngraph run scenario.yaml --results out.json --stdout` | `out.json` | yes | +| `ngraph run scenario.yaml --no-results` | nothing | no | -### Print to Terminal +Logs and status messages go to stderr in every case. -```bash -ngraph run scenarios/square_mesh.yaml --stdout -``` +## Debugging Scenarios -- Creates `.results.json` AND prints JSON to stdout -- Useful for viewing results immediately while also saving them - -### Combined Output +`ngraph run` executes every workflow step in order. Inspect a scenario before running it, and use `--verbose` with `--detail` when blueprint expansion does not produce the nodes or links you expect: ```bash -ngraph run scenarios/square_mesh.yaml --results analysis.json --stdout -``` - -- Creates custom JSON file AND prints to stdout - -### Disable File Generation (Edge Cases) - -```bash -ngraph run scenarios/square_mesh.yaml --no-results -``` - -- Executes scenario without creating any output files -- Only shows execution logs and completion status -- Useful for testing, CI/CD validation, or when only logs are needed - -## Integration with Workflows - -`ngraph run` executes every step of the workflow defined in the scenario file, in sequence, accumulating results as it goes. - -### Recommended Workflow - -1. **Inspect first**: Always use `inspect` to validate and understand your scenario -2. **Debug issues**: Use detailed inspection to troubleshoot network expansion problems -3. **Run after validation**: Execute scenarios after successful inspection -4. **Iterate**: Use inspection during scenario development to verify changes - -```bash -# Development workflow -ngraph inspect scenarios/backbone_clos.yml --detail -ngraph run scenarios/backbone_clos.yml -``` - -### Debugging Scenarios - -When developing complex scenarios with blueprints and hierarchical structures: - -```bash -# Check if scenario loads correctly ngraph inspect scenarios/square_mesh.yaml - -# Debug network expansion issues (note: global option placement) ngraph --verbose inspect scenarios/backbone_clos.yml --detail - -# Verify workflow steps are configured correctly ngraph inspect scenarios/backbone_clos.yml --detail | grep -A 5 "WORKFLOW STEPS" ``` diff --git a/docs/reference/design.md b/docs/reference/design.md index 9e560ba..8a44aa2 100644 --- a/docs/reference/design.md +++ b/docs/reference/design.md @@ -72,7 +72,7 @@ The Python layer uses the `analyze()` function and `AnalysisContext` class (`ngr 3. Execute analysis methods (max_flow, shortest_paths, sensitivity) with boolean masking 4. Translate results (costs, flows, paths) back to scenario-level objects -Core algorithms release the GIL during execution, so concurrent Python threads run analysis in parallel with minimal Python-level overhead. +Core algorithms release the GIL, so Python threads can run them concurrently. Whether that helps depends on how much of an iteration is spent inside Core; see Failure Manager below. **Primary API:** @@ -317,7 +317,7 @@ are not mapped back to scenario links in results. NetGraph's core algorithms execute in C++ via NetGraph-Core. They operate on the immutable StrictMultiDiGraph and support masking (runtime exclusions via boolean arrays), so repeated analysis under different failure scenarios needs no graph reconstruction. -All Core algorithms release the Python GIL during execution, so multiple Python threads run them concurrently without GIL contention. +All Core algorithms release the Python GIL while they run. ### Shortest-Path First (SPF) Algorithm @@ -535,9 +535,9 @@ Traditional IP routing with Interior Gateway Protocols (OSPF, IS-IS): - Routes computed based on link costs/metrics only, ignoring available capacity - Single SPF computation determines equal-cost paths; forwarding is fixed until topology/cost change -- Traffic follows predetermined paths even as links saturate -- Models best-effort forwarding with potential packet loss when demand exceeds capacity +- Traffic follows predetermined paths even as links saturate: a next hop that an earlier demand filled stays in the hash, it does not get skipped - No iterative augmentation: flow placed in single pass over fixed equal-cost DAG +- Two readings of what happens when demand exceeds capacity, chosen per demand by preset: lossless admission (`SHORTEST_PATHS_ECMP`: how much can be carried with no drops, the planning question) or best-effort delivery (`SHORTEST_PATHS_ECMP_LOSSY`: how much arrives when every link carries what fits and drops the rest, the failure-impact question) - Use case: Simulating production IP networks, validating IGP designs **SDN/TE Semantics (`require_capacity=true` + `shortest_path=false`, default):** @@ -549,7 +549,8 @@ Software-Defined Networking and Traffic Engineering: - Iterative augmentation continues until max-flow achieved or capacity exhausted - Flow placement respects capacity constraints, never oversubscribing links - Models centralized traffic engineering with real-time capacity awareness -- Use case: Optimal demand placement, capacity planning, failure impact analysis +- Placement is greedy and sequential: demands are placed one at a time in priority order (input order within a priority) and never revisited, so an earlier demand keeps capacity a later one would have used, and totals under contention depend on demand order. It is not a global optimum +- Use case: Capacity-aware demand placement, capacity planning, failure impact analysis This distinction is fundamental: IP networks route on cost alone with fixed forwarding tables (congestion managed via queuing/drops), while TE systems route dynamically on both cost and available capacity (congestion avoided via admission control). The `require_capacity` parameter controls whether SPF filters to available capacity; `shortest_path` controls whether routes are recomputed iteratively or fixed after initial SPF. @@ -569,8 +570,19 @@ Beyond routing semantics, NetGraph controls how flow splits across equal-cost pa - Example: Two 100G links get 50/50; one 100G + one 10G still attempt 50/50 (10G saturates first) - Models IP hash-based load balancing (5-tuple hashing distributes flows uniformly) - Single-pass admission: computes one global scale factor to avoid oversubscription + - The split set is the DAG edges that still have residual, so a later placement on the same DAG hashes only over the members that are not yet full (progressive behaviour, which is what `place_max_flow` and the LSP policies rely on) - For IP ECMP simulation: use with `require_capacity=false` + `shortest_path=true` +- **EQUAL_BALANCED_FIXED** (lossless ECMP admission with a load-blind forwarding table): + - Same equal split and global scale, but the split set is every DAG edge with capacity, saturated or not + - A member filled by an earlier demand therefore drives the scale to 0: any further traffic hashed onto it would be lost, so nothing more is admitted losslessly + - Backs the `SHORTEST_PATHS_ECMP` preset + +- **EQUAL_BALANCED_LOSSY** (best-effort ECMP forwarding): + - Same split set, no scale: every edge carries `min(share, residual)` and drops the rest; a deficit propagates downstream and the placed amount is what reaches the sink + - Example: one 100G + one 10G offered 100G deliver 60G and drop 40G on the 10G link + - The dropped volume per edge is reported (`FlowGraph.place_with_drops`); backs the `SHORTEST_PATHS_ECMP_LOSSY` preset + `FlowState.place_on_dag` implements placement over a fixed SPF DAG (the DAG never changes within a call): - **PROPORTIONAL**: Constructs reversed residual graph from predecessor DAG. Uses Dinic-style BFS leveling and DFS push from sink to source. Within each edge group (parallel edges between node pair), splits flow proportionally to residual capacity. Distributes pushed flow back to underlying edges maintaining proportional ratios. Can be called iteratively on updated residuals. @@ -605,28 +617,33 @@ For traffic matrix placement, `FlowPolicyPreset` values bundle the routing seman | Preset | Behavior | Use Case | | -------- | ---------- | ---------- | -| `SHORTEST_PATHS_ECMP` | IP/IGP with hash-based ECMP | Traditional routers (OSPF/IS-IS), equal splits across equal-cost paths | -| `SHORTEST_PATHS_WCMP` | IP/IGP with weighted ECMP | Routers with WCMP support, proportional splits based on link capacity | +| `SHORTEST_PATHS_ECMP` | IP/IGP with hash-based ECMP, lossless admission | Traditional routers (OSPF/IS-IS); `placed` is what the network carries with no drops | +| `SHORTEST_PATHS_ECMP_LOSSY` | IP/IGP with hash-based ECMP, best-effort delivery | Same routers under overload; `placed` is what arrives, `dropped` is lost on the way, per-link drops reported with flow details | +| `SHORTEST_PATHS_WCMP` | IP/IGP with weighted ECMP | Routers with WCMP support, proportional splits by residual capacity (equal to link capacity on an unloaded network) | | `TE_WCMP_UNLIM` | MPLS-TE / SDN with WCMP | Capacity-aware TE with unlimited tunnels, iterative placement | | `TE_ECMP_16_LSP` | MPLS-TE with 16 LSPs | Fixed 16 ECMP tunnels per demand, models RSVP-TE with LSP limits | | `TE_ECMP_UP_TO_256_LSP` | MPLS-TE with up to 256 LSPs | Scalable TE with tunnel limit, models SR-TE or large-scale RSVP | **Detailed Configuration Mapping (preset internals):** -| Preset | `require_capacity` | `multi_edge` | `max_flow_count` | `flow_placement` | -| -------- | -------------------- | -------------- | ------------------ | ------------------ | -| `SHORTEST_PATHS_ECMP` | `false` | `true` | `1` | `EQUAL_BALANCED` | -| `SHORTEST_PATHS_WCMP` | `false` | `true` | `1` | `PROPORTIONAL` | -| `TE_WCMP_UNLIM` | `true` | `true` | unlimited | `PROPORTIONAL` | -| `TE_ECMP_16_LSP` | `true` | `false` | `16` | `EQUAL_BALANCED` | -| `TE_ECMP_UP_TO_256_LSP` | `true` | `false` | `256` | `EQUAL_BALANCED` | +`ngraph.model.flow.policy_config.preset_config` is the single source of these values; the SPF-cached placement engine and Core's `FlowPolicy` both read from it. + +| Preset | `require_capacity` | `shortest_path` | `multi_edge` | `max_flow_count` | `flow_placement` | +| -------- | -------------------- | ----------------- | -------------- | ------------------ | ------------------ | +| `SHORTEST_PATHS_ECMP` | `false` | `true` | `true` | `1` | `EQUAL_BALANCED_FIXED` | +| `SHORTEST_PATHS_ECMP_LOSSY` | `false` | `true` | `true` | `1` | `EQUAL_BALANCED_LOSSY` | +| `SHORTEST_PATHS_WCMP` | `false` | `true` | `true` | `1` | `PROPORTIONAL` | +| `TE_WCMP_UNLIM` | `true` | `false` | `true` | unlimited | `PROPORTIONAL` | +| `TE_ECMP_16_LSP` | `true` | `false` | `false` | `16` | `EQUAL_BALANCED` | +| `TE_ECMP_UP_TO_256_LSP` | `true` | `false` | `false` | `256` | `EQUAL_BALANCED` | **Key parameters (preset-managed):** - `require_capacity`: When `false`, paths are selected based on link costs alone (models IP/IGP routing). When `true`, paths adapt to residual capacity during placement (models SDN/TE). See [Routing Semantics](#routing-semantics-ipigp-vs-sdnte) for details. +- `shortest_path`: When `true`, each demand is placed in a single pass on its cost-only shortest-path DAG and the remainder is dropped; when `false`, the remainder is rerouted tier by tier on residual-aware paths. - `multi_edge`: When `true`, uses all parallel equal-cost edges (hop-by-hop ECMP); when `false`, each flow uses a single path (tunnel/LSP semantics). - `max_flow_count`: Internal per-preset limit on flows/LSPs for TE presets; not a user-facing parameter. -- `flow_placement`: `EQUAL_BALANCED` splits equally across paths; `PROPORTIONAL` splits by residual capacity. +- `flow_placement`: `EQUAL_BALANCED_FIXED` splits equally over the topology's next hops and admits losslessly; `EQUAL_BALANCED_LOSSY` splits the same way and drops what does not fit; `EQUAL_BALANCED` splits equally over next hops with headroom (progressive, used by the LSP presets); `PROPORTIONAL` splits by residual capacity. **Example: Modeling IP vs MPLS Networks** @@ -721,7 +738,7 @@ The flow tolerance constant `kMinFlow` (1/4096 ≈ 2.4e-4) determines when flow 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\). +In practice the loop stops as soon as the residual network disconnects source from sink, the number of phases equals the number of cost tiers actually used, and the `kMinFlow` threshold caps phases at \(F / k_{MinFlow}\) for total flow \(F\). ### Managers and Workflow Orchestration @@ -731,21 +748,24 @@ Managers handle scenario dynamics and prepare inputs for algorithmic steps. - Deterministic expansion: node selection follows the network's stable node ordering; no randomization - Supports `combine` mode (aggregate via pseudo source/sink nodes attached with large-capacity, zero-cost augmentation edges) and `pairwise` mode (individual (src,dst) pairs, self-pairs excluded, volume split evenly across pairs) +- In `combine` mode the pseudo source is a virtual source, a pool of the selected sources. A TE preset lets capacity decide which members originate: the aggregate is carried by whichever sources have room. Hop-by-hop presets (`SHORTEST_PATHS_*`) cannot steer where traffic originates, so every member that can reach a target originates an even share of the volume and routes it to its nearest targets through the pseudo sink; a member with no path to any target is not part of the split, and the result is still reported as one demand. A plain SPF from the pseudo source would keep only the sources closest to the targets, which is not a routing property of the network, so the DAG is built with Core's reverse SPF (`spf_to`) toward the pseudo sink with every attachment edge forced into it +- Under `SHORTEST_PATHS_ECMP` the pool is admitted as one demand: the fan-out DAG is placed in a single equal-balanced pass, so the split ratios stay fixed and one global scale applies. A member that can carry only part of its share throttles every member alike: two equal-cost sources of capacity 100 and 10 admit 20 of a 110 demand. `SHORTEST_PATHS_ECMP_LOSSY` and `SHORTEST_PATHS_WCMP` carry each member's share independently (65 of 110 in that example); their totals and per-link drops equal those of a simultaneous pass, and shared links are attributed to members in selection order +- A fixed per-source traffic matrix, where each source's share is its own and an isolated source's share is unserved rather than moved to the others, is a different question: expand per source with `group_mode: per_group` and `group_by: name`, or use `pairwise` - `group_mode` controls grouping: `flatten` (default, merge all groups then apply mode), `per_group`, and `group_pairwise`; volume is split evenly across groups or group pairs - In combine mode, nodes selected on both sides are excluded from the target set (prevents a zero-cost pseudo-node bypass); a demand or group whose target set empties is skipped - Validation: duplicate demand ids and pseudo-endpoint collisions raise `ValueError` (either would silently merge distinct demands' attachment edges) -- Demands sorted by ascending priority before placement (lower value = higher priority) -- Placement uses SPF caching for simple policies (ECMP, WCMP, TE_WCMP_UNLIM), FlowPolicy for complex multi-flow policies +- Demands sorted by ascending priority before placement (lower value = higher priority); within a priority they keep input order, and each demand sees the residual left by the ones before it. Nothing is revisited, so contended capacity goes to the earlier demand and the totals of rerouting presets depend on demand order +- Placement uses SPF caching for the hop-by-hop presets and `TE_WCMP_UNLIM`, FlowPolicy for the LSP presets - Non-mutating: operates on Core flow graphs with exclusions; Network remains unmodified **Failure Manager** (`ngraph.analysis.failure_manager`): Applies a `FailurePolicy` to compute exclusion sets and runs analyses with those exclusions. -- Parallel execution via `ThreadPoolExecutor` with zero-copy network sharing across worker threads; nothing is pickled, so functions defined in `__main__` or notebooks run at full parallelism +- Parallel execution via `ThreadPoolExecutor` with zero-copy network sharing across worker threads; nothing is pickled, so functions defined in `__main__` or notebooks can run in parallel. Threads only pay off when an iteration spends most of its time inside the core engine with the GIL released (max-flow, the LSP presets): a demand placement iteration for the hop-by-hop presets or `TE_WCMP_UNLIM` is mostly Python work around microsecond engine calls, and adding threads makes it slower on a GIL interpreter. `TrafficMatrixPlacement` therefore resolves `parallelism: auto` to 1 for such demand sets and to the CPU count when an LSP preset is present or the interpreter is free-threaded - Deterministic results when seed is provided (each iteration derives `seed + iteration_index`); with `seed=None`, the failure policy's own seed is used as a fallback when present - Baseline execution: a no-failure baseline is always run first as a separate reference for comparing degraded vs. intact capacity - Deduplication: identical exclusion patterns execute once and are weighted by multiplicity (`occurrence_count` on results; `metadata["occurrence_counts"]` aligned with the results list). Stored failure traces describe each pattern's representative (first) iteration; this weighting assumes deterministic analysis functions (the built-ins are) - Thread-safe analysis: Network shared by reference; exclusion sets passed per-iteration -- Automatic graph pre-building: Before parallel iterations, the engine calls the analysis function's `prepare_inputs(network, kwargs)` hook (carried by all built-in analysis functions) once per run and merges the returned kwargs — typically a pre-built `AnalysisContext`, plus the precomputed demand expansion and resolved IDs for demand placement — into every iteration's call; per-iteration exclusions are applied via O(|excluded|) mask operations. Custom analysis functions opt in by setting a `prepare_inputs` attribute; functions without it run with their kwargs unchanged, and passing `context` explicitly skips the hook. +- Graph pre-building: before the iterations start, the engine calls the analysis function's `prepare_inputs(network, kwargs)` hook once per run and merges what it returns into every iteration's call. The built-in functions return a pre-built `AnalysisContext`, and demand placement also returns the expanded demands and resolved node ids. Each iteration then only builds masks, O(|excluded|). Custom analysis functions opt in by setting a `prepare_inputs` attribute; functions without it run with their kwargs unchanged, and passing `context` explicitly skips the hook. Both the demand expansion logic and failure manager separate policy (how to expand demands or pick failures) from core algorithms. They prepare concrete inputs (expanded demands or exclusion sets) for each workflow iteration. @@ -779,9 +799,9 @@ This design ensures consistency (every step has metadata and data keys) and JSON ### Design Elements and Comparisons -NetGraph's design includes several features that differentiate it from traditional network analysis tools: +Design choices worth knowing about: -- Declarative Scenario DSL: A YAML DSL with blueprints and programmatic expansion allows abstract definitions (e.g., a fully meshed Clos) to be expanded into concrete nodes and links. Strict schema validation ensures that scenarios are well-formed and rejects unknown or invalid fields. +- Declarative Scenario DSL: A YAML DSL with blueprints and expansion rules turns abstract definitions (e.g., a fully meshed Clos) into concrete nodes and links. Schema validation rejects unknown or invalid fields before expansion. - Runtime Exclusions vs graph copying: Analysis-time exclusions avoid copying large structures for each scenario. The design separates static topology from dynamic failure states. @@ -820,12 +840,13 @@ Graph construction involves Python processing, NumPy array creation, and C++ obj **SPF Caching for Demand Placement:** -Both TrafficMatrixPlacement and MaximumSupportedDemand (MSD) use a unified placement function (`place_demands()` in `ngraph.analysis.placement`) with SPF caching for cacheable policies (ECMP, WCMP, TE_WCMP_UNLIM): +Both TrafficMatrixPlacement and MaximumSupportedDemand (MSD) use a unified placement function (`place_demands()` in `ngraph.analysis.placement`) with SPF caching for cacheable policies (the `SHORTEST_PATHS_*` presets and `TE_WCMP_UNLIM`): -- Initial SPF computed once per unique source; subsequent demands from the same source reuse the cached DAG -- For TE policies, DAG is recomputed when capacity constraints require alternate paths +- One SPF per unique source per mask state; later demands from the same source reuse the cached DAG. A combine-mode demand under `SHORTEST_PATHS_ECMP` uses one reverse SPF toward its pseudo sink instead; under the other hop-by-hop presets it is placed per source and shares each source's cache entry +- For TE policies, DAG is recomputed when capacity constraints require alternate paths. Each recomputation either saturates an edge of the tier it placed on or makes no progress, so the loop is bounded by the edge count rather than by a fixed iteration cap - Complex multi-flow policies (TE_ECMP_16_LSP, TE_ECMP_UP_TO_256_LSP) use FlowPolicy directly - MSD additionally pre-resolves node IDs once at cache build time and reuses them across all alpha probes +- MSD calls a scale feasible when every demand is placed to the engine's numeric resolution (a shortfall of at most 1/4096, the smallest flow the engine places) and no demand with volume placed nothing; an exact-match rule would call a demand spread over many LSPs infeasible at every scale This reduces SPF computations from O(demands) to O(unique_sources) for workloads where many demands share the same source nodes. MSD gains the most, since it evaluates many alpha values during binary search. @@ -837,44 +858,9 @@ common failure policies. **Complexity:** -- SPF: \(O((V+E) \log V)\) using binary heap -- Max-flow: \(O(E \cdot (V^2 E + (V+E) \log V))\) worst case for the successive-shortest-paths scheme with blocking-flow placement (derived in "Maximum Flow Algorithm" above) - - Practical performance is far better: the number of augmentation phases equals the small number of cost tiers actually used, and the `kMinFlow` threshold caps phases at \(F / k_{MinFlow}\) for total flow \(F\) - - Early termination when the residual network disconnects source from sink provides significant speedup in typical networks - -**Scalability:** - -Benchmarks on structured topologies (Clos, grid) and realistic network graphs demonstrate scalability to networks with thousands of nodes and tens of thousands of edges. C++ execution with CSR adjacency and GIL release provides order-of-magnitude speedups over pure Python graph libraries for compute-intensive analysis. - -## Summary - -NetGraph's hybrid architecture combines: - -**Python Layer:** - -- Declarative scenario DSL with schema validation -- Domain model (Network, Node, Link, RiskGroup) -- Runtime exclusions for non-destructive failure simulation -- Workflow orchestration and result aggregation -- Managers for demand expansion and failure enumeration - -**C++ Layer:** - -- Native C++ graph algorithms (SPF, K-shortest paths, max-flow) -- Immutable StrictMultiDiGraph with CSR adjacency -- Configurable flow placement policies (ECMP/WCMP simulation) -- Runtime masking for repeated analysis without graph rebuilds - -**Integration:** - -- `AnalysisContext` builds Core graphs, manages name/ID mapping, and bridges Python ↔ C++ -- Stable node/edge ID mapping for result traceability -- Zero-copy NumPy array interface for data transfer -- GIL release during computation for concurrent thread execution - -This design adapts standard algorithms to network engineering use cases (flow splitting, -failure simulation, cost-aware routing), running them in native C++ while keeping the -scenario, workflow, and result interfaces in Python. +- SPF: \(O((V+E) \log V)\) with a binary heap +- Max-flow: \(O(E \cdot (V^2 E + (V+E) \log V))\) worst case; see "Maximum Flow Algorithm" for the derivation and the much smaller practical phase count +- Demand placement: one cached SPF per source per mask state plus one placement call per demand for the hop-by-hop presets; `TE_WCMP_UNLIM` adds one residual-aware SPF per cost tier it spills into; the LSP presets run Core's FlowPolicy per demand, which is one to two orders of magnitude more work per demand ## Cross-references diff --git a/docs/reference/dsl.md b/docs/reference/dsl.md index 6f61579..550f9ff 100644 --- a/docs/reference/dsl.md +++ b/docs/reference/dsl.md @@ -1290,10 +1290,12 @@ 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 + route's bottleneck never reaches the second; `SHORTEST_PATHS_ECMP` gives every surviving route the same share, so total placement is limited by the - smallest surviving route. Route order is therefore significant under - proportional presets. + smallest surviving route; `SHORTEST_PATHS_ECMP_LOSSY` offers every surviving + route the same share and each carries what fits, so `placed` is the + delivered total (per-link drops are not reported for pinned routes). 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. @@ -1321,7 +1323,11 @@ The `source` and `target` fields accept either: Controls how source and target node sets are paired: -- `combine`: Aggregate all sources into one virtual source, all targets into one virtual target. Produces a single flow. +- `combine`: Aggregate all sources into one virtual source and all targets into one virtual target. Produces a single flow entry. The virtual source is a pool of the selected sources: + - With a `TE_*` preset, capacity decides which sources originate the traffic. This answers "what is the largest volume this group can deliver to that group". + - With a `SHORTEST_PATHS_*` preset, every source that can reach a target originates an even share and routes it to its nearest targets, because hop-by-hop routing cannot choose where traffic originates. A source with no path (its links or node failed) drops out of the split. + - Under `SHORTEST_PATHS_ECMP` the pool is admitted as one demand with its split fixed, so a source that can carry only part of its share throttles the others too. Two equal-cost sources of capacity 100 and 10 admit 20 of a 110 demand without loss; `SHORTEST_PATHS_ECMP_LOSSY` delivers 65 and drops 45 at the small source. + - For a fixed per-source matrix, where an isolated source's share is unserved rather than moved to the others, expand per source (`group_by: name` with `group_mode: per_group`) or use `pairwise`. - `pairwise`: Create individual flows between all source-target node pairs. Volume is distributed across pairs. **Overlapping selections in `combine` mode:** Nodes selected by both `source` and `target` are excluded from the target side, so overlapping selections cannot route volume through a zero-cost pseudo-node bypass; placement is bounded by real network capacity. With `group_mode: flatten`, a demand whose source and target selections fully overlap leaves no targets after exclusion and expands to nothing; if no demand in the whole expansion produces anything, analysis fails with `No demands could be expanded`. @@ -1362,8 +1368,9 @@ In `per_group`, `group_pairwise`, and `pairwise` expansions the configured volum ### Flow Policies -- `SHORTEST_PATHS_ECMP`: IP/IGP routing with hash-based ECMP; equal split across equal-cost paths -- `SHORTEST_PATHS_WCMP`: IP/IGP routing with weighted ECMP; proportional split by link capacity +- `SHORTEST_PATHS_ECMP`: IP/IGP routing with hash-based ECMP; equal split across equal-cost paths, admitted without loss. `placed` is what the network carries with no drops; a next hop filled by an earlier demand blocks later demands hashed onto it +- `SHORTEST_PATHS_ECMP_LOSSY`: the same routing, forwarded best-effort. Every link carries what fits and drops the rest; `placed` is what arrives and `dropped` what was lost. With `include_flow_details` each entry reports `dropped_edges`, the lost volume per link +- `SHORTEST_PATHS_WCMP`: IP/IGP routing with weighted ECMP; proportional split by residual capacity (equal to link capacity on an unloaded network) - `TE_WCMP_UNLIM`: MPLS-TE / SDN with capacity-aware WCMP; unlimited tunnels - `TE_ECMP_16_LSP`: MPLS-TE with exactly 16 ECMP LSPs per demand - `TE_ECMP_UP_TO_256_LSP`: MPLS-TE with up to 256 ECMP LSPs per demand diff --git a/docs/reference/schemas.md b/docs/reference/schemas.md index cc0bdf9..0aba633 100644 --- a/docs/reference/schemas.md +++ b/docs/reference/schemas.md @@ -9,7 +9,7 @@ Quick links: - [API Reference](api.md) — Python API for programmatic scenario creation - [Auto-Generated API Reference](api-full.md) — complete class and method documentation -NetGraph includes JSON Schema definitions for YAML scenario files, providing IDE validation, autocompletion, and automated testing. +A JSON Schema describes the scenario YAML. It drives load-time validation, IDE completion, and tests. ## Schema Location @@ -26,7 +26,7 @@ The schema validates: - Top-level section organization - Basic constraint checking -Runtime: The schema is applied unconditionally during load in `ngraph.scenario.Scenario.from_yaml` (via `ngraph.dsl.loader.load_scenario_yaml`). Additional business rules are enforced in code (e.g., blueprint expansion) and may still raise errors for semantically invalid inputs. +`Scenario.from_yaml` always validates against the schema (in `ngraph.dsl.loader.load_scenario_yaml`) before expansion. Rules the schema cannot express, such as blueprint parameter names or risk-group references, are checked in code and raise `ValueError`. ## IDE Integration (VS Code) @@ -98,18 +98,4 @@ jsonschema.validate(data, schema) ## Schema Maintenance -**Update triggers**: - -- New top-level sections added -- Property types or validation rules change -- New workflow step types - -**Update process**: - -1. Implement feature in `ngraph/scenario.py` validation logic -2. Test runtime validation -3. Update JSON Schema to match implementation -4. Run `make test` to verify schema tests -5. Update documentation - -Authority: code implementation in `ngraph/scenario.py` and `ngraph/dsl/blueprints/expand.py` is authoritative, not the schema. +Update the schema whenever a top-level section, a field's type, or a workflow step type changes, then run `make test` (the integration tests load every bundled scenario and the DSL examples). The code is authoritative: `ngraph/dsl/loader.py` validates, and `ngraph/dsl/blueprints/expand.py` and the model classes enforce what the schema cannot express. diff --git a/docs/reference/workflow.md b/docs/reference/workflow.md index f68eb3e..616709a 100644 --- a/docs/reference/workflow.md +++ b/docs/reference/workflow.md @@ -8,11 +8,11 @@ Quick links: - [API Reference](api.md) — Python API for programmatic scenario creation - [Auto-Generated API Reference](api-full.md) — complete class and method documentation -NetGraph workflows are analysis execution pipelines that perform capacity analysis, demand placement, and statistics computation. +A workflow is the ordered list of analysis steps a scenario runs. ## Overview -Workflows are ordered steps executed on a scenario. Each step computes a result (e.g., stats, Monte Carlo analysis, export) and writes it under its step name in the results store. +Each step computes one result (statistics, a Monte Carlo analysis, an export) and writes it under its step name in the results store. ```yaml workflow: @@ -31,7 +31,7 @@ workflow: ## Execution Model - Steps run sequentially via `WorkflowStep.execute()`, which records timing and metadata and stores outputs under `{metadata, data}` for the step. -- Monte Carlo steps (`MaxFlow`, `TrafficMatrixPlacement`) execute iterations using the Failure Manager. Each iteration analyzes the network with exclusion sets applied to mask failed nodes/links without mutating the base network. Workers are controlled by `parallelism: auto|int`. +- Monte Carlo steps (`MaxFlow`, `TrafficMatrixPlacement`) execute iterations using the Failure Manager. Each iteration analyzes the network with exclusion sets applied to mask failed nodes/links without mutating the base network. Workers are controlled by `parallelism: auto|int`. For `MaxFlow`, `auto` is the CPU count. For `TrafficMatrixPlacement`, `auto` is 1 unless the demand set uses an LSP preset or the interpreter is free-threaded, because iterations for the other presets are Python-bound and threads only slow them down; an explicit integer is always honoured. - Seeding: a scenario-level `seed` derives per-step seeds unless a step sets an explicit `seed`. Metadata includes `scenario_seed`, `step_seed`, `seed_source`, and `active_seed`. `seed_source`/`active_seed` reflect the seed the step actually uses: a step constructed without its own seed reports `seed_source: none` even when the scenario has a seed (YAML-loaded scenarios derive per-step seeds at parse time, so those report `scenario-derived`). ## Core Workflow Steps @@ -99,7 +99,7 @@ Monte Carlo placement of a named demand set with optional alpha scaling. Baselin demand_set: default failure_policy: random_failures # Optional: policy name in failures section iterations: 100 # Number of failure iterations - parallelism: auto + parallelism: auto # 1 for hop-by-hop/TE_WCMP presets, CPU count with LSP presets include_flow_details: true # cost_distribution per flow include_used_edges: false # include per-demand used edge lists store_failure_patterns: false @@ -116,6 +116,9 @@ Outputs: - data.baseline and data.flow_results: see Results Export Shape below - data.context: demand_set, include_flow_details, include_used_edges, base_demands, alpha, alpha_source +- each flow entry's `data` holds `edges`/`edges_kind: used` with + `include_used_edges`, and `dropped_edges` (volume lost per link) for + `SHORTEST_PATHS_ECMP_LOSSY` demands with `include_flow_details` Note: `placement_rounds` is deprecated and has no effect. It is still accepted in YAML for backward compatibility and is not exported in `data.context`; setting it to any value other than `auto` also logs a deprecation warning. @@ -140,7 +143,7 @@ Search for the maximum uniform traffic multiplier `alpha_star` that is fully pla Parameters: - `demand_set`: Name of the demand set to analyze (default: "default"). -- `acceptance_rule`: Acceptance rule for feasibility (currently only "hard" is supported). +- `acceptance_rule`: Acceptance rule for feasibility (currently only "hard" is supported): every demand must be placed to within the core engine's resolution of 1/4096 and no demand may place nothing. - `alpha_start`: Initial alpha value to probe. - `growth_factor`: Multiplier for bracketing phase (must be > 1.0). - `alpha_min`: Minimum alpha bound for search. @@ -148,7 +151,7 @@ Parameters: - `resolution`: Convergence threshold for bisection. - `max_bracket_iters`: Maximum iterations for bracketing phase. - `max_bisect_iters`: Maximum iterations for bisection phase. -- `placement_rounds`: Deprecated; accepted for backward compatibility but has no effect (placement optimization is handled by the core engine). +- `placement_rounds`: Deprecated; accepted for backward compatibility but has no effect (each demand is placed in one deterministic pass, so repeated rounds change nothing). Outputs: diff --git a/ngraph/analysis/demand.py b/ngraph/analysis/demand.py index 9666133..b2bae2c 100644 --- a/ngraph/analysis/demand.py +++ b/ngraph/analysis/demand.py @@ -33,6 +33,10 @@ class ExpandedDemand: policy_preset: FlowPolicy configuration preset. static_paths: Routes this demand is pinned to, empty when it is routed by the policy. + src_members: Real source node names behind a combine-mode pseudo + source, in selection order; empty for pairwise demands. Hop-by-hop + presets originate an even share of the volume at each member that + can reach a target instead of routing from the pseudo source. """ src_name: str @@ -41,6 +45,7 @@ class ExpandedDemand: priority: int policy_preset: FlowPolicyPreset static_paths: Tuple[StaticPath, ...] = () + src_members: Tuple[str, ...] = () @dataclass @@ -77,6 +82,13 @@ def _expand_combine( ) -> tuple[list[ExpandedDemand], list[AugmentationEdge]]: """Expand combine mode: aggregate sources/sinks through pseudo nodes. + The pseudo source is a virtual source, a pool of the selected sources: a + TE preset carries the aggregate with whichever sources have capacity. + Hop-by-hop presets cannot steer origination, so placement originates an + even share at every member of ``src_members`` that can reach a target + instead; the pseudo sink still delivers each share to that source's + nearest targets. + Nodes selected on both sides are excluded from the target set. Without this guard a shared node would be attached to both pseudo endpoints, forming a zero-cost pseudo_src -> node -> pseudo_snk bypass over two @@ -114,6 +126,7 @@ def _expand_combine( volume=td.volume, priority=td.priority, policy_preset=policy_preset, + src_members=tuple(src_names), ) return [expanded], augmentations diff --git a/ngraph/analysis/functions.py b/ngraph/analysis/functions.py index eea1055..4b585fa 100644 --- a/ngraph/analysis/functions.py +++ b/ngraph/analysis/functions.py @@ -309,12 +309,20 @@ def demand_placement_analysis( pre-built ``context`` 2. Expand demands into concrete (src, dst, volume) tuples (or use a pre-computed expansion) - 3. Place each demand using SPF caching for cacheable policies. - SHORTEST_PATHS_* presets admit flow onto the cost-only shortest paths - of the base topology and drop overflow (IGP semantics); TE_* presets + 3. Place each demand using SPF caching for cacheable policies, in priority + order and input order within a priority; nothing is revisited. + SHORTEST_PATHS_* presets place in one pass on the cost-only shortest + paths of the base topology: ``_ECMP`` admits what the equal-cost next + hops carry without loss, ``_ECMP_LOSSY`` delivers what survives + per-link drops, ``_WCMP`` splits by residual capacity. A combine-mode + demand is a virtual source: with one of these presets every source + that can reach a target originates an even share, and under + ``_ECMP`` the pool is admitted at one global scale. TE_* presets reroute remaining volume onto residual-capacity paths. 4. Fall back to FlowPolicy for presets outside CACHEABLE_PRESETS - 5. Aggregate results into FlowIterationResult + 5. Aggregate results into FlowIterationResult. With + ``include_flow_details`` a lossy demand's entry carries + ``data["dropped_edges"]``, the dropped volume per ``link_id:direction``. SPF Caching Optimization: For cacheable policies (ECMP, WCMP, TE_WCMP_UNLIM), SPF results are @@ -381,23 +389,26 @@ def demand_placement_analysis( ) # Phase 4: Convert to FlowEntry format - flow_entries = [ - FlowEntry( - source=e.src_name, - destination=e.dst_name, - priority=e.priority, - demand=e.volume, - placed=e.placed, - dropped=e.volume - e.placed, - cost_distribution=e.cost_distribution, - data=( - {"edges": sorted(e.used_edges), "edges_kind": "used"} - if e.used_edges - else {} - ), + flow_entries = [] + for e in result.entries or []: + data: dict[str, Any] = {} + if e.used_edges: + data["edges"] = sorted(e.used_edges) + data["edges_kind"] = "used" + if e.dropped_edges: + data["dropped_edges"] = dict(sorted(e.dropped_edges.items())) + flow_entries.append( + FlowEntry( + source=e.src_name, + destination=e.dst_name, + priority=e.priority, + demand=e.volume, + placed=e.placed, + dropped=e.volume - e.placed, + cost_distribution=e.cost_distribution, + data=data, + ) ) - for e in result.entries or [] - ] dropped_flows = sum(1 for e in flow_entries if e.dropped > 0.0) summary = FlowSummary( diff --git a/ngraph/analysis/placement.py b/ngraph/analysis/placement.py index 16f1b89..14fd6c0 100644 --- a/ngraph/analysis/placement.py +++ b/ngraph/analysis/placement.py @@ -3,25 +3,33 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, Sequence +from typing import TYPE_CHECKING, Any, Dict, Sequence, Set, Tuple import netgraph_core import numpy as np from ngraph.analysis.static_paths import build_static_path_bundles +from ngraph.logging import get_logger from ngraph.model.demand.spec import StaticPath -from ngraph.model.flow.policy_config import FlowPolicyPreset, create_flow_policy +from ngraph.model.flow.policy_config import ( + HOP_BY_HOP_PRESETS, + FlowPolicyPreset, + create_flow_policy, + preset_config, +) if TYPE_CHECKING: from ngraph.analysis.context import AnalysisContext from ngraph.analysis.demand import ExpandedDemand +logger = get_logger(__name__) + +#: Presets placed by the SPF-cached engine. Hop-by-hop presets place in one +#: pass on the cost-only DAG of their source; TE_WCMP_UNLIM reroutes the +#: remainder tier by tier on residual-aware DAGs. The LSP presets need Core's +#: FlowPolicy (many flows, reoptimization) and are not cacheable. CACHEABLE_PRESETS: frozenset[FlowPolicyPreset] = frozenset( - { - FlowPolicyPreset.SHORTEST_PATHS_ECMP, - FlowPolicyPreset.SHORTEST_PATHS_WCMP, - FlowPolicyPreset.TE_WCMP_UNLIM, - } + HOP_BY_HOP_PRESETS | {FlowPolicyPreset.TE_WCMP_UNLIM} ) _CACHEABLE_TE: frozenset[FlowPolicyPreset] = frozenset( @@ -30,9 +38,14 @@ } ) -# Threshold for recording a placed amount as a flow entry. The core engine -# itself never augments below kMinFlow = 1/4096 (see NetGraph-Core -# constants.hpp), so any nonzero amount it returns clears this comfortably. +#: Smallest flow the core engine distinguishes: it never augments below +#: kMinFlow = 1/4096 (NetGraph-Core constants.hpp) and rounds per-flow +#: targets up to it. A demand short by at most this much is placed to the +#: engine's numeric resolution, which is what feasibility means here. +FLOW_RESOLUTION = 1.0 / 4096.0 + +# Threshold for recording a placed amount as a flow entry. Any nonzero amount +# the core returns clears FLOW_RESOLUTION and hence this comfortably. _MIN_FLOW = 1e-9 # Cached-path FlowIndex ids start far above the ids Core's FlowPolicy assigns @@ -45,10 +58,19 @@ @dataclass(slots=True) class PlacementSummary: - """Aggregated placement totals.""" + """Aggregated placement totals. + + Attributes: + total_demand: Sum of demand volumes. + total_placed: Sum of placed volumes. + max_shortfall: Largest ``volume - placed`` over all demands. + unserved_demands: Demands with positive volume that placed nothing. + """ total_demand: float total_placed: float + max_shortfall: float = 0.0 + unserved_demands: int = 0 @property def ratio(self) -> float: @@ -56,12 +78,37 @@ def ratio(self) -> float: @property def is_feasible(self) -> bool: - return self.ratio >= 1.0 - 1e-12 + """True when every demand is placed to the engine's resolution. + + The core engine cannot place less than ``FLOW_RESOLUTION`` on a flow, + so a demand whose per-flow share is below it (many LSPs carrying a + small volume) always comes back short by a fraction of that amount. + Requiring an exact match would call such a demand infeasible at any + scale; a shortfall within the resolution is the engine saying "placed". + A demand that placed nothing at all is never feasible, whatever its + volume: below the resolution the engine cannot evaluate it, and above + it nothing was carried. + """ + return self.max_shortfall <= FLOW_RESOLUTION and self.unserved_demands == 0 @dataclass(slots=True) class PlacementEntry: - """Single demand placement result.""" + """Single demand placement result. + + Attributes: + src_name: Source node name (real or pseudo). + dst_name: Destination node name (real or pseudo). + priority: Priority class. + volume: Requested volume. + placed: Placed volume. For ``SHORTEST_PATHS_ECMP_LOSSY`` this is the + volume delivered to the destination. + cost_distribution: Placed volume by path cost, when requested. + used_edges: ``link_id:direction`` of every edge carrying this demand, + when requested. + dropped_edges: Dropped volume by ``link_id:direction``, when requested; + only lossy presets drop. + """ src_name: str dst_name: str @@ -70,6 +117,7 @@ class PlacementEntry: placed: float cost_distribution: dict[float, float] = field(default_factory=dict) used_edges: set[str] = field(default_factory=set) + dropped_edges: dict[str, float] = field(default_factory=dict) @dataclass(slots=True) @@ -86,29 +134,55 @@ class PlacementResult: entries: list[PlacementEntry] | None = None +@dataclass(slots=True) +class _CachedPlacement: + """What one cached placement produced, before aggregation into an entry.""" + + placed: float = 0.0 + cost_distribution: Dict[float, float] = field(default_factory=dict) + used_edges: Set[str] = field(default_factory=set) + dropped_edges: Dict[str, float] = field(default_factory=dict) + + def merge(self, other: "_CachedPlacement") -> None: + self.placed += other.placed + for cost, amount in other.cost_distribution.items(): + self.cost_distribution[cost] = ( + self.cost_distribution.get(cost, 0.0) + amount + ) + self.used_edges |= other.used_edges + for edge, amount in other.dropped_edges.items(): + self.dropped_edges[edge] = self.dropped_edges.get(edge, 0.0) + amount + + +_PRESET_MODES: Dict[ + FlowPolicyPreset, Tuple[netgraph_core.EdgeSelection, netgraph_core.FlowPlacement] +] = {} + + +def _preset_modes( + preset: FlowPolicyPreset, +) -> Tuple[netgraph_core.EdgeSelection, netgraph_core.FlowPlacement]: + """Edge selection and placement mode of a cacheable preset. + + Read from ``preset_config`` so the cached engine and Core's FlowPolicy + agree by construction; memoized because presets are immutable. + """ + modes = _PRESET_MODES.get(preset) + if modes is None: + config = preset_config(preset) + modes = (config.selection, config.flow_placement) + _PRESET_MODES[preset] = modes + return modes + + def _get_edge_selection(preset: FlowPolicyPreset) -> netgraph_core.EdgeSelection: """Get EdgeSelection for a cacheable preset.""" - if preset in ( - FlowPolicyPreset.SHORTEST_PATHS_ECMP, - FlowPolicyPreset.SHORTEST_PATHS_WCMP, - ): - return netgraph_core.EdgeSelection( - multi_edge=True, - require_capacity=False, - tie_break=netgraph_core.EdgeTieBreak.DETERMINISTIC, - ) - return netgraph_core.EdgeSelection( - multi_edge=True, - require_capacity=True, - tie_break=netgraph_core.EdgeTieBreak.PREFER_HIGHER_RESIDUAL, - ) + return _preset_modes(preset)[0] def _get_flow_placement(preset: FlowPolicyPreset) -> netgraph_core.FlowPlacement: """Get FlowPlacement for a cacheable preset.""" - if preset == FlowPolicyPreset.SHORTEST_PATHS_ECMP: - return netgraph_core.FlowPlacement.EQUAL_BALANCED - return netgraph_core.FlowPlacement.PROPORTIONAL + return _preset_modes(preset)[1] def place_demands( @@ -123,10 +197,27 @@ def place_demands( collect_entries: bool = False, include_cost_distribution: bool = False, include_used_edges: bool = False, - dag_cache: dict[tuple[int, bool], tuple[np.ndarray, Any]] | None = None, + dag_cache: dict[tuple, tuple[np.ndarray, Any]] | None = None, ) -> PlacementResult: """Place demands on a flow graph with SPF caching. + Demands are placed one at a time in the given order (callers sort by + priority), each seeing the residual left by the ones before it. Nothing + is revisited, so within a priority class earlier demands win contended + capacity and the totals of rerouting presets depend on demand order. + + Hop-by-hop presets (``HOP_BY_HOP_PRESETS``) place each demand in one pass + on the cost-only shortest-path DAG of its source. A combine-mode demand is + a virtual source, a pool of the selected sources: with such a preset + (``ExpandedDemand.src_members`` set) every member that can reach a target + originates an even share of the volume, since hop-by-hop routing has no + controller that could choose where traffic originates, and each share is + routed to that member's nearest targets. Under lossless ECMP the pool is + admitted as one demand at a single scale. TE presets keep the aggregated + pseudo source and let capacity decide which members originate. A fixed + per-source matrix is a different question, answered by pairwise or + per-group expansion. + Args: demands: Expanded demands (policy_preset, priority, names). volumes: Volume per demand, positionally aligned with `demands`; @@ -140,11 +231,13 @@ def place_demands( resolved_ids: Pre-resolved (src_id, dst_id) pairs. Computed from the demand names if None. collect_entries: If True, populate result.entries. - include_cost_distribution: Include cost distribution in entries. + include_cost_distribution: Include cost distribution and, for lossy + presets, dropped volume per link in entries. include_used_edges: Include used edges in entries. - dag_cache: Optional persistent SPF DAG cache keyed by - (src_id, uses_capacity_aware_selection). Base DAGs depend only on - the static graph and masks, so repeated calls with the same + dag_cache: Optional persistent SPF DAG cache. Base DAGs are keyed by + ``(src_id, uses_capacity_aware_selection)`` and combine-mode + fan-out DAGs by ``(dst_id, "fanout")``. All of them depend only + on the static graph and masks, so repeated calls with the same context and masks (e.g. MSD probes) can share one cache. Returns: @@ -180,11 +273,14 @@ def place_demands( entries: list[PlacementEntry] | None = [] if collect_entries else None total_demand = 0.0 total_placed = 0.0 + max_shortfall = 0.0 + unserved_demands = 0 flow_idx_counter = _CACHED_FLOW_ID_BASE # Core's FlowPolicy assigns flow ids internally per policy instance, so # two policy-based demands sharing (src, dst, priority) would produce # colliding FlowIndex values and silently merge/steal each other's flows. policy_triples: set[tuple[int, int, int]] = set() + node_id_of = ctx.node_mapper.node_id_of for demand, volume, (src_id, dst_id) in zip( demands, volumes, resolved_ids, strict=True @@ -192,21 +288,102 @@ def place_demands( total_demand += volume 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, - volume, - demand.priority, - demand.policy_preset, - dag_cache, - ctx, - flow_graph, - node_mask, - edge_mask, - flow_idx_counter, - include_cost_distribution, - include_used_edges, - ) + if demand.src_members and demand.policy_preset in HOP_BY_HOP_PRESETS: + # Hop-by-hop forwarding cannot steer where traffic originates: + # every source that can reach a target sends an even share + # toward the (aggregated) targets, and the shares are reported + # as one demand. + member_ids = [] + for member in demand.src_members: + member_id = node_id_of.get(member) + if member_id is None: + raise ValueError( + f"Demand source {member!r} is not present in the " + "analysis context graph; rebuild the context from " + "the same demands_config." + ) + member_ids.append(member_id) + if ( + _preset_modes(demand.policy_preset)[1] + == netgraph_core.FlowPlacement.EQUAL_BALANCED_FIXED + ): + # Lossless admission applies to the demand as a whole: one + # pass over a DAG that fans out evenly from the pseudo + # source, so a source that cannot carry its share throttles + # every source alike. + outcome, flow_idx_counter = _place_cached_fanout( + src_id, + dst_id, + member_ids, + volume, + demand.priority, + demand.policy_preset, + dag_cache, + ctx, + flow_graph, + node_mask, + edge_mask, + flow_idx_counter, + include_cost_distribution, + include_used_edges, + ) + else: + # Best-effort and proportional presets carry each share + # independently; totals and per-link drops equal those of a + # simultaneous pass, and shared links are attributed to + # sources in selection order. The virtual source is a + # pool: a member with no path to any target is not part + # of the split. + outcome = _CachedPlacement() + selection = _preset_modes(demand.policy_preset)[0] + reachable = [ + member_id + for member_id in member_ids + if _base_dag( + (member_id, False), + member_id, + selection, + dag_cache, + ctx, + node_mask, + edge_mask, + )[0][dst_id] + != float("inf") + ] + share = volume / len(reachable) if reachable else 0.0 + for member_id in reachable: + partial, flow_idx_counter = _place_cached( + member_id, + dst_id, + share, + demand.priority, + demand.policy_preset, + dag_cache, + ctx, + flow_graph, + node_mask, + edge_mask, + flow_idx_counter, + include_cost_distribution, + include_used_edges, + ) + outcome.merge(partial) + else: + outcome, flow_idx_counter = _place_cached( + src_id, + dst_id, + volume, + demand.priority, + demand.policy_preset, + dag_cache, + ctx, + flow_graph, + node_mask, + edge_mask, + flow_idx_counter, + include_cost_distribution, + include_used_edges, + ) else: triple = (src_id, dst_id, demand.priority) if triple in policy_triples: @@ -227,7 +404,7 @@ def place_demands( "volumes or use distinct priorities." ) policy_triples.add(triple) - placed, cost_dist, used_edges = _place_with_policy( + outcome = _place_with_policy( src_id, dst_id, volume, @@ -244,7 +421,12 @@ def place_demands( dst_name=demand.dst_name, ) - total_placed += placed + total_placed += outcome.placed + shortfall = volume - outcome.placed + if shortfall > max_shortfall: + max_shortfall = shortfall + if volume > 0.0 and outcome.placed <= 0.0: + unserved_demands += 1 if entries is not None: entries.append( @@ -253,25 +435,67 @@ def place_demands( dst_name=demand.dst_name, priority=demand.priority, volume=volume, - placed=placed, - cost_distribution=cost_dist if include_cost_distribution else {}, - used_edges=used_edges if include_used_edges else set(), + placed=outcome.placed, + cost_distribution=outcome.cost_distribution, + used_edges=outcome.used_edges, + dropped_edges=outcome.dropped_edges, ) ) return PlacementResult( - summary=PlacementSummary(total_demand=total_demand, total_placed=total_placed), + summary=PlacementSummary( + total_demand=total_demand, + total_placed=total_placed, + max_shortfall=max_shortfall, + unserved_demands=unserved_demands, + ), entries=entries, ) +def _edge_label(ctx: "AnalysisContext", ext_ids: Any, edge_id: int) -> str | None: + """``link_id:direction`` of a Core edge, or None for pseudo edges.""" + ref = ctx.edge_mapper.decode_ext_id(int(ext_ids[edge_id])) + return f"{ref.link_id}:{ref.direction}" if ref else None + + +def _base_dag( + cache_key: tuple, + src_id: int, + selection: netgraph_core.EdgeSelection, + dag_cache: dict[tuple, tuple[np.ndarray, Any]], + ctx: "AnalysisContext", + node_mask: np.ndarray, + edge_mask: np.ndarray, +) -> tuple[np.ndarray, Any]: + """Cost-only (or capacity-gated) SPF DAG from ``src_id``, cached under ``cache_key``. + + Base DAGs depend only on the static graph and the masks, so one entry + serves every demand from the same source within a mask state. + """ + entry = dag_cache.get(cache_key) + if entry is None: + entry = ctx.algorithms.spf( + ctx.handle, + src=src_id, + dst=None, + selection=selection, + node_mask=node_mask, + edge_mask=edge_mask, + multipath=True, + dtype="float64", + ) + dag_cache[cache_key] = entry + return entry + + def _place_cached( src_id: int, dst_id: int, volume: float, priority: int, preset: FlowPolicyPreset, - dag_cache: dict[tuple[int, bool], tuple[np.ndarray, Any]], + dag_cache: dict[tuple, tuple[np.ndarray, Any]], ctx: "AnalysisContext", flow_graph: netgraph_core.FlowGraph, node_mask: np.ndarray, @@ -279,54 +503,60 @@ def _place_cached( flow_idx_start: int, include_cost_distribution: bool, include_used_edges: bool, -) -> tuple[float, dict[float, float], set[str], int]: +) -> tuple[_CachedPlacement, int]: """Place single demand with SPF caching.""" - selection = _get_edge_selection(preset) - placement = _get_flow_placement(preset) + selection, placement = _preset_modes(preset) is_te = preset in _CACHEABLE_TE - # ECMP and WCMP share one EdgeSelection; TE presets share the other. - # Keying by selection family (not preset) lets mixed workloads reuse - # the same base SPF DAG. + lossy = placement == netgraph_core.FlowPlacement.EQUAL_BALANCED_LOSSY + # Hop-by-hop presets share one cost-only EdgeSelection; TE presets share + # the capacity-aware one. Keying by selection family (not preset) lets + # mixed workloads reuse the same base SPF DAG. cache_key = (src_id, is_te) + outcome = _CachedPlacement() flow_indices: list[netgraph_core.FlowIndex] = [] flow_costs: list[tuple[float, float]] = [] + raw_drops: list[tuple[int, float]] = [] flow_idx_counter = flow_idx_start - placed = 0.0 remaining = volume - if cache_key not in dag_cache: - dists, dag = ctx.algorithms.spf( - ctx.handle, - src=src_id, - dst=None, - selection=selection, - node_mask=node_mask, - edge_mask=edge_mask, - multipath=True, - dtype="float64", - ) - dag_cache[cache_key] = (dists, dag) - - dists, dag = dag_cache[cache_key] + dists, dag = _base_dag( + cache_key, src_id, selection, dag_cache, ctx, node_mask, edge_mask + ) if dists[dst_id] == float("inf"): - return 0.0, {}, set(), flow_idx_counter + return outcome, flow_idx_counter cost = float(dists[dst_id]) flow_idx = netgraph_core.FlowIndex(src_id, dst_id, priority, flow_idx_counter) flow_idx_counter += 1 - amount = flow_graph.place(flow_idx, src_id, dst_id, dag, remaining, placement) + if lossy: + amount, drops = flow_graph.place_with_drops( + flow_idx, src_id, dst_id, dag, remaining, placement + ) + raw_drops.extend(drops) + # Volume carried part of the way and dropped downstream still occupies + # the links it crossed, so the flow counts as using them. + if amount > _MIN_FLOW or drops: + flow_indices.append(flow_idx) + else: + amount = flow_graph.place(flow_idx, src_id, dst_id, dag, remaining, placement) + if amount > _MIN_FLOW: + flow_indices.append(flow_idx) if amount > _MIN_FLOW: - flow_indices.append(flow_idx) flow_costs.append((cost, amount)) - placed += amount + outcome.placed += amount remaining -= amount if is_te and remaining > _MIN_FLOW: - for _ in range(100): + # Reroute the remainder tier by tier. Every iteration either + # saturates at least one edge of the residual DAG it placed on (a + # proportional placement runs a max-flow over the DAG) or makes no + # progress and stops, so the edge count bounds the iterations. + max_iterations = ctx.multidigraph.num_edges() + for _ in range(max_iterations): residual = np.ascontiguousarray( flow_graph.residual_view(), dtype=np.float64 ) @@ -362,27 +592,126 @@ def _place_cached( flow_indices.append(flow_idx) flow_costs.append((fresh_cost, additional)) - placed += additional + outcome.placed += additional remaining -= additional if remaining < _MIN_FLOW: break + else: + logger.warning( + "TE rerouting for demand %s->%s stopped at the edge-count bound " + "(%d iterations) with %.6g still unplaced; this should not happen " + "and indicates an engine invariant violation", + ctx.node_mapper.to_name(src_id), + ctx.node_mapper.to_name(dst_id), + max_iterations, + remaining, + ) - cost_dist: dict[float, float] = {} if include_cost_distribution: for c, amt in flow_costs: - cost_dist[c] = cost_dist.get(c, 0.0) + amt + outcome.cost_distribution[c] = outcome.cost_distribution.get(c, 0.0) + amt + if raw_drops: + ext_ids = ctx.multidigraph.ext_edge_ids_view() + for edge_id, amt in raw_drops: + label = _edge_label(ctx, ext_ids, edge_id) + if label: + outcome.dropped_edges[label] = outcome.dropped_edges.get( + label, 0.0 + ) + float(amt) - used_edges: set[str] = set() if include_used_edges: ext_ids = ctx.multidigraph.ext_edge_ids_view() for fidx in flow_indices: for edge_id, _ in flow_graph.get_flow_edges(fidx): - ref = ctx.edge_mapper.decode_ext_id(int(ext_ids[edge_id])) - if ref: - used_edges.add(f"{ref.link_id}:{ref.direction}") + label = _edge_label(ctx, ext_ids, edge_id) + if label: + outcome.used_edges.add(label) - return placed, cost_dist, used_edges, flow_idx_counter + return outcome, flow_idx_counter + + +def _place_cached_fanout( + root_id: int, + dst_id: int, + member_ids: Sequence[int], + volume: float, + priority: int, + preset: FlowPolicyPreset, + dag_cache: dict[tuple, tuple[np.ndarray, Any]], + ctx: "AnalysisContext", + flow_graph: netgraph_core.FlowGraph, + node_mask: np.ndarray, + edge_mask: np.ndarray, + flow_idx_start: int, + include_cost_distribution: bool, + include_used_edges: bool, +) -> tuple[_CachedPlacement, int]: + """Place a combine-mode demand in one pass with an even origination split. + + The DAG fans out from the pseudo source ``root_id`` over every attachment + edge regardless of cost and then follows the shortest paths of each source + toward ``dst_id``. Equal-balanced placement over it splits the volume + evenly across the sources and admits the demand at one global scale, so a + source that cannot carry its share throttles every source alike. The + virtual source is a pool: a source with no path to any target is not a + member of the split, and the others share the volume evenly. + + The DAG is cached per destination (one reverse SPF per demand per mask + state) rather than per source. + """ + selection, placement = _preset_modes(preset) + cache_key = (dst_id, "fanout") + outcome = _CachedPlacement() + flow_idx_counter = flow_idx_start + + if cache_key not in dag_cache: + graph = ctx.multidigraph + row = graph.row_offsets_view() + fanout = graph.adj_edge_index_view()[int(row[root_id]) : int(row[root_id + 1])] + dists, dag = ctx.algorithms.spf_to( + ctx.handle, + dst_id, + selection=selection, + node_mask=node_mask, + edge_mask=edge_mask, + multipath=True, + fanout_edges=[int(e) for e in fanout], + dtype="float64", + ) + dag_cache[cache_key] = (dists, dag) + + dists, dag = dag_cache[cache_key] + # Members with no path to any target are not in the fan-out (spf_to skips + # their attachment edge), so the split is over the reachable ones only. + member_costs = [ + c for c in (float(dists[m]) for m in member_ids) if c != float("inf") + ] + if not member_costs: + return outcome, flow_idx_counter + + flow_idx = netgraph_core.FlowIndex(root_id, dst_id, priority, flow_idx_counter) + flow_idx_counter += 1 + amount = flow_graph.place(flow_idx, root_id, dst_id, dag, volume, placement) + if amount <= _MIN_FLOW: + return outcome, flow_idx_counter + outcome.placed = amount + + if include_cost_distribution: + # The fan-out is an equal split and lossless admission keeps the + # ratios, so every source carries the same share. + share = amount / len(member_costs) + for c in member_costs: + outcome.cost_distribution[c] = outcome.cost_distribution.get(c, 0.0) + share + + if include_used_edges: + ext_ids = ctx.multidigraph.ext_edge_ids_view() + for edge_id, _ in flow_graph.get_flow_edges(flow_idx): + label = _edge_label(ctx, ext_ids, edge_id) + if label: + outcome.used_edges.add(label) + + return outcome, flow_idx_counter def _place_with_policy( @@ -400,7 +729,7 @@ def _place_with_policy( static_paths: Sequence[StaticPath] = (), src_name: str = "", dst_name: str = "", -) -> tuple[float, dict[float, float], set[str]]: +) -> _CachedPlacement: """Place a single demand using FlowPolicy. Used for non-cacheable presets and for any demand pinned to explicit @@ -420,8 +749,7 @@ def _place_with_policy( 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] = {} - used_edges: set[str] = set() + outcome = _CachedPlacement(placed=placed) if include_cost_distribution or include_used_edges: ext_ids = ctx.multidigraph.ext_edge_ids_view() @@ -429,15 +757,17 @@ def _place_with_policy( if include_cost_distribution: cost, flow_vol = float(flow_data[2]), float(flow_data[3]) if flow_vol > 0: - cost_dist[cost] = cost_dist.get(cost, 0.0) + flow_vol + outcome.cost_distribution[cost] = ( + outcome.cost_distribution.get(cost, 0.0) + flow_vol + ) if include_used_edges: fidx = netgraph_core.FlowIndex( flow_key[0], flow_key[1], flow_key[2], flow_key[3] ) for edge_id, _ in flow_graph.get_flow_edges(fidx): - ref = ctx.edge_mapper.decode_ext_id(int(ext_ids[edge_id])) - if ref: - used_edges.add(f"{ref.link_id}:{ref.direction}") + label = _edge_label(ctx, ext_ids, edge_id) + if label: + outcome.used_edges.add(label) - return placed, cost_dist, used_edges + return outcome diff --git a/ngraph/model/flow/policy_config.py b/ngraph/model/flow/policy_config.py index a32069d..15c4f14 100644 --- a/ngraph/model/flow/policy_config.py +++ b/ngraph/model/flow/policy_config.py @@ -1,7 +1,10 @@ """Flow policy preset configurations for NetGraph. -Named routing presets and the factory that materializes them as NetGraph-Core -FlowPolicy objects built from a FlowPolicyConfig. +Named routing presets, the single mapping from a preset to a NetGraph-Core +``FlowPolicyConfig``, and the factory that materializes a preset as a Core +``FlowPolicy``. Both placement engines (the SPF-cached fast path in +``ngraph.analysis.placement`` and Core's FlowPolicy) read their edge selection +and placement mode from ``preset_config`` so the two cannot drift. """ from __future__ import annotations @@ -26,18 +29,29 @@ class FlowPolicyPreset(IntEnum): These presets map to specific combinations of path algorithms, flow placement strategies, and edge selection modes provided by NetGraph-Core. + + The ``SHORTEST_PATHS_*`` presets model hop-by-hop IP/IGP forwarding: routes + follow link costs alone and each demand is placed in one pass on the + cost-only shortest-path DAG. The ``TE_*`` presets model a controller that + selects paths with knowledge of residual capacity. """ SHORTEST_PATHS_ECMP = 1 - """Hop-by-hop equal-cost multi-path routing (ECMP). + """Hop-by-hop equal-cost multi-path routing (ECMP), lossless admission. - Single flow with equal-cost path splitting, similar to IP forwarding with ECMP. + Traffic is hashed equally over every equal-cost next hop and admitted at + the largest volume that causes no loss on any of them. A next hop + saturated by an earlier demand blocks admission of any later demand + hashed onto it, because the forwarding table does not react to load. + ``placed`` is the volume the network carries without drops. """ SHORTEST_PATHS_WCMP = 2 """Hop-by-hop weighted cost multi-path routing (WCMP). - Single flow with proportional splitting over equal-cost paths. + Single flow with proportional splitting over equal-cost paths. Weights + follow residual capacity, which equals link capacity on an unloaded + network and adapts to load placed by earlier demands. """ TE_WCMP_UNLIM = 3 @@ -75,86 +89,76 @@ class FlowPolicyPreset(IntEnum): Configuration: multipath=False ensures tunnel-based ECMP (not hash-based ECMP). """ + SHORTEST_PATHS_ECMP_LOSSY = 6 + """Hop-by-hop ECMP, best-effort forwarding with loss. + + Traffic is hashed equally over every equal-cost next hop; each link + carries what fits and drops the rest, and a deficit propagates + downstream. ``placed`` is the volume delivered to the destination and + ``dropped`` is what was lost on the way. With flow details enabled the + result records the dropped volume per link. + """ -def 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. + +#: Presets that model hop-by-hop IP/IGP forwarding: cost-only routes, one +#: placement pass per demand, no rerouting. In combine mode these presets +#: originate an even share of the demand at every source that can reach a +#: target. +HOP_BY_HOP_PRESETS: frozenset[FlowPolicyPreset] = frozenset( + { + FlowPolicyPreset.SHORTEST_PATHS_ECMP, + FlowPolicyPreset.SHORTEST_PATHS_WCMP, + FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY, + } +) + + +def preset_config(preset: FlowPolicyPreset) -> netgraph_core.FlowPolicyConfig: + """Build the Core ``FlowPolicyConfig`` a preset stands for. + + This is the single source of the preset semantics. The SPF-cached + placement engine reads ``selection`` and ``flow_placement`` from it, and + ``create_flow_policy`` materializes it as a Core ``FlowPolicy``. + + Hop-by-hop presets set ``require_capacity=False`` (routes follow costs + only) and ``shortest_path=True`` (one placement on the cost-only DAG), so + a FlowPolicy built from them places exactly what the cached engine + places. Args: - algorithms: NetGraph-Core Algorithms instance. - graph: NetGraph-Core Graph handle. - preset: Preset whose path algorithm, placement, edge selection, and - 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. + preset: Preset to describe. Returns: - netgraph_core.FlowPolicy: Configured policy instance. + A fresh ``FlowPolicyConfig``; callers may adjust it further. Raises: ValueError: If an unknown FlowPolicyPreset value is provided. - - Example: - >>> backend = netgraph_core.Backend.cpu() - >>> algs = netgraph_core.Algorithms(backend) - >>> graph = algs.build_graph(strict_multidigraph) - >>> policy = create_flow_policy(algs, graph, FlowPolicyPreset.SHORTEST_PATHS_ECMP) """ + config = netgraph_core.FlowPolicyConfig() + config.path_alg = netgraph_core.PathAlg.SPF - 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() - config.path_alg = netgraph_core.PathAlg.SPF - config.flow_placement = netgraph_core.FlowPlacement.EQUAL_BALANCED - config.selection = netgraph_core.EdgeSelection( - multi_edge=True, - require_capacity=False, - tie_break=netgraph_core.EdgeTieBreak.DETERMINISTIC, - ) - config.min_flow_count = 1 - config.max_flow_count = 1 - return _build(config) - - elif preset == FlowPolicyPreset.SHORTEST_PATHS_WCMP: - # Hop-by-hop weighted ECMP (WCMP) over equal-cost paths (proportional split) - config = netgraph_core.FlowPolicyConfig() - config.path_alg = netgraph_core.PathAlg.SPF - config.flow_placement = netgraph_core.FlowPlacement.PROPORTIONAL + if preset in HOP_BY_HOP_PRESETS: + # Hop-by-hop IP/IGP forwarding: cost-only routing, single pass, one + # flow whose split rule is the only thing that differs per preset. config.selection = netgraph_core.EdgeSelection( multi_edge=True, require_capacity=False, tie_break=netgraph_core.EdgeTieBreak.DETERMINISTIC, ) + config.require_capacity = False + config.shortest_path = True config.min_flow_count = 1 config.max_flow_count = 1 - return _build(config) - - elif preset == FlowPolicyPreset.TE_WCMP_UNLIM: + if preset == FlowPolicyPreset.SHORTEST_PATHS_ECMP: + config.flow_placement = netgraph_core.FlowPlacement.EQUAL_BALANCED_FIXED + elif preset == FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY: + config.flow_placement = netgraph_core.FlowPlacement.EQUAL_BALANCED_LOSSY + else: + config.flow_placement = netgraph_core.FlowPlacement.PROPORTIONAL + return config + + if preset == FlowPolicyPreset.TE_WCMP_UNLIM: # Traffic engineering with WCMP (proportional split) and capacity-aware selection - config = netgraph_core.FlowPolicyConfig() - config.path_alg = netgraph_core.PathAlg.SPF config.flow_placement = netgraph_core.FlowPlacement.PROPORTIONAL config.selection = netgraph_core.EdgeSelection( multi_edge=True, @@ -163,44 +167,79 @@ def _build(config: netgraph_core.FlowPolicyConfig) -> netgraph_core.FlowPolicy: ) config.min_flow_count = 1 # max_flow_count defaults to None (unlimited) - return _build(config) + return config - elif preset == FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP: - # TE with up to 256 LSPs using ECMP flow placement + if preset in ( + FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP, + FlowPolicyPreset.TE_ECMP_16_LSP, + ): + # TE with ECMP flow placement over single-path tunnels. # multipath=False ensures each LSP is a single path (MPLS tunnel semantics) - config = netgraph_core.FlowPolicyConfig() - config.path_alg = netgraph_core.PathAlg.SPF config.flow_placement = netgraph_core.FlowPlacement.EQUAL_BALANCED config.selection = netgraph_core.EdgeSelection( multi_edge=False, require_capacity=True, tie_break=netgraph_core.EdgeTieBreak.PREFER_HIGHER_RESIDUAL, ) - config.multipath = False # Each LSP uses a single path (tunnel-based ECMP) - config.min_flow_count = 1 - config.max_flow_count = 256 + config.multipath = False config.reoptimize_flows_on_each_placement = True - return _build(config) + if preset == FlowPolicyPreset.TE_ECMP_16_LSP: + config.min_flow_count = 16 + config.max_flow_count = 16 + else: + config.min_flow_count = 1 + config.max_flow_count = 256 + return config + + raise ValueError(f"Unknown flow policy preset: {preset}") - elif preset == FlowPolicyPreset.TE_ECMP_16_LSP: - # TE with exactly 16 LSPs using ECMP flow placement - # multipath=False ensures each LSP is a single path (MPLS tunnel semantics) - config = netgraph_core.FlowPolicyConfig() - config.path_alg = netgraph_core.PathAlg.SPF - config.flow_placement = netgraph_core.FlowPlacement.EQUAL_BALANCED - config.selection = netgraph_core.EdgeSelection( - multi_edge=False, - require_capacity=True, - tie_break=netgraph_core.EdgeTieBreak.PREFER_HIGHER_RESIDUAL, - ) - config.multipath = False # Each LSP uses a single path (tunnel-based ECMP) - config.min_flow_count = 16 - config.max_flow_count = 16 - config.reoptimize_flows_on_each_placement = True - return _build(config) - else: - raise ValueError(f"Unknown flow policy preset: {preset}") +def 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. + + Args: + algorithms: NetGraph-Core Algorithms instance. + graph: NetGraph-Core Graph handle. + preset: Preset whose path algorithm, placement, edge selection, and + flow-count bounds to apply (see ``preset_config``). + 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. + + Raises: + ValueError: If an unknown FlowPolicyPreset value is provided. + + Example: + >>> backend = netgraph_core.Backend.cpu() + >>> algs = netgraph_core.Algorithms(backend) + >>> graph = algs.build_graph(strict_multidigraph) + >>> policy = create_flow_policy(algs, graph, FlowPolicyPreset.SHORTEST_PATHS_ECMP) + """ + config = preset_config(preset) + 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, + # reoptimization and the single-augmentation IP mode are inert or + # rejected 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 + ) def serialize_policy_preset(cfg: Any) -> Optional[str]: diff --git a/ngraph/workflow/maximum_supported_demand_step.py b/ngraph/workflow/maximum_supported_demand_step.py index 0753654..48bb41c 100644 --- a/ngraph/workflow/maximum_supported_demand_step.py +++ b/ngraph/workflow/maximum_supported_demand_step.py @@ -88,7 +88,7 @@ class MaximumSupportedDemand(WorkflowStep): max_bracket_iters: Maximum iterations for bracketing phase. max_bisect_iters: Maximum iterations for bisection phase. placement_rounds: Deprecated; accepted for backward compatibility but - has no effect (placement optimization is handled by the core engine). + has no effect (each demand is placed in one deterministic pass). """ demand_set: str = "default" @@ -106,7 +106,7 @@ def __post_init__(self) -> None: if self.placement_rounds != "auto": logger.warning( "MaximumSupportedDemand 'placement_rounds' is deprecated and has " - "no effect; placement optimization is handled by the core engine." + "no effect; each demand is placed in one deterministic pass." ) try: self.alpha_start = float(self.alpha_start) @@ -200,7 +200,7 @@ def _binary_search(self, probe: "Any") -> float: start_alpha = float(self.alpha_start) g = float(self.growth_factor) - feasible0, _ = probe(start_alpha) + feasible0, details0 = probe(start_alpha) lower: float | None = None upper: float | None = None @@ -231,11 +231,13 @@ def _binary_search(self, probe: "Any") -> float: else: upper = start_alpha alpha = start_alpha + best_ratio = float(details0.get("placement_ratio", 0.0)) for _ in range(self.max_bracket_iters): alpha = max(alpha / g, self.alpha_min) if alpha == upper: break - feas, _ = probe(alpha) + feas, details = probe(alpha) + best_ratio = max(best_ratio, float(details.get("placement_ratio", 0.0))) if feas: lower = alpha break @@ -244,12 +246,22 @@ def _binary_search(self, probe: "Any") -> float: # Mirror the upward branch: bracket iterations can run out # before the halving sequence reaches alpha_min (e.g. a large # alpha_start), so probe alpha_min directly before giving up. - if upper <= self.alpha_min: - raise ValueError("No feasible alpha found above alpha_min") - feas, _ = probe(self.alpha_min) - if not feas: - raise ValueError("No feasible alpha found above alpha_min") - lower = self.alpha_min + if upper > self.alpha_min: + feas, details = probe(self.alpha_min) + best_ratio = max( + best_ratio, float(details.get("placement_ratio", 0.0)) + ) + if feas: + lower = self.alpha_min + if lower is None: + raise ValueError( + f"No feasible alpha found above alpha_min={self.alpha_min:g} " + f"(best placement ratio over probes {best_ratio:.6f}). Some " + "demand cannot be placed fully at any scale: check that " + "every source reaches its targets, that no required " + "element is disabled, and that demand volumes are not " + "orders of magnitude below link capacities." + ) assert lower is not None and upper is not None and lower < upper @@ -302,6 +314,12 @@ def _evaluate_alpha( Uses pre-built cache; only scales demand volumes by alpha. Placement is deterministic so a single evaluation is sufficient. + + Feasible means every demand is placed to the core engine's numeric + resolution (``PlacementSummary.is_feasible``): the engine never + places less than 1/4096 on a flow, so a demand spread over many LSPs + can come back short by a fraction of that even on an empty network, + and an exact-match rule would call it infeasible at every scale. """ ctx = cache.ctx volumes = [d.volume * alpha for d in cache.base_expanded] diff --git a/ngraph/workflow/traffic_matrix_placement_step.py b/ngraph/workflow/traffic_matrix_placement_step.py index 3d3871a..bd2fda9 100644 --- a/ngraph/workflow/traffic_matrix_placement_step.py +++ b/ngraph/workflow/traffic_matrix_placement_step.py @@ -22,12 +22,16 @@ from __future__ import annotations +import sys import time from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Iterable from ngraph.analysis.failure_manager import FailureManager +from ngraph.analysis.placement import CACHEABLE_PRESETS from ngraph.logging import get_logger +from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.workflow.base import ( WorkflowStep, register_workflow_step, @@ -41,6 +45,48 @@ logger = get_logger(__name__) +def _python_is_free_threaded() -> bool: + """True on a free-threaded (no-GIL) interpreter, where threads scale.""" + is_gil_enabled = getattr(sys, "_is_gil_enabled", None) + return is_gil_enabled is not None and not is_gil_enabled() + + +def resolve_placement_parallelism( + parallelism: int | str, demands: Iterable[TrafficDemand] +) -> int: + """Resolve the worker count for demand placement iterations. + + An explicit integer is used as given. ``"auto"`` becomes the CPU count + only when iterations can run concurrently: on a free-threaded interpreter, + or when the demand set uses a preset outside ``CACHEABLE_PRESETS`` (the + LSP presets), whose placement runs inside the core engine with the GIL + released. Iterations for cacheable presets are dominated by Python-side + work between very short engine calls, so on a GIL interpreter threads + only add contention and ``"auto"`` resolves to 1. + + Args: + parallelism: Positive worker count or ``"auto"``. + demands: Demands of the set to place; unset presets count as the + default ``SHORTEST_PATHS_ECMP``. + + Returns: + Positive worker count. + + Raises: + ValueError: If ``parallelism`` is neither a positive integer nor + ``"auto"``. + """ + resolved = resolve_parallelism(parallelism) + if parallelism != "auto" or resolved == 1: + return resolved + if _python_is_free_threaded(): + return resolved + presets = {td.flow_policy or FlowPolicyPreset.SHORTEST_PATHS_ECMP for td in demands} + if presets - CACHEABLE_PRESETS: + return resolved + return 1 + + @dataclass class TrafficMatrixPlacement(WorkflowStep): """Monte Carlo demand placement using a named demand set. @@ -55,9 +101,13 @@ class TrafficMatrixPlacement(WorkflowStep): failure_policy: Failure policy name in scenario.failure_policy_set. If None, no failure policy is applied. iterations: Number of failure iterations to run; must be >= 0. - parallelism: Worker thread count, or "auto" for the CPU count. + parallelism: Worker thread count, or "auto". Auto uses the CPU count + when iterations can run concurrently (an LSP preset in the demand + set, or a free-threaded interpreter) and 1 otherwise, because + cacheable presets are Python-bound under the GIL and threads only + slow them down. See ``resolve_placement_parallelism``. placement_rounds: Deprecated; accepted for backward compatibility but - has no effect (placement optimization is handled by the core engine). + has no effect (each demand is placed in one deterministic pass). seed: Optional seed for reproducibility. store_failure_patterns: Record the failure trace on each result. Iterations are deduplicated, so a trace describes the first @@ -88,7 +138,7 @@ def __post_init__(self) -> None: if self.placement_rounds != "auto": logger.warning( "TrafficMatrixPlacement 'placement_rounds' is deprecated and has " - "no effect; placement optimization is handled by the core engine." + "no effect; each demand is placed in one deterministic pass." ) if self.iterations < 0: raise ValueError("iterations must be >= 0") @@ -147,7 +197,12 @@ def run(self, scenario: "Scenario") -> None: failure_policy_set=scenario.failure_policy_set, policy_name=self.failure_policy, ) - effective_parallelism = resolve_parallelism(self.parallelism) + effective_parallelism = resolve_placement_parallelism(self.parallelism, td_list) + if self.parallelism == "auto": + logger.info( + "Resolved parallelism 'auto' to %d worker(s) for this demand set", + effective_parallelism, + ) raw = fm.run_demand_placement_monte_carlo( demands_config=demands_config, diff --git a/pyproject.toml b/pyproject.toml index 5943f43..077c121 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" # --------------------------------------------------------------------- [project] name = "ngraph" -version = "0.22.0" +version = "0.23.0" description = "A tool and a library for network modeling and analysis." readme = "README.md" authors = [{ name = "Andrey Golovanov" }] @@ -35,7 +35,7 @@ dependencies = [ "pyyaml>=6.0", "pandas>=2.0", "jsonschema>=4.0", - "netgraph-core>=0.8.0", + "netgraph-core>=0.9.0", ] [project.urls] diff --git a/tests/analysis/test_demand_expansion_semantics.py b/tests/analysis/test_demand_expansion_semantics.py index 75e51da..ad762f7 100644 --- a/tests/analysis/test_demand_expansion_semantics.py +++ b/tests/analysis/test_demand_expansion_semantics.py @@ -191,7 +191,22 @@ def test_per_group_combine_excludes_own_group_nodes(self) -> None: } assert not attached_to_src & attached_to_snk - def test_per_group_combine_placement_bounded_by_real_capacity(self) -> None: + @pytest.mark.parametrize( + "preset,expected_placed", + [ + # Each group's combine demand is a virtual source: A/2 and B/2 + # have no path at all and leave the pool, so A/1 -> B/1 and + # B/1 -> A/1 each carry the single capacity-1.0 link's worth + # whatever the preset. The bypass previously placed all 100. + ("SHORTEST_PATHS_ECMP", 2.0), + ("SHORTEST_PATHS_ECMP_LOSSY", 2.0), + ("SHORTEST_PATHS_WCMP", 2.0), + ("TE_WCMP_UNLIM", 2.0), + ], + ) + def test_per_group_combine_placement_bounded_by_real_capacity( + self, preset: str, expected_placed: float + ) -> None: """Placement is bounded by real capacity, not the pseudo bypass.""" network = self._two_group_single_link_network() demands_config = [ @@ -201,6 +216,7 @@ def test_per_group_combine_placement_bounded_by_real_capacity(self) -> None: "volume": 100.0, "mode": "combine", "group_mode": "per_group", + "flow_policy": preset, } ] @@ -211,11 +227,9 @@ def test_per_group_combine_placement_bounded_by_real_capacity(self) -> None: demands_config=demands_config, ) - # The single capacity-1.0 link (plus its reverse edge) bounds - # placement at 2.0; the bypass previously placed all 100. assert result.summary.total_demand == pytest.approx(100.0) - assert result.summary.total_placed == pytest.approx(2.0) - assert result.summary.overall_ratio == pytest.approx(0.02) + assert result.summary.total_placed == pytest.approx(expected_placed) + assert result.summary.overall_ratio == pytest.approx(expected_placed / 100.0) def test_flatten_combine_full_overlap_raises(self) -> None: network = self._two_group_single_link_network() diff --git a/tests/analysis/test_placement.py b/tests/analysis/test_placement.py index 0ecd825..ae2bce4 100644 --- a/tests/analysis/test_placement.py +++ b/tests/analysis/test_placement.py @@ -190,7 +190,8 @@ def test_get_placement_for_ecmp(self) -> None: import netgraph_core placement = _get_flow_placement(FlowPolicyPreset.SHORTEST_PATHS_ECMP) - assert placement == netgraph_core.FlowPlacement.EQUAL_BALANCED + # Lossless hash-ECMP admission with a load-blind next-hop set. + assert placement == netgraph_core.FlowPlacement.EQUAL_BALANCED_FIXED def test_get_placement_for_wcmp(self) -> None: """Test FlowPlacement for WCMP preset.""" @@ -1066,9 +1067,13 @@ def _build_demands_config( (FlowPolicyPreset.SHORTEST_PATHS_ECMP, ["A"], ["D"], False), (FlowPolicyPreset.SHORTEST_PATHS_ECMP, ["A"], ["D", "E"], False), (FlowPolicyPreset.SHORTEST_PATHS_ECMP, ["A", "B"], ["D"], False), + (FlowPolicyPreset.SHORTEST_PATHS_ECMP, ["A"], ["D", "E"], True), # WCMP tests (FlowPolicyPreset.SHORTEST_PATHS_WCMP, ["A"], ["D"], False), (FlowPolicyPreset.SHORTEST_PATHS_WCMP, ["A"], ["D", "E"], False), + (FlowPolicyPreset.SHORTEST_PATHS_WCMP, ["A"], ["D", "E"], True), + # Lossy ECMP + (FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY, ["A"], ["D", "E"], True), # TE_WCMP_UNLIM tests (FlowPolicyPreset.TE_WCMP_UNLIM, ["A"], ["D"], False), (FlowPolicyPreset.TE_WCMP_UNLIM, ["A"], ["D"], True), @@ -1079,8 +1084,11 @@ def _build_demands_config( "ecmp_single_src_single_dest", "ecmp_single_src_multi_dest", "ecmp_multi_src_single_dest", + "ecmp_single_src_multi_dest_constrained", "wcmp_single_src_single_dest", "wcmp_single_src_multi_dest", + "wcmp_single_src_multi_dest_constrained", + "lossy_single_src_multi_dest_constrained", "te_single_src_single_dest_unconstrained", "te_single_src_single_dest_constrained", "te_multi_src_single_dest_constrained", @@ -1096,14 +1104,12 @@ def test_cached_equals_noncached( multi_dest_constrained_network: Network, multi_source_multi_dest_network: Network, ) -> None: - """Cached placement matches FlowPolicy placement in uncontended cases. - - The equivalence holds when demands do not compete for capacity. Under - contention the cached path is canonical for SHORTEST_PATHS presets: - it admits flow onto the cost-only shortest paths of the base topology - and drops the overflow (IGP semantics), whereas Core's FlowPolicy - reroutes onto costlier residual paths (TE semantics, available via - the TE_* presets). + """Cached placement matches FlowPolicy placement, contended or not. + + Both engines read the same ``preset_config``: hop-by-hop presets are + cost-only and single-pass in the FlowPolicy too, so under contention + both admit onto the base shortest paths and leave the overflow, + while TE presets reroute in both. """ # Select network based on source count if len(sources) > 1: diff --git a/tests/analysis/test_placement_models.py b/tests/analysis/test_placement_models.py new file mode 100644 index 0000000..1e884cb --- /dev/null +++ b/tests/analysis/test_placement_models.py @@ -0,0 +1,592 @@ +"""Placement model semantics: combine-mode origination, lossless and lossy +ECMP across demands, TE rerouting bounds, and feasibility at engine resolution. +""" + +from __future__ import annotations + +import netgraph_core +import numpy as np +import pytest + +from ngraph.analysis import AnalysisContext +from ngraph.analysis.functions import ( + build_demand_placement_inputs, + demand_placement_analysis, +) +from ngraph.analysis.placement import ( + CACHEABLE_PRESETS, + FLOW_RESOLUTION, + PlacementSummary, + place_demands, +) +from ngraph.model.flow.policy_config import ( + HOP_BY_HOP_PRESETS, + FlowPolicyPreset, + create_flow_policy, +) +from ngraph.model.network import Link, Network, Node + +ECMP = FlowPolicyPreset.SHORTEST_PATHS_ECMP +WCMP = FlowPolicyPreset.SHORTEST_PATHS_WCMP +LOSSY = FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY +TE = FlowPolicyPreset.TE_WCMP_UNLIM + + +def _run(net, demands, excluded_links=None, **kw): + return demand_placement_analysis( + net, set(), set(excluded_links or ()), demands, **kw + ) + + +def _demand(source, target, volume, preset, mode="pairwise", **extra): + return { + "source": source, + "target": target, + "volume": volume, + "mode": mode, + "flow_policy": preset, + **extra, + } + + +@pytest.fixture +def unequal_sources() -> Network: + """S1 -> T cap 100 and S2 -> T cap 10, equal cost.""" + net = Network() + for n in ("S1", "S2", "T"): + net.add_node(Node(n)) + net.add_link(Link("S1", "T", capacity=100, cost=1)) + net.add_link(Link("S2", "T", capacity=10, cost=1)) + return net + + +@pytest.fixture +def far_source() -> Network: + """S1 -> T cost 1 and S2 -> T cost 2, both cap 100.""" + net = Network() + for n in ("S1", "S2", "T"): + net.add_node(Node(n)) + net.add_link(Link("S1", "T", capacity=100, cost=1)) + net.add_link(Link("S2", "T", capacity=100, cost=2)) + return net + + +@pytest.fixture +def parallel_pair() -> Network: + """S -> T over two equal-cost links, cap 10 and cap 100.""" + net = Network() + for n in ("S", "T"): + net.add_node(Node(n)) + net.add_link(Link("S", "T", capacity=10, cost=1)) + net.add_link(Link("S", "T", capacity=100, cost=1)) + return net + + +class TestPresetSets: + def test_lossy_preset_is_hop_by_hop_and_cacheable(self) -> None: + assert LOSSY in HOP_BY_HOP_PRESETS + assert LOSSY in CACHEABLE_PRESETS + assert HOP_BY_HOP_PRESETS <= CACHEABLE_PRESETS + assert TE not in HOP_BY_HOP_PRESETS + + +class TestCombineModeOrigination: + """A combine demand is a virtual source: its reachable members originate even shares.""" + + @pytest.mark.parametrize( + "preset,expected", [(ECMP, 20.0), (WCMP, 65.0), (LOSSY, 65.0)] + ) + def test_hop_by_hop_splits_evenly_across_sources( + self, unequal_sources: Network, preset, expected + ) -> None: + # Even split: 55 from S1, 55 from S2. Lossless ECMP admits the demand as + # a whole at the scale S2 can carry (10/55), so S1 sends 10 too: 20. + # Best-effort and proportional presets carry each share independently: + # 55 + 10 = 65. + r = _run(unequal_sources, [_demand("^S", "^T$", 110, preset, mode="combine")]) + assert len(r.flows) == 1, "a combine demand stays one result entry" + assert r.summary.total_placed == pytest.approx(expected) + assert r.flows[0].placed == pytest.approx(expected) + assert r.flows[0].demand == pytest.approx(110.0) + + def test_hop_by_hop_uses_every_source_not_only_the_nearest( + self, far_source: Network + ) -> None: + r = _run( + far_source, + [_demand("^S", "^T$", 150, ECMP, mode="combine")], + include_flow_details=True, + ) + assert r.summary.total_placed == pytest.approx(150.0) + # 75 at cost 1 from S1 and 75 at cost 2 from S2. + assert r.flows[0].cost_distribution == { + 1.0: pytest.approx(75.0), + 2.0: pytest.approx(75.0), + } + + def test_te_keeps_interchangeable_sources( + self, unequal_sources: Network, far_source: Network + ) -> None: + # TE carries the aggregate with whichever sources have capacity. + r = _run(unequal_sources, [_demand("^S", "^T$", 110, TE, mode="combine")]) + assert r.summary.total_placed == pytest.approx(110.0) + r = _run(far_source, [_demand("^S", "^T$", 150, TE, mode="combine")]) + assert r.summary.total_placed == pytest.approx(150.0) + + def test_used_edges_are_the_union_over_sources( + self, unequal_sources: Network + ) -> None: + r = _run( + unequal_sources, + [_demand("^S", "^T$", 110, ECMP, mode="combine")], + include_used_edges=True, + ) + edges = set(r.flows[0].data["edges"]) + assert edges == {"S1|T|0:fwd", "S2|T|0:fwd"} + + def test_lossless_combine_throttles_the_whole_demand( + self, unequal_sources: Network + ) -> None: + """Pairwise demands are admitted one by one; a combine demand is one demand.""" + combine = _run( + unequal_sources, + [_demand("^S", "^T$", 110, ECMP, mode="combine")], + include_flow_details=True, + ) + pairwise = _run( + unequal_sources, [_demand("^S", "^T$", 110, ECMP, mode="pairwise")] + ) + assert combine.summary.total_placed == pytest.approx(20.0) + assert combine.flows[0].cost_distribution == {1.0: pytest.approx(20.0)} + assert pairwise.summary.total_placed == pytest.approx(65.0) + + def test_lossless_combine_scales_globally_over_shared_links(self) -> None: + """S1 -> M cap 100, S2 -> M cap 20, M -> T cap 60, demand 100. + + Even split 50/50. S2's link admits 20 of 50 (scale 0.4) and the shared + link admits 60 of 100 (scale 0.6); the demand takes the smaller: 40. + Per-source admission would have given S1 50 and S2 10. + """ + net = Network() + for n in ("S1", "S2", "M", "T"): + net.add_node(Node(n)) + net.add_link(Link("S1", "M", capacity=100, cost=1)) + net.add_link(Link("S2", "M", capacity=20, cost=1)) + net.add_link(Link("M", "T", capacity=60, cost=1)) + lossless = _run( + net, + [_demand("^S", "^T$", 100, ECMP, mode="combine")], + include_used_edges=True, + ) + assert lossless.summary.total_placed == pytest.approx(40.0) + assert set(lossless.flows[0].data["edges"]) == { + "S1|M|0:fwd", + "S2|M|0:fwd", + "M|T|0:fwd", + } + lossy = _run( + net, + [_demand("^S", "^T$", 100, LOSSY, mode="combine")], + include_flow_details=True, + ) + assert lossy.summary.total_placed == pytest.approx(60.0) + assert lossy.flows[0].data["dropped_edges"] == { + "S2|M|0:fwd": pytest.approx(30.0), + "M|T|0:fwd": pytest.approx(10.0), + } + + def test_combine_is_a_pool_an_isolated_source_leaves_the_split( + self, unequal_sources: Network + ) -> None: + """The virtual source pools its members: S2 cannot reach T, so S1 + originates the whole volume (110 on a 100 link).""" + lossless = _run( + unequal_sources, + [_demand("^S", "^T$", 110, ECMP, mode="combine")], + excluded_links={"S2|T|0"}, + include_flow_details=True, + ) + assert lossless.summary.total_placed == pytest.approx(100.0) + assert lossless.flows[0].cost_distribution == {1.0: pytest.approx(100.0)} + lossy = _run( + unequal_sources, + [_demand("^S", "^T$", 110, LOSSY, mode="combine")], + excluded_links={"S2|T|0"}, + include_flow_details=True, + ) + assert lossy.summary.total_placed == pytest.approx(100.0) + assert lossy.flows[0].data["dropped_edges"] == { + "S1|T|0:fwd": pytest.approx(10.0) + } + wcmp = _run( + unequal_sources, + [_demand("^S", "^T$", 110, WCMP, mode="combine")], + excluded_links={"S2|T|0"}, + ) + assert wcmp.summary.total_placed == pytest.approx(100.0) + + def test_pool_with_no_reachable_member_places_nothing( + self, unequal_sources: Network + ) -> None: + for preset in (ECMP, LOSSY, WCMP): + r = _run( + unequal_sources, + [_demand("^S", "^T$", 110, preset, mode="combine")], + excluded_links={"S1|T|0", "S2|T|0"}, + ) + assert r.summary.total_placed == pytest.approx(0.0) + assert r.summary.dropped_flows == 1 + + def test_fixed_matrix_view_is_per_group(self, unequal_sources: Network) -> None: + """Independent originators: per-source demands, an isolated one is unserved.""" + r = _run( + unequal_sources, + [ + _demand( + {"path": "^S", "group_by": "name"}, + "^T$", + 110, + ECMP, + mode="combine", + group_mode="per_group", + ) + ], + excluded_links={"S2|T|0"}, + ) + assert len(r.flows) == 2 + assert r.summary.total_placed == pytest.approx(55.0) + assert r.summary.dropped_flows == 1 + + def test_msd_probes_reuse_the_fanout_dag(self, unequal_sources: Network) -> None: + ctx, expansion, ids = build_demand_placement_inputs( + unequal_sources, [_demand("^S", "^T$", 110, ECMP, mode="combine")] + ) + cache: dict = {} + for alpha in (1.0, 0.1): + fg = netgraph_core.FlowGraph(ctx.multidigraph) + result = place_demands( + expansion.demands, + [110.0 * alpha], + fg, + ctx, + ctx.build_node_mask(), + ctx.build_edge_mask(), + resolved_ids=ids, + dag_cache=cache, + ) + assert result.summary.total_placed == pytest.approx( + min(20.0, 110.0 * alpha) + ) + assert len(cache) == 1 + + def test_share_goes_to_nearest_target(self) -> None: + """A source's share is routed to its closest target under IGP.""" + net = Network() + for n in ("S", "T1", "T2"): + net.add_node(Node(n)) + net.add_link(Link("S", "T1", capacity=100, cost=1)) + net.add_link(Link("S", "T2", capacity=100, cost=5)) + r = _run( + net, + [_demand("^S$", "^T", 50, ECMP, mode="combine")], + include_used_edges=True, + ) + assert r.summary.total_placed == pytest.approx(50.0) + assert r.flows[0].data["edges"] == ["S|T1|0:fwd"] + + def test_msd_inputs_carry_members(self, unequal_sources: Network) -> None: + ctx, expansion, ids = build_demand_placement_inputs( + unequal_sources, [_demand("^S", "^T$", 110, ECMP, mode="combine")] + ) + assert expansion.demands[0].src_members == ("S1", "S2") + fg = netgraph_core.FlowGraph(ctx.multidigraph) + result = place_demands( + expansion.demands, + [d.volume for d in expansion.demands], + fg, + ctx, + ctx.build_node_mask(), + ctx.build_edge_mask(), + resolved_ids=ids, + ) + assert result.summary.total_placed == pytest.approx(20.0) + + +class TestEcmpAcrossDemands: + """ECMP hashes over the topology's next hops regardless of load.""" + + def test_single_demand_admits_to_the_weakest_member( + self, parallel_pair: Network + ) -> None: + r = _run(parallel_pair, [_demand("^S$", "^T$", 30, ECMP)]) + assert r.flows[0].placed == pytest.approx(20.0) + + def test_saturated_member_blocks_later_lossless_demand( + self, parallel_pair: Network + ) -> None: + r = _run( + parallel_pair, + [ + _demand("^S$", "^T$", 20, ECMP, priority=0), + _demand("^S$", "^T$", 10, ECMP, priority=1), + ], + ) + assert r.flows[0].placed == pytest.approx(20.0) + assert r.flows[1].placed == pytest.approx(0.0), ( + "half of the second demand would be hashed onto the saturated 10-unit " + "link and lost, so nothing is admitted losslessly" + ) + + def test_lossy_delivers_what_survives_and_reports_drops( + self, parallel_pair: Network + ) -> None: + r = _run( + parallel_pair, + [_demand("^S$", "^T$", 100, LOSSY)], + include_flow_details=True, + ) + entry = r.flows[0] + assert entry.placed == pytest.approx(60.0) + assert entry.dropped == pytest.approx(40.0) + assert entry.data["dropped_edges"] == {"S|T|0:fwd": pytest.approx(40.0)} + assert entry.cost_distribution == {1.0: pytest.approx(60.0)} + + def test_lossy_later_demand_still_loses_its_hashed_share( + self, parallel_pair: Network + ) -> None: + r = _run( + parallel_pair, + [ + _demand("^S$", "^T$", 20, LOSSY, priority=0), + _demand("^S$", "^T$", 10, LOSSY, priority=1), + ], + include_flow_details=True, + ) + assert r.flows[0].placed == pytest.approx(20.0) + assert r.flows[0].data == {}, "nothing dropped, so no dropped_edges key" + assert r.flows[1].placed == pytest.approx(5.0) + assert r.flows[1].data["dropped_edges"] == {"S|T|0:fwd": pytest.approx(5.0)} + + def test_lossy_used_edges_include_links_that_carried_then_dropped(self) -> None: + net = Network() + for n in ("S", "M", "T"): + net.add_node(Node(n)) + net.add_link(Link("S", "M", capacity=100, cost=1)) + net.add_link(Link("M", "T", capacity=30, cost=1)) + r = _run( + net, + [_demand("^S$", "^T$", 100, LOSSY)], + include_used_edges=True, + include_flow_details=True, + ) + assert r.flows[0].placed == pytest.approx(30.0) + assert r.flows[0].data["edges"] == ["M|T|0:fwd", "S|M|0:fwd"] + assert r.flows[0].data["dropped_edges"] == {"M|T|0:fwd": pytest.approx(70.0)} + + def test_dropped_edges_absent_without_flow_details( + self, parallel_pair: Network + ) -> None: + r = _run(parallel_pair, [_demand("^S$", "^T$", 100, LOSSY)]) + assert r.flows[0].placed == pytest.approx(60.0) + assert "dropped_edges" not in r.flows[0].data + + +class TestTeReroutingBound: + def test_te_reroutes_over_more_than_100_cost_tiers(self) -> None: + net = Network() + net.add_node(Node("S")) + net.add_node(Node("T")) + tiers = 150 + for i in range(tiers): + m = f"M{i:03d}" + net.add_node(Node(m)) + net.add_link(Link("S", m, capacity=1, cost=1 + i)) + net.add_link(Link(m, "T", capacity=1, cost=1)) + r = _run(net, [_demand("^S$", "^T$", tiers, TE)], include_flow_details=True) + assert r.flows[0].placed == pytest.approx(float(tiers)) + assert len(r.flows[0].cost_distribution) == tiers + + +class TestFeasibilityResolution: + def test_shortfall_within_engine_resolution_is_feasible(self) -> None: + s = PlacementSummary( + total_demand=1.0, + total_placed=1.0 - FLOW_RESOLUTION / 2, + max_shortfall=FLOW_RESOLUTION / 2, + ) + assert s.is_feasible + s = PlacementSummary(total_demand=1.0, total_placed=0.999, max_shortfall=0.001) + assert not s.is_feasible + + def test_many_lsps_on_tiny_volume_are_feasible(self) -> None: + """256 LSPs cannot each carry less than 1/4096; the residue is the engine's, not the network's.""" + net = Network() + for n in ("S", "T"): + net.add_node(Node(n)) + for i in range(8): + m = f"M{i}" + net.add_node(Node(m)) + net.add_link(Link("S", m, capacity=100, cost=1)) + net.add_link(Link(m, "T", capacity=100, cost=1)) + ctx, expansion, ids = build_demand_placement_inputs( + net, [_demand("^S$", "^T$", 0.05, FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP)] + ) + fg = netgraph_core.FlowGraph(ctx.multidigraph) + result = place_demands( + expansion.demands, + [0.05], + fg, + ctx, + ctx.build_node_mask(), + ctx.build_edge_mask(), + resolved_ids=ids, + ) + assert 0 < result.summary.max_shortfall <= FLOW_RESOLUTION + assert result.summary.is_feasible + + def test_max_shortfall_is_the_largest_per_demand_gap(self) -> None: + net = Network() + for n in ("A", "B"): + net.add_node(Node(n)) + net.add_link(Link("A", "B", capacity=10, cost=1)) + ctx, expansion, ids = build_demand_placement_inputs( + net, + [ + _demand("^A$", "^B$", 8, ECMP), + _demand("^A$", "^B$", 8, ECMP, priority=1), + ], + ) + fg = netgraph_core.FlowGraph(ctx.multidigraph) + result = place_demands( + expansion.demands, + [8.0, 8.0], + fg, + ctx, + ctx.build_node_mask(), + ctx.build_edge_mask(), + resolved_ids=ids, + ) + # The first demand takes 8, the second gets the 2 units of headroom. + assert result.summary.max_shortfall == pytest.approx(6.0) + assert not result.summary.is_feasible + + +class TestEnginesAgreeUnderContention: + """FlowPolicy built from a hop-by-hop preset places what the cached engine places.""" + + @pytest.fixture + def detour(self) -> Network: + net = Network() + for n in ("A", "B", "C"): + net.add_node(Node(n)) + net.add_link(Link("A", "B", capacity=10, cost=1)) + net.add_link(Link("A", "C", capacity=100, cost=5)) + net.add_link(Link("C", "B", capacity=100, cost=5)) + return net + + @pytest.mark.parametrize("preset", [ECMP, WCMP, LOSSY]) + def test_flow_policy_does_not_reroute(self, detour: Network, preset) -> None: + cached = _run(detour, [_demand("^A$", "^B$", 50, preset)]) + assert cached.flows[0].placed == pytest.approx(10.0) + + ctx = AnalysisContext.from_network(detour) + fg = netgraph_core.FlowGraph(ctx.multidigraph) + policy = create_flow_policy( + ctx.algorithms, + ctx.handle, + preset, + node_mask=ctx.build_node_mask(), + edge_mask=ctx.build_edge_mask(), + ) + placed, _ = policy.place_demand( + fg, ctx.node_mapper.to_id("A"), ctx.node_mapper.to_id("B"), 0, 50.0 + ) + assert placed == pytest.approx(10.0) + assert {float(v[2]) for v in policy.flows.values()} == {1.0}, ( + "the flow stays on the cost-1 path" + ) + + def test_te_flow_policy_still_reroutes(self, detour: Network) -> None: + ctx = AnalysisContext.from_network(detour) + fg = netgraph_core.FlowGraph(ctx.multidigraph) + policy = create_flow_policy(ctx.algorithms, ctx.handle, TE) + placed, _ = policy.place_demand( + fg, ctx.node_mapper.to_id("A"), ctx.node_mapper.to_id("B"), 0, 50.0 + ) + assert placed == pytest.approx(50.0) + + def test_masks_still_apply_to_hop_by_hop_policies(self, detour: Network) -> None: + ctx = AnalysisContext.from_network(detour) + edge_mask = ctx.build_edge_mask({"A|B|0"}) + fg = netgraph_core.FlowGraph(ctx.multidigraph) + policy = create_flow_policy( + ctx.algorithms, + ctx.handle, + ECMP, + node_mask=ctx.build_node_mask(), + edge_mask=edge_mask, + ) + placed, _ = policy.place_demand( + fg, ctx.node_mapper.to_id("A"), ctx.node_mapper.to_id("B"), 0, 50.0 + ) + assert placed == pytest.approx(50.0), ( + "with A-B failed, the cost-10 path is the shortest path" + ) + assert isinstance(edge_mask, np.ndarray) + + +class TestUnservedDemands: + def test_demand_that_places_nothing_is_never_feasible(self) -> None: + """Below the engine's resolution a zero placement is not 'within resolution'.""" + net = Network() + for n in ("A", "B", "C"): + net.add_node(Node(n)) + net.add_link(Link("A", "B", capacity=10, cost=1)) + ctx, expansion, ids = build_demand_placement_inputs( + net, [_demand("^A$", "^B$", 1e-6, ECMP), _demand("^A$", "^C$", 1e-6, ECMP)] + ) + fg = netgraph_core.FlowGraph(ctx.multidigraph) + result = place_demands( + expansion.demands, + [1e-6, 1e-6], + fg, + ctx, + ctx.build_node_mask(), + ctx.build_edge_mask(), + resolved_ids=ids, + ) + assert result.summary.max_shortfall <= FLOW_RESOLUTION + assert result.summary.unserved_demands == 2 + assert not result.summary.is_feasible + + def test_summary_defaults_keep_old_constructor_working(self) -> None: + assert PlacementSummary(total_demand=0.0, total_placed=0.0).is_feasible + + +class TestLossyWithStaticPaths: + def test_pinned_routes_carry_what_fits(self) -> None: + """A -> B direct cap 10 and A -> C -> B cap 100, demand 50 pinned to both. + + Lossy: 25 offered per route, 10 + 25 delivered. Lossless ECMP: equal + carried share, bottleneck 10, so 20. + """ + net = Network() + for n in ("A", "B", "C"): + net.add_node(Node(n)) + net.add_link(Link("A", "B", capacity=10, cost=1)) + net.add_link(Link("A", "C", capacity=100, cost=1)) + net.add_link(Link("C", "B", capacity=100, cost=1)) + routes = [["A", "B"], ["A", "C", "B"]] + lossy = _run( + net, + [_demand("^A$", "^B$", 50, LOSSY, static_paths=routes)], + include_flow_details=True, + ) + assert lossy.flows[0].placed == pytest.approx(35.0) + assert lossy.flows[0].dropped == pytest.approx(15.0) + assert lossy.flows[0].cost_distribution == { + 1.0: pytest.approx(10.0), + 2.0: pytest.approx(25.0), + } + lossless = _run(net, [_demand("^A$", "^B$", 50, ECMP, static_paths=routes)]) + assert lossless.flows[0].placed == pytest.approx(20.0, abs=1e-3) diff --git a/tests/model/demand/test_builder.py b/tests/model/demand/test_builder.py index 69c90b4..0ab9cd7 100644 --- a/tests/model/demand/test_builder.py +++ b/tests/model/demand/test_builder.py @@ -262,3 +262,11 @@ def test_build_demand_set_rejects_bool_flow_policy(): with pytest.raises(ValueError, match="Invalid flow_policy"): build_demand_set(raw) + + +def test_coerce_flow_policy_lossy_ecmp_preset(): + assert coerce_flow_policy(6) == FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY + assert ( + coerce_flow_policy("shortest_paths_ecmp_lossy") + == FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY + ) diff --git a/tests/model/flow/test_policy_config.py b/tests/model/flow/test_policy_config.py index 1d07312..e8f7375 100644 --- a/tests/model/flow/test_policy_config.py +++ b/tests/model/flow/test_policy_config.py @@ -157,3 +157,70 @@ def test_flow_policy_preset_from_name(): preset = FlowPolicyPreset["TE_ECMP_16_LSP"] assert preset == FlowPolicyPreset.TE_ECMP_16_LSP + + +# --------------------------------------------------------------------------- +# preset_config: the single mapping both placement engines read from +# --------------------------------------------------------------------------- + + +def test_preset_config_hop_by_hop_presets_are_cost_only_single_pass(): + """IGP presets route on cost alone and place once, in both engines.""" + from ngraph.model.flow.policy_config import HOP_BY_HOP_PRESETS, preset_config + + expected_placement = { + FlowPolicyPreset.SHORTEST_PATHS_ECMP: netgraph_core.FlowPlacement.EQUAL_BALANCED_FIXED, + FlowPolicyPreset.SHORTEST_PATHS_WCMP: netgraph_core.FlowPlacement.PROPORTIONAL, + FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY: netgraph_core.FlowPlacement.EQUAL_BALANCED_LOSSY, + } + assert set(expected_placement) == set(HOP_BY_HOP_PRESETS) + for preset, placement in expected_placement.items(): + cfg = preset_config(preset) + assert cfg.require_capacity is False, preset + assert cfg.selection.require_capacity is False, preset + assert cfg.shortest_path is True, preset + assert cfg.max_flow_count == 1, preset + assert cfg.flow_placement == placement, preset + + +def test_preset_config_te_presets_are_capacity_aware(): + from ngraph.model.flow.policy_config import preset_config + + for preset in ( + FlowPolicyPreset.TE_WCMP_UNLIM, + FlowPolicyPreset.TE_ECMP_16_LSP, + FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP, + ): + cfg = preset_config(preset) + assert cfg.require_capacity is True, preset + assert cfg.selection.require_capacity is True, preset + assert cfg.shortest_path is False, preset + assert preset_config(FlowPolicyPreset.TE_WCMP_UNLIM).max_flow_count is None + assert preset_config(FlowPolicyPreset.TE_ECMP_16_LSP).min_flow_count == 16 + assert preset_config(FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP).max_flow_count == 256 + + +def test_preset_config_returns_a_fresh_object(): + from ngraph.model.flow.policy_config import preset_config + + a = preset_config(FlowPolicyPreset.SHORTEST_PATHS_ECMP) + a.max_flow_count = 7 + assert preset_config(FlowPolicyPreset.SHORTEST_PATHS_ECMP).max_flow_count == 1 + + +def test_lossy_preset_value_and_policy(simple_graph): + algs, graph_handle, _ = simple_graph + assert FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY == 6 + policy = create_flow_policy( + algs, graph_handle, FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY + ) + assert policy is not None + + +def test_static_paths_disable_single_pass_mode(simple_graph): + """Core rejects shortest_path with pinned routes, so the factory clears it.""" + algs, graph_handle, _ = simple_graph + policy = create_flow_policy( + algs, graph_handle, FlowPolicyPreset.SHORTEST_PATHS_ECMP, static_path_count=2 + ) + assert policy is not None diff --git a/tests/workflow/test_msd_resolution.py b/tests/workflow/test_msd_resolution.py new file mode 100644 index 0000000..1d5622e --- /dev/null +++ b/tests/workflow/test_msd_resolution.py @@ -0,0 +1,87 @@ +"""MaximumSupportedDemand feasibility at the engine's numeric resolution.""" + +from __future__ import annotations + +import pytest + +from ngraph.model.demand.matrix import DemandSet +from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.flow.policy_config import FlowPolicyPreset +from ngraph.model.network import Link, Network, Node +from ngraph.results import Results +from ngraph.workflow.maximum_supported_demand_step import MaximumSupportedDemand + + +class _Scenario: + def __init__(self, network: Network, demand_set: DemandSet) -> None: + self.network = network + self.demand_set = demand_set + self.results = Results() + + +def _fabric(leaves: int, spines: int) -> Network: + net = Network() + for i in range(leaves): + net.add_node(Node(f"leaf{i:02d}")) + for j in range(spines): + net.add_node(Node(f"spine{j:02d}")) + for i in range(leaves): + for j in range(spines): + net.add_link(Link(f"leaf{i:02d}", f"spine{j:02d}", capacity=100.0, cost=1)) + return net + + +def _run_msd(net: Network, demands: list[TrafficDemand], **params) -> dict: + ds = DemandSet() + ds.add("default", demands) + scenario = _Scenario(net, ds) + step = MaximumSupportedDemand(demand_set="default", **params) + step.name = "msd" + scenario.results.enter_step("msd") + try: + step.run(scenario) # type: ignore[arg-type] + finally: + scenario.results.exit_step() + return scenario.results.get_step("msd")["data"] + + +def test_many_lsps_over_small_pairwise_volumes_find_alpha_star(): + """256 LSPs per pair on 0.03-unit demands quantize at 1/4096; that is not infeasibility.""" + net = _fabric(8, 4) + demands = [ + TrafficDemand( + source=f"^leaf{i:02d}$", + target="^leaf", + volume=1.0, + mode="pairwise", + flow_policy=FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP, + id=f"d{i}", + ) + for i in range(8) + ] + data = _run_msd( + net, + demands, + alpha_start=1.0, + resolution=0.5, + max_bracket_iters=6, + max_bisect_iters=4, + ) + # Each leaf sends 1.0 over 4 uplinks of 100: alpha_star sits between 64 and 512. + assert 64.0 <= data["alpha_star"] <= 512.0 + assert data["probes"][0]["feasible"] is True + + +def test_unreachable_target_reports_ratio_in_error(): + net = Network() + for n in ("A", "B", "C"): + net.add_node(Node(n)) + net.add_link(Link("A", "B", capacity=10, cost=1)) + demands = [ + TrafficDemand(source="^A$", target="^B$", volume=1.0, mode="pairwise", id="ok"), + TrafficDemand( + source="^A$", target="^C$", volume=1.0, mode="pairwise", id="isolated" + ), + ] + with pytest.raises(ValueError, match=r"best placement ratio over probes 0\.5000"): + _run_msd(net, demands, alpha_start=1.0, max_bracket_iters=3, max_bisect_iters=2) diff --git a/tests/workflow/test_placement_parallelism.py b/tests/workflow/test_placement_parallelism.py new file mode 100644 index 0000000..2c218c5 --- /dev/null +++ b/tests/workflow/test_placement_parallelism.py @@ -0,0 +1,64 @@ +"""resolve_placement_parallelism: 'auto' means threads only where they help.""" + +from __future__ import annotations + +import os + +import pytest + +from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.flow.policy_config import FlowPolicyPreset +from ngraph.workflow import traffic_matrix_placement_step as tm +from ngraph.workflow.traffic_matrix_placement_step import resolve_placement_parallelism + + +def _td(preset): + return TrafficDemand(source="A", target="B", volume=1.0, flow_policy=preset) + + +@pytest.fixture(autouse=True) +def gil_interpreter(monkeypatch): + monkeypatch.setattr(tm, "_python_is_free_threaded", lambda: False) + + +def test_explicit_worker_count_is_honoured(): + assert ( + resolve_placement_parallelism(4, [_td(FlowPolicyPreset.SHORTEST_PATHS_ECMP)]) + == 4 + ) + assert resolve_placement_parallelism(1, [_td(FlowPolicyPreset.TE_ECMP_16_LSP)]) == 1 + + +@pytest.mark.parametrize( + "preset", + [ + None, # default preset is SHORTEST_PATHS_ECMP + FlowPolicyPreset.SHORTEST_PATHS_ECMP, + FlowPolicyPreset.SHORTEST_PATHS_WCMP, + FlowPolicyPreset.SHORTEST_PATHS_ECMP_LOSSY, + FlowPolicyPreset.TE_WCMP_UNLIM, + ], +) +def test_auto_is_serial_for_cacheable_presets(preset): + assert resolve_placement_parallelism("auto", [_td(preset)]) == 1 + + +@pytest.mark.parametrize( + "preset", [FlowPolicyPreset.TE_ECMP_16_LSP, FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP] +) +def test_auto_uses_cpu_count_for_engine_bound_presets(preset): + demands = [_td(FlowPolicyPreset.SHORTEST_PATHS_ECMP), _td(preset)] + assert resolve_placement_parallelism("auto", demands) == max(1, os.cpu_count() or 1) + + +def test_auto_uses_cpu_count_when_free_threaded(monkeypatch): + monkeypatch.setattr(tm, "_python_is_free_threaded", lambda: True) + demands = [_td(FlowPolicyPreset.SHORTEST_PATHS_ECMP)] + assert resolve_placement_parallelism("auto", demands) == max(1, os.cpu_count() or 1) + + +def test_invalid_values_still_raise(): + with pytest.raises(ValueError): + resolve_placement_parallelism("many", []) + with pytest.raises(ValueError): + resolve_placement_parallelism(0, [])