"""Shared pure topological-sort helpers for render scheduling.
Two schedulers order work by dependency: the intra-scene ``RenderGraph``
(named passes over string resources, graphics package) and the inter-scene
:class:`~simvx.core.scene_target_graph.SceneTargetGraph` (scene render jobs
over sampled render targets). Both call the single Kahn implementation here,
per design/render_parallelism_and_multi_gpu.md P1-1 (extract the shared
topo-sort helper, no fork).
Two entry points with deliberately different cycle policies:
* :func:`topo_sort` raises :class:`CycleError`. An intra-scene pass cycle is
always a bug (``RenderGraph.compile``).
* :func:`topo_sort_lagged` never raises. An inter-scene cycle (two mirrors
facing each other) is legitimate: Tarjan SCC finds each cyclic component
and the minimal deterministic back-edge(s) are dropped to restore a DAG.
Dropped edges are returned so the caller can treat them as async (lagged)
edges sampling the previous frame's target, scoped to the cyclic boundary.
Both sorts are deterministic: ties break by position in the input sequence
(pass a name-sorted sequence for lexicographic tie-breaking).
"""
from __future__ import annotations
from collections.abc import Hashable, Iterable, Mapping, Sequence
__all__ = ["CycleError", "topo_sort", "topo_sort_lagged"]
[docs]
class CycleError(ValueError):
"""A dependency cycle was found where none is allowed.
``remaining`` holds the nodes that could not be scheduled (every node
participating in, or downstream of, a cycle).
"""
def __init__(self, remaining: Iterable) -> None:
self.remaining: tuple = tuple(remaining)
super().__init__(f"Cycle detected involving: {sorted(map(str, self.remaining))}")
[docs]
def topo_sort[K: Hashable](nodes: Sequence[K], deps: Mapping[K, Iterable[K]]) -> list[K]:
"""Kahn topological sort of *nodes*; raises :class:`CycleError` on a cycle.
Args:
nodes: Every node, in tie-break order (position = priority).
deps: ``node -> nodes that must come first``. Entries not present in
*nodes* are ignored (dangling dependencies are allowed, matching
``RenderGraph``'s external-resource inputs). Self-dependencies are
ignored.
Returns:
The nodes in dependency order (producers first), ties broken by
position in *nodes*.
"""
index_of = {node: i for i, node in enumerate(nodes)}
edges = _index_edges(nodes, deps, index_of)
ordered_idx, remaining = _kahn(len(nodes), edges)
if remaining:
raise CycleError(nodes[i] for i in remaining)
return [nodes[i] for i in ordered_idx]
[docs]
def topo_sort_lagged[K: Hashable](
nodes: Sequence[K], deps: Mapping[K, Iterable[K]]
) -> tuple[list[K], set[tuple[K, K]]]:
"""Kahn topological sort that degrades cycles to lagged edges, never raises.
Same contract as :func:`topo_sort`, but on a cycle it runs Tarjan SCC,
drops the minimal deterministic back-edge within each non-trivial SCC and
retries until acyclic. Every node is always scheduled exactly once.
Returns:
``(ordered, dropped_edges)`` where ``dropped_edges`` is the set of
``(producer, consumer)`` pairs removed to break cycles: the consumer
must read the producer's previous output across that edge (a 1-frame
lag under serial per-target execution). Empty on the acyclic case.
"""
index_of = {node: i for i, node in enumerate(nodes)}
n = len(nodes)
work = _index_edges(nodes, deps, index_of)
broken: set[tuple[int, int]] = set()
while True:
ordered_idx, remaining = _kahn(n, work)
if not remaining:
return [nodes[i] for i in ordered_idx], {(nodes[p], nodes[c]) for p, c in broken}
back_edge = _pick_back_edge(n, work)
if back_edge is None: # pragma: no cover - remaining implies a cycle
ordered_idx.extend(sorted(remaining))
return [nodes[i] for i in ordered_idx], {(nodes[p], nodes[c]) for p, c in broken}
work.discard(back_edge)
broken.add(back_edge)
# --------------------------------------------------------------------------- #
# Index-based engine (nodes mapped to 0..n-1; edges are (producer, consumer))
# --------------------------------------------------------------------------- #
def _index_edges(nodes: Sequence, deps: Mapping, index_of: Mapping) -> set[tuple[int, int]]:
"""Translate ``deps`` into ``(producer_index, consumer_index)`` edges."""
edges: set[tuple[int, int]] = set()
for consumer in nodes:
ci = index_of[consumer]
for producer in deps.get(consumer, ()):
pi = index_of.get(producer)
if pi is not None and pi != ci:
edges.add((pi, ci))
return edges
def _kahn(n: int, edges: set[tuple[int, int]]) -> tuple[list[int], set[int]]:
"""Deterministic Kahn sort over indices; returns ``(ordered, unscheduled)``.
``unscheduled`` is non-empty exactly when the edge set is cyclic; ties
break by index (lowest first).
"""
adj: dict[int, list[int]] = {i: [] for i in range(n)}
indeg = [0] * n
for p, c in edges:
adj[p].append(c)
indeg[c] += 1
queue = sorted(i for i in range(n) if indeg[i] == 0)
ordered: list[int] = []
while queue:
node = queue.pop(0)
ordered.append(node)
grew = False
for nxt in sorted(adj[node]):
indeg[nxt] -= 1
if indeg[nxt] == 0:
queue.append(nxt)
grew = True
if grew:
queue.sort()
return ordered, set(range(n)) - set(ordered)
def _pick_back_edge(n: int, edges: set[tuple[int, int]]) -> tuple[int, int] | None:
"""Return one deterministic back-edge inside a non-trivial SCC, or None.
Uses Tarjan to find SCCs, picks the lowest-index non-trivial SCC, and
within it the lexicographically smallest edge that points "backwards"
(toward an equal-or-lower index member), which is guaranteed to exist in a
cycle. Removing it strictly reduces the cycle without orphaning producers
outside the SCC.
"""
sccs = _tarjan_scc(n, edges)
# Deterministic: consider SCCs ordered by their smallest member.
for scc in sorted((sorted(s) for s in sccs if len(s) > 1), key=lambda s: s[0]):
members = set(scc)
in_scc = sorted((p, c) for (p, c) in edges if p in members and c in members)
# Prefer an edge that goes to an equal-or-lower index (a true back-edge
# in the deterministic index order); fall back to the smallest in-SCC
# edge so we always make progress.
for p, c in in_scc:
if c <= p:
return (p, c)
if in_scc:
return in_scc[0]
return None
def _tarjan_scc(n: int, edges: set[tuple[int, int]]) -> list[set[int]]:
"""Tarjan's strongly-connected-components, iterative (no recursion limit)."""
adj: dict[int, list[int]] = {i: [] for i in range(n)}
for p, c in edges:
adj[p].append(c)
for i in range(n):
adj[i].sort()
index_counter = 0
indices: dict[int, int] = {}
lowlink: dict[int, int] = {}
on_stack: set[int] = set()
stack: list[int] = []
result: list[set[int]] = []
for start in range(n):
if start in indices:
continue
# Iterative DFS with an explicit work stack of (node, child_iterator).
work: list[tuple[int, int]] = [(start, 0)]
while work:
node, child_i = work[-1]
if child_i == 0:
indices[node] = lowlink[node] = index_counter
index_counter += 1
stack.append(node)
on_stack.add(node)
if child_i < len(adj[node]):
work[-1] = (node, child_i + 1)
nxt = adj[node][child_i]
if nxt not in indices:
work.append((nxt, 0))
elif nxt in on_stack:
lowlink[node] = min(lowlink[node], indices[nxt])
else:
if lowlink[node] == indices[node]:
comp: set[int] = set()
while True:
w = stack.pop()
on_stack.discard(w)
comp.add(w)
if w == node:
break
result.append(comp)
work.pop()
if work:
parent = work[-1][0]
lowlink[parent] = min(lowlink[parent], lowlink[node])
return result