fix of a max flow bug

Author: networmix

ngraph/algorithms/capacity.py

@@ -100,8 +100,10 @@ def _init_graph_data(
                 residual_cap[node][adj_node] = (
                     fwd_capacity if fwd_capacity >= MIN_CAP else 0.0
                 )
-                # Reverse edge in the BFS sense starts with 0 capacity
-                residual_cap[adj_node][node] = 0.0
+                # Reverse edge in the BFS sense starts with 0 capacity.
+                # Do not overwrite if a positive capacity was already assigned
+                # when processing the opposite orientation (bidirectional links).
+                residual_cap[adj_node].setdefault(node, 0.0)
 
             elif flow_placement == FlowPlacement.EQUAL_BALANCED:
                 # min(...) * number_of_parallel_edges
@@ -112,8 +114,9 @@ def _init_graph_data(
                     )
                 else:
                     residual_cap[adj_node][node] = 0.0
-                # The forward edge in reversed orientation starts at 0 capacity
-                residual_cap[node][adj_node] = 0.0
+                # The forward edge in reversed orientation starts at 0 capacity.
+                # Do not overwrite non-zero value if already assigned earlier.
+                residual_cap[node].setdefault(adj_node, 0.0)
 
             else:
                 raise ValueError(f"Unsupported flow placement: {flow_placement}")

tests/algorithms/test_calc_capacity.py

@@ -3,7 +3,11 @@
 
 import pytest
 
-from ngraph.algorithms.capacity import FlowPlacement, calc_graph_capacity
+from ngraph.algorithms.capacity import (
+    FlowPlacement,
+    _init_graph_data,
+    calc_graph_capacity,
+)
 from ngraph.algorithms.flow_init import init_flow_graph
 from ngraph.algorithms.spf import spf
 from ngraph.graph.strict_multidigraph import EdgeID, NodeID, StrictMultiDiGraph
@@ -391,6 +395,138 @@ def test_calc_graph_capacity_self_loop_with_other_edges(self):
         )
         assert max_flow_regular == 5.0  # Should be limited by A->B capacity
 
+    def test_reverse_residual_init_graph_data_proportional(self):
+        """_init_graph_data should expose dst->leaf residual capacity in PROPORTIONAL.
+
+        Build a tiny graph with forward edges leaf->dc and dc->sink, and reverse dc->leaf.
+        SPF pred contains dc predecessors (leaves) and sink predecessor (dc).
+        The reversed residual must have positive capacity dc->leaf equal to sum(capacity - flow).
+        """
+        g = StrictMultiDiGraph()
+        for n in ("source", "A/dc", "A/leaf", "sink"):
+            g.add_node(n)
+        # Forward edges
+        e1 = g.add_edge("A/leaf", "A/dc", capacity=10.0, cost=1, flow=0.0, flows={})
+        g.add_edge("A/dc", "sink", capacity=float("inf"), cost=0, flow=0.0, flows={})
+        # Reverse edge to simulate bidirectional link
+        g.add_edge("A/dc", "A/leaf", capacity=10.0, cost=1, flow=0.0, flows={})
+
+        # SPF-like predecessor dict: include both directions present in graph
+        # sink<-A/dc, A/dc<-A/leaf, and A/leaf<-A/dc (reverse link)
+        pred: PredDict = {
+            "source": {},
+            "A/dc": {"A/leaf": [e1]},
+            "A/leaf": {"A/dc": list(g.edges_between("A/dc", "A/leaf"))},
+            "sink": {"A/dc": list(g.edges_between("A/dc", "sink"))},
+        }
+
+        # Run init
+        succ, levels, residual_cap, flow_dict = _init_graph_data(
+            g,
+            pred,
+            init_node="sink",
+            flow_placement=FlowPlacement.PROPORTIONAL,
+            capacity_attr="capacity",
+            flow_attr="flow",
+        )
+        # residuals must reflect both forward directions, and zero-init must not overwrite
+        assert residual_cap["A/dc"]["A/leaf"] == 10.0
+        assert residual_cap["A/leaf"]["A/dc"] == 10.0
+
+    def test_reverse_residual_init_graph_data_equal_balanced(self):
+        """_init_graph_data should set reverse residual in EQUAL_BALANCED as min*count.
+
+        With two parallel edges leaf->dc with caps (5, 7), min=5 and count=2 -> reverse cap = 10.
+        """
+        g = StrictMultiDiGraph()
+        for n in ("source", "A/dc", "A/leaf", "sink"):
+            g.add_node(n)
+        # Two parallel forward edges leaf->dc
+        e1 = g.add_edge("A/leaf", "A/dc", capacity=5.0, cost=1, flow=0.0, flows={})
+        e2 = g.add_edge("A/leaf", "A/dc", capacity=7.0, cost=1, flow=0.0, flows={})
+        g.add_edge("A/dc", "sink", capacity=float("inf"), cost=0, flow=0.0, flows={})
+        # Reverse edge present too
+        g.add_edge("A/dc", "A/leaf", capacity=7.0, cost=1, flow=0.0, flows={})
+
+        pred: PredDict = {
+            "source": {},
+            "A/dc": {"A/leaf": [e1, e2]},
+            "sink": {"A/dc": list(g.edges_between("A/dc", "sink"))},
+        }
+
+        succ, levels, residual_cap, flow_dict = _init_graph_data(
+            g,
+            pred,
+            init_node="sink",
+            flow_placement=FlowPlacement.EQUAL_BALANCED,
+            capacity_attr="capacity",
+            flow_attr="flow",
+        )
+        # In EQUAL_BALANCED, the reverse residual is assigned on leaf->dc orientation (adj->node)
+        # i.e., residual_cap[leaf][dc] = min(capacities) * count = 5*2 = 10
+        assert residual_cap["A/leaf"]["A/dc"] == 10.0
+        # forward side initialized to 0 in reversed orientation
+        assert residual_cap["A/dc"]["A/leaf"] == 0.0
+
+    def test_dc_to_dc_reverse_edge_first_hop_proportional(self):
+        """Reverse-edge-first hop at destination should yield positive flow.
+
+        Topology (with reverse edges to simulate bidirectional links):
+          A_leaf -> A_dc  (10)
+          A_leaf -> B_leaf (10)
+          B_leaf -> B_dc  (10)
+          A_dc  -> A_leaf (10)  # reverse present
+          B_dc  -> B_leaf (10)  # reverse present
+
+        Pseudo nodes: source -> A_dc, B_dc -> sink
+        Expected max_flow(source, sink) = 10.0 in PROPORTIONAL mode.
+        """
+        g = StrictMultiDiGraph()
+        for n in ("A_dc", "A_leaf", "B_leaf", "B_dc", "source", "sink"):
+            g.add_node(n)
+
+        # Forward edges
+        g.add_edge("A_leaf", "A_dc", capacity=10.0, cost=1)
+        g.add_edge("A_leaf", "B_leaf", capacity=10.0, cost=1)
+        g.add_edge("B_leaf", "B_dc", capacity=10.0, cost=1)
+        # Reverse edges
+        g.add_edge("A_dc", "A_leaf", capacity=10.0, cost=1)
+        g.add_edge("B_dc", "B_leaf", capacity=10.0, cost=1)
+
+        # Pseudo source/sink
+        g.add_edge("source", "A_dc", capacity=float("inf"), cost=0)
+        g.add_edge("B_dc", "sink", capacity=float("inf"), cost=0)
+
+        r = init_flow_graph(g)
+        # Compute SPF with dst_node to mirror real usage in calc_max_flow
+        _costs, pred = spf(r, "source", dst_node="sink")
+        max_flow, _flow_dict = calc_graph_capacity(
+            r, "source", "sink", pred, flow_placement=FlowPlacement.PROPORTIONAL
+        )
+        assert max_flow == 10.0
+
+    def test_dc_to_dc_unidirectional_zero(self):
+        """Without reverse edges, DC cannot send to leaf; flow must be zero."""
+        g = StrictMultiDiGraph()
+        for n in ("A_dc", "A_leaf", "B_leaf", "B_dc", "source", "sink"):
+            g.add_node(n)
+
+        # Forward edges only
+        g.add_edge("A_leaf", "A_dc", capacity=10.0, cost=1)
+        g.add_edge("A_leaf", "B_leaf", capacity=10.0, cost=1)
+        g.add_edge("B_leaf", "B_dc", capacity=10.0, cost=1)
+
+        # Pseudo source/sink
+        g.add_edge("source", "A_dc", capacity=float("inf"), cost=0)
+        g.add_edge("B_dc", "sink", capacity=float("inf"), cost=0)
+
+        r = init_flow_graph(g)
+        _costs, pred = spf(r, "source", dst_node="sink")
+        max_flow, _flow_dict = calc_graph_capacity(
+            r, "source", "sink", pred, flow_placement=FlowPlacement.PROPORTIONAL
+        )
+        assert max_flow == 0.0
+
     def test_calc_graph_capacity_self_loop_empty_pred(self):
         """
         Test self-loop behavior when pred is empty.

tests/algorithms/test_max_flow.py

@@ -514,6 +514,63 @@ def test_max_flow_overlapping_source_sink_with_bidirectional(self):
         max_flow_b_b = calc_max_flow(g, "B", "B")
         assert max_flow_b_b == 0.0
 
+    def test_dc_leaf_bidirectional_parallel_branches(self):
+        """DC→leaf reversed hop with two parallel branches; compare placements.
+
+        Topology (all costs=1 unless noted):
+          S -> A_dc (inf)
+          A_dc -> A_leaf1 (cap=50)
+          A_dc -> A_leaf2 (cap=50)
+          A_leaf1 -> B_leaf1 (cap=100)
+          A_leaf2 -> B_leaf1 (cap=1)
+          B_leaf1 -> B_dc (cap=100)
+          B_dc -> T (inf)
+
+        Expected:
+          - PROPORTIONAL: branch1 min(50,100)=50, branch2 min(50,1)=1 → total 51
+          - EQUAL_BALANCED: nominal split 0.5 at A_dc; limiting ratio at edge (A_leaf2->B_leaf1): 1/0.5=2 → total 2
+        """
+        g = StrictMultiDiGraph()
+        for n in ("S", "A_dc", "A_leaf1", "A_leaf2", "B_leaf1", "B_dc", "T"):
+            g.add_node(n)
+
+        # Pseudo edges
+        g.add_edge("S", "A_dc", capacity=float("inf"), cost=0)
+        g.add_edge("B_dc", "T", capacity=float("inf"), cost=0)
+
+        # Reversed hop from DC to leaves (present as real edges here)
+        g.add_edge("A_dc", "A_leaf1", capacity=50.0, cost=1)
+        g.add_edge("A_dc", "A_leaf2", capacity=50.0, cost=1)
+
+        # Leaf to leaf aggregation
+        g.add_edge("A_leaf1", "B_leaf1", capacity=100.0, cost=1)
+        g.add_edge("A_leaf2", "B_leaf1", capacity=1.0, cost=1)
+
+        # Final hop into destination DC
+        g.add_edge("B_leaf1", "B_dc", capacity=100.0, cost=1)
+
+        # Full max flow: both placements reach the same min-cut (51)
+        flow_prop = calc_max_flow(
+            g, "S", "T", flow_placement=FlowPlacement.PROPORTIONAL
+        )
+        assert flow_prop == 51.0
+
+        flow_eq = calc_max_flow(
+            g, "S", "T", flow_placement=FlowPlacement.EQUAL_BALANCED
+        )
+        assert flow_eq == 51.0
+
+        # Single augmentation differs by placement
+        flow_prop_sp = calc_max_flow(
+            g, "S", "T", shortest_path=True, flow_placement=FlowPlacement.PROPORTIONAL
+        )
+        assert flow_prop_sp == 51.0
+
+        flow_eq_sp = calc_max_flow(
+            g, "S", "T", shortest_path=True, flow_placement=FlowPlacement.EQUAL_BALANCED
+        )
+        assert flow_eq_sp == 2.0
+
     def test_max_flow_self_loop_all_return_modes(self):
         """
         Test self-loop (s == t) behavior with all possible return value combinations.

tests/solver/test_maxflow_api.py

@@ -174,3 +174,28 @@ def test_sensitivity_analysis_keys_align_with_saturated_edges() -> None:
     sat = net.saturated_edges("^S$", "^T$", mode="combine")[("^S$", "^T$")]
     assert set((u, v, k) for (u, v, k) in sat) == set(delta_by_edge.keys())
     assert all(isinstance(delta, (int, float)) for delta in delta_by_edge.values())
+
+
+def test_network_dc_to_dc_reverse_edge_first_hop() -> None:
+    """Integration: DC->DC flow that requires a reverse edge on first hop.
+
+    Nodes: A/dc, A/leaf, B/leaf, B/dc. Links (forward):
+      A/leaf->A/dc (10), A/leaf->B/leaf (10), B/leaf->B/dc (10)
+
+    The wrapper builds a StrictMultiDiGraph with add_reverse=True, creating
+    reverse DC->leaf edges, so A/dc can reach B/dc via DC->leaf->leaf->DC.
+
+    Expect positive flow (10.0) in combine mode.
+    """
+    net = Network()
+    for name in ["A/dc", "A/leaf", "B/leaf", "B/dc"]:
+        net.add_node(Node(name))
+
+    # Forward links only; the graph builder will add reverse edges
+    net.add_link(Link("A/leaf", "A/dc", capacity=10.0, cost=1.0))
+    net.add_link(Link("A/leaf", "B/leaf", capacity=10.0, cost=1.0))
+    net.add_link(Link("B/leaf", "B/dc", capacity=10.0, cost=1.0))
+
+    res = net.max_flow(r"^A/dc$", r"^B/dc$", mode="combine")
+    assert (r"^A/dc$", r"^B/dc$") in res
+    assert res[(r"^A/dc$", r"^B/dc$")] == 10.0