diff --git a/core/graph.py b/core/graph.py index 190d0bda..7f13c2ec 100644 --- a/core/graph.py +++ b/core/graph.py @@ -148,6 +148,10 @@ def build(slots: Sequence[dict], schemas: Optional[Schemas] = None) -> Tuple[dic return {}, problems known = seen + # Built once. Rebuilding it per link made validating a chain of N slots do N + # dict rebuilds, and this runs on the event loop: the cost of checking a + # pipeline is paid by every other request the server is holding. + by_id = slots_by_id(slots) for slot in slots: slot_id, class_type = slot["id"], slot["node"] @@ -166,7 +170,7 @@ def build(slots: Sequence[dict], schemas: Optional[Schemas] = None) -> Tuple[dic if source not in known: problems.append(f"{slot_id}.{name} is fed by {source!r}, which is not in the pipeline") continue - produced = schemas.outputs(slots_by_id(slots)[source]["node"]) + produced = schemas.outputs(by_id[source]["node"]) if index >= len(produced): problems.append(f"{slot_id}.{name} reads output {index} of {source!r}, " f"which has {len(produced)}") diff --git a/core/tests/test_graph.py b/core/tests/test_graph.py index d1de1872..e62bcd57 100644 --- a/core/tests/test_graph.py +++ b/core/tests/test_graph.py @@ -212,6 +212,24 @@ def test_a_long_chain_is_walked_without_running_out_of_stack(): assert len(problems) == 1 +def test_checking_a_big_pipeline_does_not_cost_more_than_reading_it(): + # Slots arrive from a request and nothing caps how many, and build() runs on + # the event loop -- so the cost of checking one pipeline is paid by every + # other request the server is holding. Looking the source of each link up in + # a dict rebuilt per link made this quadratic: 20000 slots took 21 seconds. + # The budget is deliberately loose; it is there to catch a return to N^2, + # not to measure this machine. + import time + slots = [{"id": "s0", "node": "LoadModel", "inputs": {"name": "x"}}] + slots += [{"id": f"s{i}", "node": "AddLora", + "inputs": {"model": [f"s{i - 1}", 0], "strength": 1.0}} + for i in range(1, 20000)] + started = time.perf_counter() + _, problems = graph.build(slots, SCHEMAS) + assert problems == [] + assert time.perf_counter() - started < 2.0 + + def test_a_diamond_is_not_a_cycle(): # Two paths reaching one source is ordinary wiring; a walk that marks a slot # seen without unmarking it would call this a loop.