beam: make topo_sort cycle detection iterative (no RecursionError)
This commit is contained in:
1 parent
2d785702d1
commit
51d8d18189
1 file changed
+31
-13
@@ -83,19 +83,37 @@ def topo_sort(graph: Mapping[Any, Iterable[Any]]) -> TopoResult:
|
||||
color: dict[Any, int] = {n: 0 for n in nodes} # 0 unvisited, 1 visiting, 2 done
|
||||
parent: dict[Any, Any] = {}
|
||||
|
||||
def find_back_edge(u: Any) -> tuple[Any, Any] | None:
|
||||
"""DFS from u; return the (node, ancestor) back edge that closes a cycle."""
|
||||
color[u] = 1
|
||||
for v in sorted(deps_of[u], key=_key):
|
||||
state = color.get(v, 0)
|
||||
if state == 0:
|
||||
parent[v] = u
|
||||
found = find_back_edge(v)
|
||||
if found is not None:
|
||||
return found
|
||||
elif state == 1:
|
||||
return (u, v)
|
||||
color[u] = 2
|
||||
# Iterative DFS (avoids RecursionError on deep chains). Returns the
|
||||
# (node, ancestor) back edge closing a cycle, or None.
|
||||
def find_back_edge(start: Any) -> tuple[Any, Any] | None:
|
||||
stack: list[tuple[Any, Any]] = [(start, None)] # (node, parent)
|
||||
# iterator state per node
|
||||
iters: dict[Any, Any] = {}
|
||||
color[start] = 1
|
||||
parent[start] = None
|
||||
# We manage traversal manually.
|
||||
cur = start
|
||||
# Use an explicit work stack of (node, iterator over deps)
|
||||
work: list[tuple[Any, Any]] = [(start, iter(sorted(deps_of[start], key=_key)))]
|
||||
color[start] = 1
|
||||
while work:
|
||||
u, it = work[-1]
|
||||
advanced = False
|
||||
for v in it:
|
||||
state = color.get(v, 0)
|
||||
if state == 0:
|
||||
color[v] = 1
|
||||
parent[v] = u
|
||||
work.append((v, iter(sorted(deps_of[v], key=_key))))
|
||||
advanced = True
|
||||
break
|
||||
elif state == 1:
|
||||
# back edge u -> v
|
||||
return (u, v)
|
||||
if not advanced:
|
||||
# exhausted
|
||||
color[u] = 2
|
||||
work.pop()
|
||||
return None
|
||||
|
||||
back: tuple[Any, Any] | None = None
|
||||
|
||||
Reference in new issue
Block a user