sidecar: a hold group is one per connection, and the compiler to match

cli/ffrwd/lower.py

@@ -16629,7 +16629,15 @@ def _node_instance(
                 if number not in numbers_given:
                     numbers_given.append(number)
                 feeds.append((number, one, conformed))
-            ports[hold.port_param] = numbers_given if listed else numbers_given[0]
+            if listed:
+                # Every many input of a group lists the group's connections
+                # in one param, in the order they were first written.
+                given = ports.get(hold.port_param)
+                listing = list(given) if isinstance(given, list) else []
+                listing += [number for number in numbers_given if number not in listing]
+                ports[hold.port_param] = listing
+            else:
+                ports[hold.port_param] = numbers_given[0]
         if ports:
             # The port is one of the call's params, and the shape is asked
             # for the params the call ends up with.
@@ -17236,7 +17244,10 @@ def _check_feeder_groups(self) -> None:
         group split across sources is one whole set of calls per source. A
         source that feeds only some of them, such as the picture from one
         lateral and the sound from another, would leave each call without
-        its partner, and is refused.
+        its partner, and is refused. A group whose every connection reaches
+        one and the same call is that call's many inputs, a stream of each
+        per connection, and a connection may bring some of them and not
+        others.
         """
         by_group: dict[str, list[tuple[frozenset[str], int, exp.Expr]]] = {}
         for (group, sources), (port, argument) in self._feeder_groups.items():
@@ -17250,10 +17261,16 @@ def reached(port: int) -> list[str]:
                     {f"{call.function}({call.param})" for call in self._feeds[port].calls}
                 )
 
+            def nodes(port: int) -> set[str]:
+                return {call.node for call in self._feeds[port].calls}
+
             first_sources, first_port, _ = connections[0]
+            one_call = len(nodes(first_port)) == 1
             for sources, port, argument in connections[1:]:
                 if reached(port) == reached(first_port):
                     continue
+                if one_call and nodes(port) == nodes(first_port):
+                    continue
                 raise _error(
                     ErrorCode.UNSUPPORTED_SQL,
                     f"the group '{group}' reads {_listed_sources(first_sources)} in "

cli/ffrwd/registry.py

@@ -154,12 +154,12 @@
 import re
 import subprocess
 import tempfile
+import threading
 from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Literal
 
 from ffrwd import binaries
-from ffrwd.errors import FfrwdError
 from ffrwd.ir import StreamType
 
 _TIMEOUT_SECONDS = 10.0
@@ -743,6 +743,9 @@ def __init__(self) -> None:
         self._ffmpeg: str | None = None
         self._version_line: str | None = None
         self._loaded = False
+        # Run-time lateral instances compile on threads of their own, and
+        # the first to read the registry loads it while the others wait.
+        self._loading = threading.Lock()
         self._filters: dict[str, DynamicFilter] = {}
         self._sources: dict[str, SourceFilter] = {}
         self._options: dict[str, dict[str, FilterOption]] = {}
@@ -753,15 +756,16 @@ def __init__(self) -> None:
     def _ensure_loaded(self) -> None:
         if self._loaded:
             return
-        self._loaded = True
-
-        try:
-            ffmpeg = binaries.ffmpeg_path()
-        except FfrwdError:
-            # FFRWD_OFFLINE refused the provider's download: asked again,
-            # it says so again.
-            self._loaded = False
-            raise
+        with self._loading:
+            if self._loaded:
+                return
+            # FFRWD_OFFLINE refusing the provider's download raises, and
+            # leaves it unloaded: asked again, it says so again.
+            self._load()
+            self._loaded = True
+
+    def _load(self) -> None:
+        ffmpeg = binaries.ffmpeg_path()
         if ffmpeg is None:
             return
         version_line = _get_version_line(ffmpeg)
@@ -893,14 +897,16 @@ def to_snapshot_payload(self) -> dict[str, object] | None:
 
 
 _registry: Registry | None = None
+_making = threading.Lock()
 
 
 def load() -> Registry:
     """Return the process-wide Registry singleton (memoized)."""
     global _registry
-    if _registry is None:
-        _registry = Registry()
-    return _registry
+    with _making:
+        if _registry is None:
+            _registry = Registry()
+        return _registry
 
 
 def load_reference(path: str | Path) -> Registry:

cli/tests/exec/test_exec_laterals.py

@@ -48,6 +48,7 @@
 _PROBE = _RELEASE / "feed_probe.wasm"
 _HEAR = _RELEASE / "feed_probe_audio.wasm"
 _STAMP = _RELEASE / "data_stamp.wasm"
+_GROUP = _RELEASE / "shape_group.wasm"
 _LAUNCH = _CLI_ROOT / "tests" / "data" / "launch.nut"
 _TIMEOUT = 120.0
 
@@ -75,6 +76,24 @@
 """
 
 
+# play with both columns; an instance whose sound no feeder reads binds its
+# sound's values by DEFAULT.
+_BOTH_PLAY = """CREATE FUNCTION play(launch data_stream, url text, start_pts number,
+                     duration number, width number, height number, fps number,
+                     pix_fmt text, rate number DEFAULT 48000,
+                     channels number DEFAULT 2, sample_fmt text DEFAULT 'f32')
+RETURNS TABLE(video video_stream, audio audio_stream) AS $$
+  SELECT setpts(ffmpeg.format(fps(scale(m.video[1], width, height), fps),
+                              pix_fmts => pix_fmt),
+                'PTS+' || start_pts::text || '/TB') AS video,
+         asetpts(ffmpeg.aformat(aresample(m.audio[1], rate), sample_fmts => sample_fmt,
+                                channel_layouts => channels::text || 'c'),
+                 'PTS+' || start_pts::text || '/TB') AS audio
+  FROM input(url) m
+$$ LANGUAGE sql;
+"""
+
+
 @pytest.fixture(autouse=True)
 def _require_everything() -> None:
     if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None:
@@ -242,3 +261,57 @@ def test_a_data_filters_output_reaches_the_lateral_and_the_file_both(
     written = _data_messages(out)
     assert [at for at, _ in written] == [0.1, 0.2, 0.3, 0.5]
     assert {message["node"] for _, message in written} == {"leaf"}
+
+
+def test_two_laterals_on_one_many_hold_group_are_a_connection_and_an_offset_each(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The fleet's ``shape-group`` takes many pictures and many sounds on one
+    hold group. `m` brings its picture and its sound, `c` its picture alone,
+    stamped a second later: each is a connection of its own, live beside the
+    other on its own origin, and `m`'s picture and sound share each
+    instance's offset."""
+    if not _GROUP.exists():
+        pytest.skip(f"module missing: {_GROUP}")
+    programme = _media(tmp_path)
+    monkeypatch.chdir(tmp_path)
+    fed = tmp_path / "fed.ndjson"
+    late = _BOTH_PLAY.replace("play(", "late(", 1).replace(
+        "'PTS+' || start_pts::text", "'PTS+' || (start_pts + 1)::text"
+    )
+    sql = (
+        "CREATE FUNCTION grp(c video_stream, a audio_stream, v video_stream[] DEFAULT NULL, "
+        "s audio_stream[] DEFAULT NULL, ports number DEFAULT NULL)\nRETURNS STRUCT("
+        "c video_stream, feeds STRUCT(pts number, member text, k number, live boolean, "
+        "at_pts number, first_pts number)[])\n"
+        f"AS '{_GROUP.as_posix()}', 'shape_group' LANGUAGE wasm;\n" + _BOTH_PLAY + late
+        + "COPY (SELECT grp(p.video[1], p.audio[1], ARRAY[m.video, c.video], "
+        "ARRAY[m.audio]).feeds FROM "
+        f"input('{programme.as_posix()}', realtime => true) p, "
+        f"input('{_LAUNCH.as_posix()}') f, LATERAL play(f.data[1]) m, "
+        f"LATERAL late(f.data[1]) c) TO '{fed.as_posix()}'"
+    )
+    rows = _run(sql, capsys)
+    feeders = [row for row in rows if row.get("event") == "feeder"]
+    assert len([row for row in feeders if row.get("exit") == 0]) == 4, feeders
+    said = [json.loads(line) for line in fed.read_text().splitlines() if line]
+
+    def starts(member: str, k: int, field: str = "at_pts") -> list[int]:
+        return [
+            int(row[field]) for row in said
+            if (row["member"], row["k"], row["live"]) == (member, k, True)
+        ]
+
+    # Two instances of each lateral, each anchored as it connects.
+    assert len(starts("v", 0)) == 2, said
+    assert len(starts("v", 1)) == 2, said
+    assert starts("s", 0) == starts("v", 0), "m's sound shares its picture's offset"
+    assert starts("s", 1) == [], "c brings no sound"
+    firsts = list(zip(starts("v", 0, "first_pts"), starts("v", 1, "first_pts"), strict=True))
+    assert all(mine < its for mine, its in firsts), f"c's feeds start on c's pts: {firsts}"
+    live: dict[tuple[str, int], bool] = {}
+    both = False
+    for row in said:
+        live[(str(row["member"]), int(row["k"]))] = bool(row["live"])
+        both = both or (live.get(("v", 0), False) and live.get(("v", 1), False))
+    assert both, f"each connection is live while the other is: {said}"

cli/tests/test_laterals.py

@@ -1025,6 +1025,86 @@ def test_a_port_written_by_hand_stays_one_integer_where_a_list_also_fits() -> No
     assert port == 9100 and isinstance(port, int)
 
 
+# compose taking each lateral's sound beside its picture, on the same group.
+_COMPOSE_AUDIO = (
+    "CREATE FUNCTION compose(v video_stream, inputs video_stream[] DEFAULT NULL, "
+    "audio audio_stream[] DEFAULT NULL, port number DEFAULT NULL) RETURNS video_stream "
+    f"AS '{COMPOSE}', 'compose' LANGUAGE wasm;"
+)
+
+
+def _compose_audio_shape(
+    module: str, params: str, bound: Sequence[shapes.Binding], grants: Sequence[str] = ()
+) -> shapes.NodeShape:
+    """compose with `audio`, many sounds held on the pictures' port."""
+    held = {
+        "kind": "hold",
+        "anchor": {"kind": "tagged", "tag": "smart_timed"},
+        "lead": 0.3,
+        "port_param": "port",
+    }
+
+    def port(
+        name: str, kind: str, pairing: dict[str, object], many: bool
+    ) -> dict[str, object]:
+        return {
+            "name": name, "kind": kind, "required": name == "v", "many": many,
+            "pairing": pairing, "rows": "ignore", "window": 1, "stride": 1, "accepts": {},
+        }
+
+    return shapes.node_shape(module, {
+        "inputs": [
+            port("v", "video", {"kind": "lockstep"}, False),
+            port("inputs", "video", held, True),
+            port("audio", "audio", held, True),
+        ],
+        "outputs": [{"name": "v", "kind": "video", "latency": 0,
+                     "format": {"kind": "like", "port": "v"}}],
+        "clock": {"kind": "input", "port": "v"},
+        "pure": True, "one_to_one": True, "bounded": True, "relation": [],
+    })
+
+
+def test_a_lateral_bringing_a_picture_alone_to_a_call_taking_sound_too_is_its_connection() -> None:
+    """ad brings compose its picture and its sound and lbar its picture alone:
+    both connections reach the one call, so the group is not split, and the
+    port list names each connection once, in the order first written."""
+    query = _TWO_ADS.replace(
+        "ARRAY[ad.video, lbar.video])", "ARRAY[ad.video, lbar.video], audio => ARRAY[ad.audio])"
+    )
+    plan = compile_all(
+        _declared(query).replace("COPY (", _COMPOSE_AUDIO + "\nCOPY (", 1),
+        describe=lambda path: _compose_modules({"type": "array", "items": {"type": "integer"}})[
+            path
+        ],
+        shape=_compose_audio_shape,
+    ).plan
+    assert plan is not None
+    assert [len(lateral.connections) for lateral in plan.laterals] == [1, 1]
+    ports = [lateral.connections[0].port for lateral in plan.laterals]
+    assert len(set(ports)) == 2
+    assert _compose_node(plan).args["port"] == ports
+
+
+def test_a_group_whose_connections_reach_different_calls_is_still_refused() -> None:
+    """The picture from one lateral into the picture switch and the sound from
+    another into the sound switch: neither switch finds its partner."""
+    error = _refused(
+        """COPY (
+  WITH prog AS (SELECT s.video[1] AS v, s.audio[1] AS a, s.data[1] AS d
+                FROM input('leaf.nut') s)
+  SELECT video(prog.v, ad.video), audio(prog.a, lbar.audio)
+  FROM prog, LATERAL play(prog.d) ad, LATERAL play(prog.d) lbar
+) TO 'out.nut'"""
+    )
+    assert error.code == ErrorCode.UNSUPPORTED_SQL
+    assert error.message.startswith("the group 'switch' reads ")
+    assert error.message.endswith(
+        "each source a group reads is a connection of its own, and each has to reach "
+        "the same calls"
+    )
+
+
 # -- a node's data, read by several and by a lateral, with no copying ffmpeg ---
 
 SELL = "modules/sell_node.wasm"

cli/tests/test_registry.py

@@ -26,6 +26,8 @@
 
 import json
 import subprocess
+import threading
+import time
 from collections.abc import Iterator
 from pathlib import Path
 
@@ -1567,3 +1569,30 @@ def counting_run(argv: list[str], **kwargs: object) -> subprocess.CompletedProce
     help_calls = [c for c in calls if any(a.startswith("filter=") for a in c)]
     assert len(help_calls) == 2  # exactly gblur + overlay, never all ~460+
     assert sum(1 for c in calls if "-filters" in c) == 1
+
+
+def test_a_registry_read_on_two_threads_at_once_is_loaded_once_and_whole(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Two lateral instances compiling at once: the second waits for the
+    first's load rather than finding no filters."""
+    _fake_ffmpeg_present(monkeypatch)
+    calls = _fake_run(monkeypatch)
+    parse = registry_mod._parse_filters_list
+    started = threading.Event()
+
+    def slow_parse(ffmpeg: str) -> object:
+        started.set()
+        time.sleep(0.2)
+        return parse(ffmpeg)
+
+    monkeypatch.setattr(registry_mod, "_parse_filters_list", slow_parse)
+    reg = load()
+    found: list[object] = []
+    first = threading.Thread(target=lambda: found.append(reg.get("gblur")))
+    first.start()
+    assert started.wait(5)
+    found.append(reg.get("gblur"))
+    first.join()
+    assert found[0] is not None and found[0] == found[1]
+    assert sum("-filters" in argv for argv in calls) == 1

docs/dialect.md

@@ -634,6 +634,12 @@ dest    := 'path' | STDOUT | ( value-expression ) | sink(value, ...)
     `port_param`) is given a stream, or a port number: written in its
     place, or by the param's name, `inset(v, port => 9100)`. Given a
     number it binds nothing, and whatever connects to that port is shown.
+  - A `[]` input held on a port and given run-time laterals takes one
+    connection per lateral, every port in one list in its param. Inputs
+    of one group share that list, so a lateral's picture and sound arrive
+    on one connection with one offset, and a lateral may bring some of a
+    group's inputs and not others:
+    `compose(k.v, ARRAY[m.video, c.video], audio => ARRAY[m.audio])`.
   - `RETURNS` a stream; rows alone (`STRUCT(...)[]`, `cue[]`), a data
     stream the query reads while it runs; `STRUCT(<name> <type>, ...)`,
     one field per output the module makes, read off the call

sidecar/NODE-CLI.md

@@ -272,9 +272,13 @@ it is conformed ends `on the GPU`. A coded picture for such an input is
 decoded on the process's device where its decoder shares it (`d3d12va` on
 Windows), as `decode` does for a reader on the GPU, so it never crosses
 host memory; elsewhere it is decoded in software and uploaded once. A
-group of
-hold inputs (`hold.group`) shares one listener and one connection, its
-picture to the video port and its sound to the audio port.
+group of hold inputs (`hold.group`) is one per connection: the inputs of a
+group share each port's listener and connection, its picture to the video
+input and its sound to the audio input. Many inputs (`many`) given a list
+of ports take a stream each per port, in the list's order, so the k-th
+stream of each input arrives on the k-th port. Each connection has its
+own source, feed and offset: a second port's feeder connecting, dropping
+or rejoining leaves the first's feed as it was.
 
 One connection at a time: a second while one is on is accepted and closed
 at once, and a connection that has sent nothing is dropped for one that
@@ -285,15 +289,16 @@ tags and time base from its start on; a source whose pts jump backwards or
 more than a second forwards is a new feed on the same connection. A
 connection may carry any of a group's members, and streams beyond them are
 left on the wire: a feeder of sound alone feeds the sound. Every member of
-a group is told the lead's offset, the picture's where the connection
+a connection is told its lead's offset, the picture's where the connection
 brings one and otherwise its first member's: its `feed-start.at` is the
 lead's, and its `first-pts` the point of its own stream that offset puts
 there, so `(at, first-pts)` maps any member's pts onto the clock. A member
 the connection does not bring (not sent, or its sound refused) gets no feed
 at all on any tick, where one not arrived yet gets a feed and no frames.
 
 A data input whose `interval` names the group (`interval.group`) arrives
-on the same connection, unbound by any pad, as the hold inputs do. What a
+on the group's first port, unbound by any pad, and is placed with that
+connection's offset. What a
 feeder writes so its rows land there, in one NUT:
 
 - the picture (raw video, rgba or yuv420p) and the sound (PCM) as above,
@@ -317,6 +322,8 @@ it. A connection's messages never outlive its feed: those queued when it
 ends are dropped with it. A group data input bound by a pad is refused;
 with its group's hold inputs bound to streams rather than a port, it is
 bound to a stream like any data input and placed with that group's offset.
+A group bound to streams pairs its inputs' streams by position: the first
+stream of each input is one group with one offset, the second another.
 
 The host says what it did on stderr, as lines and as rows behind
 `ffrwd:row `: `{"kind":"listen",...}` once the port is bound,