"""SceneTargetGraph: inter-scene render-target dependency scheduler.
design/render_parallelism_and_multi_gpu.md P1-2. Where ``RenderGraph``
(graphics package) orders named passes *within* one scene render, this graph
orders whole scene render jobs, one per offscreen target (each SubViewport,
and later each RenderView) plus the main scene, so a producer target renders
before any job that samples it in the same frame.
Jobs register the way ``RenderGraph`` passes do: :meth:`SceneTargetGraph.add`
then :meth:`SceneTargetGraph.compile`. Edges are texture-sampling
dependencies, expressed two ways:
* implicit: a job ``consumes`` bindless slot integers; the job that
``produces`` that slot must come first (mirrors ``RenderGraph``'s named
string resources, with bindless slots as the resource names);
* explicit: ``after`` lists producer job *names* (the ``feeds_from`` hint and
the main scene's "after every offscreen target" barrier).
Both graphs share the one Kahn implementation in
:mod:`simvx.core._topo_sort` (P1-1, no fork). Cycle policy differs by design:
an intra-scene pass cycle is a bug (``RenderGraph`` raises), but an
inter-scene cycle (mirror facing mirror) is legitimate, so ``compile`` never
raises: Tarjan SCC drops the minimal back-edge(s) and reports them as
:attr:`lagged_edges`, each sampling the previous frame's target, scoped to
the cyclic boundary. Under serial per-target execution a lagged edge is
automatically correct (the previous frame's content is fully resident).
Compilation is fingerprint-cached: re-registering an identical job set (the
steady-state frame) reuses the previously computed order without re-sorting.
The sort therefore runs on a structure change (a target added/removed, a slot
assigned, a sampling binding changed), not per frame. Consumed-slot
*detection* stays with the caller (``order_subviewports`` scans subtrees),
because a sampling binding such as ``material.albedo_tex_index`` can change
without bumping the tree's structure version.
Dependency-light on purpose (imports nothing from ``simvx.graphics``): the
web backend orders its offscreen targets through the same code path.
"""
from __future__ import annotations
import logging
from collections.abc import Iterable
from typing import Any
from ._topo_sort import topo_sort_lagged
log = logging.getLogger(__name__)
__all__ = ["SceneJob", "SceneTargetGraph"]
[docs]
class SceneJob:
"""One scene render job: an offscreen target (or the main scene).
Args:
name: Unique job name (deterministic across frames for a stable
structure, so the compile cache holds).
node: The owning scene node (a SubViewport), or ``None`` for the main
scene job. Carried through so :attr:`SceneTargetGraph.order` maps
straight back to nodes.
produces: The bindless slot this job renders into, or ``-1`` when it
publishes no sampleable target (the main scene, a first-frame
target with no slot yet).
consumes: Bindless slots sampled inside this job's scene. Slots no
registered job produces are ignored (dangling inputs are allowed,
as in ``RenderGraph``: the resource comes from outside the graph).
after: Names of jobs that must render first, regardless of slots
(explicit ``feeds_from`` hints; the main scene's barrier).
Unknown names are ignored.
"""
__slots__ = ("name", "node", "produces", "consumes", "after")
def __init__(
self,
name: str,
*,
node: Any = None,
produces: int = -1,
consumes: Iterable[int] = (),
after: Iterable[str] = (),
) -> None:
self.name = name
self.node = node
self.produces = int(produces)
self.consumes = frozenset(int(s) for s in consumes)
self.after = tuple(after)
[docs]
def __repr__(self) -> str: # pragma: no cover - debug aid
return f"SceneJob({self.name!r}, produces={self.produces}, consumes={sorted(self.consumes)})"
[docs]
class SceneTargetGraph:
"""Orders scene render jobs so producer targets render before consumers.
Usage (mirrors ``RenderGraph``)::
graph = SceneTargetGraph()
graph.clear() # start a (re-)registration
graph.add(SceneJob("svp:12", node=svp, produces=12, consumes={13}))
graph.add(SceneJob("svp:13", node=other, produces=13))
graph.add(SceneJob("scene:main", after=("svp:12", "svp:13")))
graph.compile() # sorts only if the structure changed
for job in graph.order: ...
graph.lagged_edges # broken cycle edges, usually empty
"""
__slots__ = ("_jobs", "_fingerprint", "_order_idx", "_lagged_idx", "_compile_count")
def __init__(self) -> None:
self._jobs: list[SceneJob] = []
self._fingerprint: tuple | None = None
self._order_idx: tuple[int, ...] = ()
self._lagged_idx: frozenset[tuple[int, int]] = frozenset()
self._compile_count: int = 0
[docs]
def clear(self) -> None:
"""Drop registered jobs (the compiled order survives until the next compile)."""
self._jobs.clear()
[docs]
def add(self, job: SceneJob) -> None:
"""Register a job. Call before :meth:`compile`. Duplicate names are an error."""
if any(j.name == job.name for j in self._jobs):
raise ValueError(f"Duplicate scene job name: {job.name!r}")
self._jobs.append(job)
[docs]
def compile(self) -> bool:
"""Topologically sort the registered jobs; producers first, never raises.
Fingerprint-cached: when the registered jobs describe the same
structure as the previous compile (same names, slots, sampled slots
and explicit edges, in the same registration order), the cached order
is remapped onto the current job objects without re-sorting. Returns
``True`` when a real sort ran, ``False`` on a cache hit.
"""
fingerprint = tuple((j.name, j.produces, tuple(sorted(j.consumes)), j.after) for j in self._jobs)
if fingerprint == self._fingerprint:
return False
n = len(self._jobs)
index_by_name = {j.name: i for i, j in enumerate(self._jobs)}
# slot -> producer job index; first registration wins (bindless slots
# are unique, so a collision is defensive only).
producer_by_slot: dict[int, int] = {}
for i, job in enumerate(self._jobs):
if job.produces >= 0:
producer_by_slot.setdefault(job.produces, i)
deps: dict[int, set[int]] = {}
for i, job in enumerate(self._jobs):
d: set[int] = set()
for slot in job.consumes:
pi = producer_by_slot.get(slot)
if pi is not None and pi != i:
d.add(pi)
for name in job.after:
pi = index_by_name.get(name)
if pi is not None and pi != i:
d.add(pi)
if d:
deps[i] = d
ordered_idx, dropped = topo_sort_lagged(tuple(range(n)), deps)
self._order_idx = tuple(ordered_idx)
self._lagged_idx = frozenset(dropped)
self._fingerprint = fingerprint
self._compile_count += 1
log.debug(
"SceneTargetGraph compiled: order=[%s], lagged=%d",
", ".join(self._jobs[i].name for i in self._order_idx),
len(self._lagged_idx),
)
return True
[docs]
@property
def order(self) -> tuple[SceneJob, ...]:
"""Compiled render order (producers first)."""
return tuple(self._jobs[i] for i in self._order_idx)
[docs]
@property
def lagged_edges(self) -> frozenset[tuple[SceneJob, SceneJob]]:
"""``(producer, consumer)`` edges broken to resolve cycles (1-frame lag)."""
return frozenset((self._jobs[p], self._jobs[c]) for p, c in self._lagged_idx)
[docs]
@property
def compile_count(self) -> int:
"""How many real sorts have run (cache hits excluded)."""
return self._compile_count
[docs]
@property
def jobs(self) -> tuple[SceneJob, ...]:
"""Registered jobs, in registration order."""
return tuple(self._jobs)