bugfixing and logging

Author: networmix

ngraph/demand/__init__.py

@@ -108,12 +108,25 @@ def place(
             # If max_fraction <= 0, do not place any new volume (unless volume is infinite).
             to_place = self.volume if self.volume == float("inf") else 0.0
 
-        # Delegate flow placement
+        # Ensure we request at least MIN_FLOW when there is meaningful leftover
+        if 0.0 < to_place < MIN_FLOW:
+            to_place = min(self.volume - self.placed_demand, MIN_FLOW)
+
+        # Delegate flow placement (do not force min_flow threshold here; policy handles it)
+        # Use a demand-unique flow_class to avoid collisions across different
+        # Demand instances that share the same numerical demand_class.
+        demand_unique_flow_class = (
+            self.demand_class,
+            self.src_node,
+            self.dst_node,
+            id(self),
+        )
+
         self.flow_policy.place_demand(
             flow_graph,
             self.src_node,
             self.dst_node,
-            self.demand_class,
+            demand_unique_flow_class,
             to_place,
         )
 

ngraph/demand/manager/expand.py

@@ -108,9 +108,15 @@ def _expand_combine(
     with infinite-capacity, zero-cost edges, and creates one aggregate
     `Demand` from pseudo-source to pseudo-sink with the full volume.
     """
-    # Flatten the source and sink node lists
-    src_nodes = [node for group_nodes in src_groups.values() for node in group_nodes]
-    dst_nodes = [node for group_nodes in snk_groups.values() for node in group_nodes]
+    # Flatten and sort source and sink node lists for deterministic order
+    src_nodes = sorted(
+        (node for group_nodes in src_groups.values() for node in group_nodes),
+        key=lambda n: n.name,
+    )
+    dst_nodes = sorted(
+        (node for group_nodes in snk_groups.values() for node in group_nodes),
+        key=lambda n: n.name,
+    )
 
     if not src_nodes or not dst_nodes or graph is None:
         return
@@ -161,17 +167,22 @@ def _expand_pairwise(
     Creates one `Demand` for each valid source-destination pair (excluding
     self-pairs) and splits total volume evenly across pairs.
     """
-    # Flatten the source and sink node lists
-    src_nodes = [node for group_nodes in src_groups.values() for node in group_nodes]
-    dst_nodes = [node for group_nodes in snk_groups.values() for node in group_nodes]
-
-    # Generate all valid (src, dst) pairs
-    valid_pairs = [
-        (s_node, t_node)
-        for s_node in src_nodes
-        for t_node in dst_nodes
-        if s_node.name != t_node.name
-    ]
+    # Flatten and sort source and sink node lists for deterministic order
+    src_nodes = sorted(
+        (node for group_nodes in src_groups.values() for node in group_nodes),
+        key=lambda n: n.name,
+    )
+    dst_nodes = sorted(
+        (node for group_nodes in snk_groups.values() for node in group_nodes),
+        key=lambda n: n.name,
+    )
+
+    # Generate all valid (src, dst) pairs in deterministic lexicographic order
+    valid_pairs = []
+    for s_node in src_nodes:
+        for t_node in dst_nodes:
+            if s_node.name != t_node.name:
+                valid_pairs.append((s_node, t_node))
     pair_count = len(valid_pairs)
     if pair_count == 0:
         return

ngraph/demand/manager/manager.py

@@ -65,8 +65,13 @@ class TrafficManager:
           on a configurable 'mode' ("combine" or "pairwise").
       3) Each Demand is associated with a FlowPolicy, which handles how flows
          are placed (split across paths, balancing, etc.).
-      4) Provides methods to place all demands incrementally with optional
-         re-optimization, reset usage, and retrieve flow/usage summaries.
+         4) Provides methods to place all demands incrementally with optional
+          re-optimization, reset usage, and retrieve flow/usage summaries.
+
+     Auto rounds semantics:
+       - placement_rounds="auto" performs up to a small number of fairness passes
+         (at most 3), with early stop when diminishing returns are detected. Each
+         pass asks the scheduler to place full leftovers without step splitting.
 
     In particular:
       - 'combine' mode:
@@ -172,21 +177,57 @@ def place_all_demands(
             raise RuntimeError("Graph not built yet. Call build_graph() first.")
 
         if isinstance(placement_rounds, str) and placement_rounds.lower() == "auto":
-            placement_rounds = self._estimate_rounds()
+            # Simple, reliable auto: up to 3 passes with early stop.
+            from ngraph.algorithms.base import MIN_FLOW
+
+            total_placed = 0.0
+            max_auto_rounds = 3
+            for _ in range(max_auto_rounds):
+                placed_now = place_demands_round_robin(
+                    graph=self.graph,
+                    demands=self.demands,
+                    placement_rounds=1,
+                    reoptimize_after_each_round=False,
+                )
+                total_placed += placed_now
 
-        # Ensure placement_rounds is an int for range() and arithmetic operations
-        placement_rounds_int = (
-            int(placement_rounds)
-            if isinstance(placement_rounds, str)
-            else placement_rounds
-        )
+                # Early stops: no progress or negligible leftover
+                if placed_now < MIN_FLOW:
+                    break
 
-        total_placed = place_demands_round_robin(
-            graph=self.graph,
-            demands=self.demands,
-            placement_rounds=placement_rounds_int,
-            reoptimize_after_each_round=reoptimize_after_each_round,
-        )
+                leftover_total = sum(
+                    max(0.0, d.volume - d.placed_demand) for d in self.demands
+                )
+                if leftover_total < 0.05 * placed_now:
+                    break
+
+                # Fairness check: if served ratios are already close, stop
+                served = [
+                    (d.placed_demand / d.volume)
+                    for d in self.demands
+                    if d.volume > 0.0 and (d.volume - d.placed_demand) >= MIN_FLOW
+                ]
+                if served:
+                    s_min, s_max = min(served), max(served)
+                    if s_max <= 0.0 or s_min >= 0.8 * s_max:
+                        break
+        else:
+            # Ensure placement_rounds is an int for range() and arithmetic operations
+            placement_rounds_int = (
+                int(placement_rounds)
+                if isinstance(placement_rounds, str)
+                else placement_rounds
+            )
+
+            if placement_rounds_int <= 0:
+                raise ValueError("placement_rounds must be positive")
+
+            total_placed = place_demands_round_robin(
+                graph=self.graph,
+                demands=self.demands,
+                placement_rounds=placement_rounds_int,
+                reoptimize_after_each_round=reoptimize_after_each_round,
+            )
 
         # Update each TrafficDemand's placed volume
         for td in self._get_traffic_demands():

ngraph/demand/manager/schedule.py

@@ -42,31 +42,53 @@ def place_demands_round_robin(
 
     for priority_class in sorted_priorities:
         demands_in_class = prio_map[priority_class]
+        placed_before_class = sum(d.placed_demand for d in demands_in_class)
 
+        # Unified fairness loop: attempt to place full leftover per demand each round.
+        # For rounds > 0, reorder by least-served ratio to improve fairness.
+        reopt_attempted = False
         for round_idx in range(placement_rounds):
             placed_in_this_round = 0.0
-            rounds_left = placement_rounds - round_idx
 
-            for demand in demands_in_class:
+            if round_idx == 0:
+                iteration_order = list(demands_in_class)
+            else:
+                iteration_order = sorted(
+                    demands_in_class,
+                    key=lambda d: (
+                        (d.placed_demand / d.volume) if d.volume > 0 else 1.0,
+                        d.placed_demand,
+                    ),
+                )
+
+            for demand in iteration_order:
                 leftover = demand.volume - demand.placed_demand
                 if leftover < base.MIN_FLOW:
                     continue
 
-                step_to_place = leftover / float(rounds_left)
-                placed_now, _remain = demand.place(
-                    flow_graph=graph,
-                    max_placement=step_to_place,
-                )
-                total_placed += placed_now
+                placed_now, _remain = demand.place(flow_graph=graph)
                 placed_in_this_round += placed_now
 
             if reoptimize_after_each_round and placed_in_this_round > 0.0:
                 _reoptimize_priority_demands(graph, demands_in_class)
 
-            # If no progress was made, no need to continue extra rounds
             if placed_in_this_round < base.MIN_FLOW:
+                any_leftover = any(
+                    (d.volume - d.placed_demand) >= base.MIN_FLOW
+                    for d in demands_in_class
+                )
+                if not any_leftover:
+                    break
+                if not reopt_attempted:
+                    _reoptimize_priority_demands(graph, demands_in_class)
+                    reopt_attempted = True
+                    continue
                 break
 
+        # Add only the net increase for this class to avoid double counting
+        placed_after_class = sum(d.placed_demand for d in demands_in_class)
+        total_placed += max(0.0, placed_after_class - placed_before_class)
+
     return total_placed
 
 
@@ -82,11 +104,18 @@ def _reoptimize_priority_demands(
             continue
         placed_volume = dmd.placed_demand
         dmd.flow_policy.remove_demand(graph)
+        # Use a demand-unique flow_class key to keep policy flows consistent
+        flow_class_key = (
+            dmd.demand_class,
+            dmd.src_node,
+            dmd.dst_node,
+            id(dmd),
+        )
         dmd.flow_policy.place_demand(
             graph,
             dmd.src_node,
             dmd.dst_node,
-            dmd.demand_class,
+            flow_class_key,
             placed_volume,
         )
         dmd.placed_demand = dmd.flow_policy.placed_demand