fix(inspect): queue each node once in breadth-first ancestry walks
The latent and detailer image walks queued every link before checking whether its node was already seen, and popped from the list head. A dense shared fan-in grew the queue far past the visit budget before any visit was charged. Both walks now use a deque and enqueue each node at most once, charging every examined edge as scanned work; outcomes are unchanged.
This commit is contained in:
Vito Sansevero
@@ -1,5 +1,6 @@
|
||||
"""Conservative prompt-pair resolution contracts, independent of ComfyUI."""
|
||||
|
||||
import collections
|
||||
import random
|
||||
import subprocess # nosec B404 - fixed interpreter command, no shell
|
||||
import sys
|
||||
@@ -2634,6 +2635,91 @@ class TestLatentAncestry(unittest.TestCase):
|
||||
self.assertEqual(_inspect(source)["state"], "resolved")
|
||||
|
||||
|
||||
class _CountingDeque(collections.deque):
|
||||
"""Counts every item queued by the resolver's breadth-first walks."""
|
||||
|
||||
queued = 0
|
||||
|
||||
def append(self, item):
|
||||
_CountingDeque.queued += 1
|
||||
super().append(item)
|
||||
|
||||
def extend(self, items):
|
||||
for item in items:
|
||||
self.append(item)
|
||||
|
||||
|
||||
def _dense_layers(
|
||||
prefix: str, bottom: list, layers: int = 4, width: int = 40, fan: int = None
|
||||
):
|
||||
"""A hub over ``layers`` x ``width`` nodes; each links to ``fan`` (default:
|
||||
all) nodes of the next layer, so every node is shared by many consumers."""
|
||||
nodes = {}
|
||||
fan = fan or width
|
||||
for layer in range(layers):
|
||||
for index in range(width):
|
||||
below = (
|
||||
[
|
||||
[f"{prefix}{layer + 1}_{(index + step) % width}", 0]
|
||||
for step in range(fan)
|
||||
]
|
||||
if layer + 1 < layers
|
||||
else [bottom]
|
||||
)
|
||||
nodes[f"{prefix}{layer}_{index}"] = {
|
||||
"class_type": "DenseBlend",
|
||||
"inputs": {f"in_{k}": link for k, link in enumerate(below)},
|
||||
}
|
||||
# One hub reads the whole first layer.
|
||||
nodes[f"{prefix}top"] = {
|
||||
"class_type": "DenseBlend",
|
||||
"inputs": {f"in_{k}": [f"{prefix}0_{k}", 0] for k in range(width)},
|
||||
}
|
||||
return nodes, [f"{prefix}top", 0], layers * width + 1
|
||||
|
||||
|
||||
class TestBreadthFirstWalkWork(unittest.TestCase):
|
||||
"""Dense shared fan-in: each node is queued at most once per walk."""
|
||||
|
||||
def counted(self, source: dict, **options) -> tuple:
|
||||
_CountingDeque.queued = 0
|
||||
with mock.patch.object(resolver, "deque", _CountingDeque):
|
||||
result = _inspect(source, **options)
|
||||
return result, _CountingDeque.queued
|
||||
|
||||
def test_dense_latent_ancestry_queues_each_node_once(self):
|
||||
source = _latent_from({"class_type": "EmptyLatentImage", "inputs": {}})
|
||||
dense, top, count = _dense_layers("L", ["5", 0])
|
||||
source.update(dense)
|
||||
source["3"]["inputs"]["latent_image"] = top
|
||||
result, queued = self.counted(source)
|
||||
self.assertEqual(result["state"], "resolved")
|
||||
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
||||
# The dense layers plus the empty latent, each queued once.
|
||||
self.assertLessEqual(queued, count + 1)
|
||||
|
||||
def test_dense_detailer_image_ancestry_queues_each_node_once(self):
|
||||
source = _detailed(same_prompt=True)
|
||||
expected = _inspect(source, output_node_id="9")
|
||||
dense, top, count = _dense_layers("D", source["50"]["inputs"]["image"])
|
||||
source.update(dense)
|
||||
source["50"]["inputs"]["image"] = top
|
||||
result, queued = self.counted(source, output_node_id="9")
|
||||
self.assertEqual(_outcomes(result), _outcomes(expected))
|
||||
# Dense layers + the decoder + its sampler, plus the latent walk of
|
||||
# that sampler (one EmptyLatentImage).
|
||||
self.assertLessEqual(queued, count + 3)
|
||||
|
||||
def test_dense_fan_in_beyond_the_budget_is_work_limit(self):
|
||||
source = _latent_from({"class_type": "EmptyLatentImage", "inputs": {}})
|
||||
dense, top, count = _dense_layers("W", ["5", 0], layers=3, width=800, fan=40)
|
||||
source.update(dense)
|
||||
source["3"]["inputs"]["latent_image"] = top
|
||||
result, queued = self.counted(source)
|
||||
self.assertEqual(result["reason_code"], "work-limit")
|
||||
self.assertLessEqual(queued, count + 1)
|
||||
|
||||
|
||||
REPO = Path(__file__).resolve().parents[1]
|
||||
SEMANTICS_UNAVAILABLE = "Inspection node semantics are unavailable"
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ When the semantics table cannot be loaded, inspection of a prompt graph is
|
||||
|
||||
import logging
|
||||
import re
|
||||
from collections import deque
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Any, Dict, Mapping, Optional, Tuple
|
||||
|
||||
@@ -987,6 +988,27 @@ class _GraphResolver:
|
||||
reason_code="multistage-differs",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def walk_from(value: list) -> Tuple[deque, set]:
|
||||
"""A breadth-first queue holding one start link, and its queued IDs."""
|
||||
start = str(value[0])
|
||||
return deque([(start, 0)]), {start}
|
||||
|
||||
def enqueue(self, queue: deque, queued: set, inputs: dict, depth: int) -> None:
|
||||
"""Queue each linked input once per walk; every edge is charged as scanned.
|
||||
|
||||
Deduplicating at enqueue keeps a dense shared fan-in to one entry per
|
||||
node, so the queue never outgrows the nodes the visit budget pays for.
|
||||
Breadth-first, a node's first enqueue is at its least depth, as before.
|
||||
"""
|
||||
self.scan(len(inputs))
|
||||
for item in inputs.values():
|
||||
if _link(item):
|
||||
key = str(item[0])
|
||||
if key not in queued:
|
||||
queued.add(key)
|
||||
queue.append((key, depth))
|
||||
|
||||
def upstream_stages(self, origin: str, value: Any) -> list:
|
||||
"""Samplers reached from a detailer's image input, in breadth-first order.
|
||||
|
||||
@@ -996,26 +1018,19 @@ class _GraphResolver:
|
||||
if not _link(value):
|
||||
raise _Unresolved("Expected an image link", "malformed")
|
||||
found: list = []
|
||||
seen: set = set()
|
||||
queue = [(str(value[0]), 0)]
|
||||
queue, queued = self.walk_from(value)
|
||||
while queue:
|
||||
node_id, depth = queue.pop(0)
|
||||
if node_id in seen:
|
||||
continue
|
||||
node_id, depth = queue.popleft()
|
||||
if node_id == origin:
|
||||
raise _Unresolved("Cyclic sampler stage ancestry", "malformed")
|
||||
if depth >= self.max_depth:
|
||||
raise _Unresolved("Resolution depth limit reached", "depth-limit")
|
||||
seen.add(node_id)
|
||||
self.budget.spend()
|
||||
name, inputs = self.node(node_id)
|
||||
if name in self.stage_classes:
|
||||
found.append(node_id)
|
||||
continue
|
||||
self.scan(len(inputs))
|
||||
queue.extend(
|
||||
(str(item[0]), depth + 1) for item in inputs.values() if _link(item)
|
||||
)
|
||||
self.enqueue(queue, queued, inputs, depth + 1)
|
||||
return found
|
||||
|
||||
def final_stages(self, sampler_ids: list) -> list:
|
||||
@@ -1065,9 +1080,9 @@ class _GraphResolver:
|
||||
|
||||
def reaches_output(self, node_id: str, consumers: dict) -> bool:
|
||||
"""Whether a sampler's output reaches a terminal node besides later stages."""
|
||||
seen, queue = {node_id}, [node_id]
|
||||
seen, queue = {node_id}, deque([node_id])
|
||||
while queue:
|
||||
current = queue.pop(0)
|
||||
current = queue.popleft()
|
||||
for consumer in consumers.get(current, ()):
|
||||
if consumer in seen:
|
||||
continue
|
||||
@@ -1210,17 +1225,13 @@ class _GraphResolver:
|
||||
"""
|
||||
if not _link(value):
|
||||
raise _Unresolved("Unknown latent ancestry", "malformed")
|
||||
seen: set = set()
|
||||
queue = [(str(value[0]), 0)]
|
||||
queue, queued = self.walk_from(value)
|
||||
while queue:
|
||||
node_id, depth = queue.pop(0)
|
||||
if node_id in seen:
|
||||
continue
|
||||
node_id, depth = queue.popleft()
|
||||
if node_id == origin:
|
||||
raise _Unresolved("Cyclic latent ancestry", "malformed")
|
||||
if depth >= self.max_depth:
|
||||
raise _Unresolved("Resolution depth limit reached", "depth-limit")
|
||||
seen.add(node_id)
|
||||
self.budget.spend()
|
||||
name, inputs = self.node(node_id)
|
||||
if name in self.stage_classes:
|
||||
@@ -1231,10 +1242,7 @@ class _GraphResolver:
|
||||
_link(inputs.get(key)) for key in _SAMPLER_LIKE_INPUTS
|
||||
):
|
||||
raise self.unsupported("Unsupported latent ancestry", node_id)
|
||||
self.scan(len(inputs))
|
||||
queue.extend(
|
||||
(str(item[0]), depth + 1) for item in inputs.values() if _link(item)
|
||||
)
|
||||
self.enqueue(queue, queued, inputs, depth + 1)
|
||||
|
||||
def anchor(self, node_id: str) -> str:
|
||||
seen = set()
|
||||
|
||||
Reference in New Issue
Block a user