From 13e3824ee0dc909882273e5c08b842697a7bfc65 Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Mon, 19 Oct 2026 23:10:46 -0700 Subject: [PATCH 1/2] hold: a first-frame source whose buffer is full is primed A first-frame source waits until it holds lead seconds before its feed comes up, and a port feed is read only while its source has room: 600 frames or 96 MiB. Twelve 1080p rgba pictures fill 96 MiB, short of the eighteen a 0.3 s lead takes at 60 fps, so a feed conformed to 1080p never came up and its reader waited on room for the rest of the run, the picture black and its decoder idle. foundmeat's main feed did this on 0.32.4, where it is now conformed to its declared 1920x1080, and on 0.32.3 with a full-size clock. A full buffer now counts as primed, as it already settles a tick for a source bound in the plan. Co-Authored-By: Claude Fable 5.1 --- sidecar/NODE-CLI.md | 4 ++- sidecar/ffrwd-wasm/src/hold.rs | 54 ++++++++++++++++++++++++++++++---- 2 files changed, 52 insertions(+), 6 deletions(-) diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md index 448e595..1ccf003 100644 --- a/sidecar/NODE-CLI.md +++ b/sidecar/NODE-CLI.md @@ -413,7 +413,9 @@ tick that held them. while the source is behind and skipping forward when it catches up, and an audio input hands the tick's samples re-cut from what arrived, or none while the source is behind. A `first-frame` source is primed until - `lead` seconds of it are held, or its end, and scheduled `lead` ahead of + `lead` seconds of it are held, or its end, or as much of it as the host + holds of a source (600 frames or 96 MiB, which is 12 pictures of 1080p + rgba), and scheduled `lead` ahead of the clock, on the clock's grid; a `shared-clock` source's first frame waits for the clock to reach its pts, and one the clock has passed shows at once from where the clock is. A port feed arrives in real time, so a diff --git a/sidecar/ffrwd-wasm/src/hold.rs b/sidecar/ffrwd-wasm/src/hold.rs index fcc388a..5428579 100644 --- a/sidecar/ffrwd-wasm/src/hold.rs +++ b/sidecar/ffrwd-wasm/src/hold.rs @@ -64,6 +64,13 @@ use crate::trace::trace_feed; pub const MAX_HELD_FRAMES: usize = 600; pub const MAX_HELD_BYTES: usize = 96 << 20; +/// Whether `source` holds as many frames or bytes as it may. +fn full(source: &Source) -> bool { + let held = source.members.iter().map(|m| m.queue.len()).sum::(); + let bytes = source.members.iter().map(|m| m.bytes).sum::(); + held >= MAX_HELD_FRAMES || bytes >= MAX_HELD_BYTES +} + /// A step in a source's time, or the clock's, that is backwards or further /// forward than this is a discontinuity. pub const MAX_STEP_SECONDS: f64 = 1.0; @@ -573,11 +580,10 @@ impl Group { let Some(source) = self.sources.back() else { return MAX_HELD_FRAMES; }; - let held = source.members.iter().map(|m| m.queue.len()).sum::(); - let bytes = source.members.iter().map(|m| m.bytes).sum::(); - if held >= MAX_HELD_FRAMES || bytes >= MAX_HELD_BYTES { + if full(source) { return 0; } + let held = source.members.iter().map(|m| m.queue.len()).sum::(); let waiting = match (&self.feed, self.now) { (Some(feed), Some((now, _))) => now < feed.at, (None, Some((now, clock_base))) => { @@ -618,13 +624,15 @@ impl Group { } /// Whether a `first-frame` source holds `lead` seconds of frames, or - /// has ended with some. + /// has ended with some, or holds all it may: a buffer full short of + /// `lead` (large pictures at a high rate) would otherwise wait for + /// frames it has no room to take. fn primed(&self, source: &Source) -> bool { let lead = &source.members[source.lead]; let (Some(first), Some(last)) = (lead.queue.front(), lead.queue.back()) else { return false; }; - if source.closed { + if source.closed || full(source) { return true; } last.pts.saturating_sub(first.pts) >= ticks_of(self.hold.lead, lead.base) @@ -1558,6 +1566,42 @@ mod tests { assert_eq!(mark(&handed[0]), Some(11)); } + #[test] + fn a_first_frame_port_feed_whose_buffer_fills_short_of_lead_still_comes_up() { + let (mut g, lines) = group( + hold(Anchor::FirstFrame, 0.3, None, None), + vec![video_member(1, MILLIS)], + true, + ); + let grid = Grid::exact(); + // Four pictures fill the buffer, 50 ms of a lead of 300. + let picture = Arc::new(vec![7u8; MAX_HELD_BYTES / 4 + 1]); + let big = |pts: i64| TickFrame { + pts, + duration: None, + data: Arc::clone(&picture).into(), + rows: Vec::new(), + }; + for pts in [0, 16, 33] { + g.arrive(0, big(pts)); + } + let handed = g.tick(100, Some(101), THIRTIETHS, &grid, true); + assert!(handed[0].feed.is_none(), "room left: still priming"); + g.arrive(0, big(50)); + assert_eq!(g.room(), 0, "the buffer is full"); + let handed = g.tick(101, Some(102), THIRTIETHS, &grid, true); + let feed = handed[0] + .feed + .clone() + .expect("a full buffer starts the feed"); + assert_eq!(feed.start.first_pts, 0); + assert!( + lines.lock().unwrap()[0].contains("comes up"), + "{:?}", + lines.lock().unwrap() + ); + } + #[test] fn a_shared_clock_feed_waits_for_the_clock_and_skips_what_the_clock_passed() { let (mut g, _) = group( From a0f75979c68869f33f7a1781bf2d8616bc6efbfb Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Mon, 19 Oct 2026 23:10:47 -0700 Subject: [PATCH 2/2] cli: a lateral that scales its picture is hinted the size it scales to A run-time lateral's picture size was read only from a shape on an input's own picture, so foundmeat's cams, scaled to 800x450 by their parameters' defaults, were still conformed to the 16x16 clock. The compile now follows the column through fps, setpts and format, and through a scale to a width and a height, each a number or a parameter the call writes or gives a DEFAULT, the shape's own values read the same way. Co-Authored-By: Claude Fable 5.1 --- cli/ffrwd/functions.py | 118 ++++++++++++++++++++++++------------- cli/tests/test_laterals.py | 54 +++++++++++++++++ docs/dialect.md | 16 +++-- docs/known_gaps.md | 16 ++--- 4 files changed, 150 insertions(+), 54 deletions(-) diff --git a/cli/ffrwd/functions.py b/cli/ffrwd/functions.py index 371a079..6d67817 100644 --- a/cli/ffrwd/functions.py +++ b/cli/ffrwd/functions.py @@ -202,9 +202,8 @@ class RuntimeLateral: as written. `params` are the values bound per message and `written` the ones the call wrote itself, as SQL. `needs` are the script's own definitions an instance calls, in the order the script wrote them. - `sizes` are the picture sizes the body declares: each video column that - is an input's own picture with a ``shape`` naming its width and height, - as ``(column, width, height)``. + `sizes` are the picture sizes the body makes, where the compile can tell + them, as ``(column, width, height)`` (:func:`_declared_sizes`). """ function: str @@ -3694,14 +3693,46 @@ def _runs_per_message(function: _Function) -> bool: ) -def _declared_sizes(function: _Function) -> tuple[tuple[str, int, int], ...]: - """The picture size a run-time lateral's body declares for each video - column that is an input's own first picture, ``f.video[1] AS video`` over - ``input(url, shape => STRUCT(1920 AS width, 1080 AS height, ...)) f``: - ``(column, width, height)``. A column made any other way, or a shape - whose size is not written as numbers, has none. +# Filters a lateral's picture may pass through with its size kept, as the +# compile reads a body for the size its picture leaves at. +_SIZE_KEEPING = frozenset({"fps", "setpts", "settb", "format", "null", "copy", "fifo"}) + + +def _declared_sizes( + function: _Function, written: Mapping[str, str] +) -> tuple[tuple[str, int, int], ...]: + """The picture size a run-time lateral's body makes for each video + column, where the compile can tell it: ``(column, width, height)``. + + The column is an input's first picture, ``f.video[1]``, whose + ``input(..., shape => STRUCT(... AS width, ... AS height, ...))`` gives + its size, passed through filters that keep the size (``fps``, + ``setpts``, ``format``), or a ``scale`` of it to a width and a height. + A value is a number, or a parameter: what the call wrote, else its + ``DEFAULT``. A column made any other way has none. """ body = function.body + params = {param.name: param for param in function.params} + + def number(node: exp.Expr | None) -> int | None: + if isinstance(node, exp.Paren): + node = node.this + if isinstance(node, exp.Literal) and not node.is_string: + value = node.to_py() + return value if isinstance(value, int) and value > 0 else None + if isinstance(node, exp.Column) and not node.table: + name = node.name + if name in written: + try: + found = float(written[name]) + except ValueError: + return None + return int(found) if found.is_integer() and found > 0 else None + param = params.get(name) + if param is not None and param.default is not None: + return number(param.default) + return None + items: list[exp.Expr] = [] found = body.args.get("from_") if isinstance(found, exp.From): @@ -3721,42 +3752,49 @@ def _declared_sizes(function: _Function) -> tuple[tuple[str, int, int], ...]: struct = argument.expression if not isinstance(struct, exp.Struct): continue - fields: dict[str, object] = {} - for entry in struct.expressions: - if isinstance(entry, exp.PropertyEQ) and isinstance(entry.expression, exp.Literal): - literal = entry.expression - fields[str(entry.this.name).lower()] = ( - None if literal.is_string else literal.to_py() - ) - width, height = fields.get("width"), fields.get("height") - if ( - isinstance(width, int) - and isinstance(height, int) - and not isinstance(width, bool) - and not isinstance(height, bool) - and width > 0 - and height > 0 - ): + fields = { + str(entry.this.name).lower(): entry.expression + for entry in struct.expressions + if isinstance(entry, exp.PropertyEQ) and isinstance(entry.expression, exp.Expr) + } + width, height = number(fields.get("width")), number(fields.get("height")) + if width is not None and height is not None: shapes[_ident_name(alias.this)] = (width, height) + + def size(node: exp.Expr) -> tuple[int, int] | None: + if isinstance(node, exp.Dot) and isinstance(node.expression, exp.Anonymous): + node = node.expression + if isinstance(node, exp.Bracket): + read = node.this + if isinstance(read, exp.Column) and read.name == "video" and subscript_index(node) == 1: + return shapes.get(read.table) + return None + if not isinstance(node, exp.Anonymous) or not node.expressions: + return None + name = str(node.this).lower() + source, *rest = node.expressions + if name in _SIZE_KEEPING: + return size(source) + if name != "scale": + return None + named = {kwarg_name(arg): arg.expression for arg in rest if isinstance(arg, exp.Kwarg)} + if set(named) - {"w", "h", "width", "height"}: + return None + positional = [arg for arg in rest if not isinstance(arg, exp.Kwarg)] + width = positional[0] if positional else named.get("w", named.get("width")) + height = positional[1] if len(positional) > 1 else named.get("h", named.get("height")) + scaled = number(width), number(height) + return (scaled[0], scaled[1]) if scaled[0] and scaled[1] else None + sizes: list[tuple[str, int, int]] = [] for column in function.columns or (): if column.type != _WASM_STREAM: continue for selected in body.expressions: - named = selected.alias_or_name if isinstance(selected, exp.Alias) else None - if named != column.name: - continue - read = selected.this - if not ( - isinstance(read, exp.Bracket) - and isinstance(read.this, exp.Column) - and read.this.name == "video" - and subscript_index(read) == 1 - ): - continue - size = shapes.get(read.this.table) - if size is not None: - sizes.append((column.name, *size)) + if isinstance(selected, exp.Alias) and selected.alias_or_name == column.name: + made = size(selected.this) + if made is not None: + sizes.append((column.name, *made)) return tuple(sizes) @@ -5991,7 +6029,7 @@ def _expand_runtime_lateral( columns=function.columns or (), written=written, needs=self._needs(function), - sizes=_declared_sizes(function), + sizes=_declared_sizes(function, written), ) line, col = _pos(item) columns: dict[str, exp.Expr] = {} diff --git a/cli/tests/test_laterals.py b/cli/tests/test_laterals.py index 73df308..543c0d4 100644 --- a/cli/tests/test_laterals.py +++ b/cli/tests/test_laterals.py @@ -1045,6 +1045,60 @@ def test_a_laterals_declared_picture_size_is_its_streams_hint_in_port_order() -> ] +# foundmeat's feeds: the main feed's picture and sound as they come, a +# cam's scaled to the size its parameters give, its sound beside it. +_FEEDS = """CREATE FUNCTION main_feed(launch data_stream, url text) +RETURNS TABLE(video video_stream, audio audio_stream) AS $$ + SELECT f.video[1] AS video, f.audio[1] AS audio + FROM input(url, format => 'mpegts', shape => STRUCT(1920 AS width, 1080 AS height, + 60 AS fps, 'h264' AS video_codec, 48000 AS rate, 2 AS channels)) f +$$ LANGUAGE sql; +CREATE FUNCTION cam_feed(launch data_stream, url text, + width number DEFAULT 800, height number DEFAULT 450, + rate number DEFAULT 30) +RETURNS TABLE(video video_stream, audio audio_stream) AS $$ + SELECT fps(scale(f.video[1], width, height), rate) AS video, f.audio[1] AS audio + FROM input(url, format => 'mpegts', shape => STRUCT(width AS width, height AS height, + rate AS fps, 'h264' AS video_codec, 48000 AS rate, 2 AS channels)) f +$$ LANGUAGE sql;""" +_THREE_FEEDS = """COPY ( + WITH prog AS (SELECT s.video[1] AS v, s.data[1] AS d FROM input('leaf.nut') s) + SELECT compose(prog.v, ARRAY[m.video, c1.video, c2.video], + audio => ARRAY[m.audio, c1.audio, c2.audio]) + FROM prog, LATERAL main_feed(prog.d) m, LATERAL cam_feed(prog.d) c1, + LATERAL cam_feed(prog.d, width => 640, height => 360) c2 +) TO 'out.nut'""" + + +def test_a_lateral_that_scales_is_hinted_the_size_it_scales_to_beside_its_sound() -> None: + """Each feed brings its picture and its sound on one connection. The + main feed's picture is hinted the size its shape declares; a cam's, the + size its body scales to: its parameters' defaults, or what the call + wrote. The sound of each is hinted nothing.""" + modules = _compose_modules({"type": "array", "items": {"type": "integer"}}) + plan = compile_all( + _FEEDS + "\n" + _COMPOSE_AUDIO + "\n" + _THREE_FEEDS, + describe=lambda path: modules[path], + shape=_compose_audio_shape, + ).plan + assert plan is not None + compose = _compose_node(plan) + ports = [lateral.connections[0].port for lateral in plan.laterals] + assert compose.args["port"] == ports + unknown = {"rate": None} + assert json.loads(compose.bound)[1:] == [ + { + "input": "inputs", + "streams": [ + {"rate": None, "width": 1920, "height": 1080}, + {"rate": None, "width": 800, "height": 450}, + {"rate": None, "width": 640, "height": 360}, + ], + }, + {"input": "audio", "streams": [unknown, unknown, unknown]}, + ] + + def test_a_port_written_by_hand_stays_one_integer_where_a_list_also_fits() -> None: query = """COPY ( SELECT compose(s.video[1], port => 9100.0) FROM input('leaf.nut') s diff --git a/docs/dialect.md b/docs/dialect.md index d3a9b91..7fcbe48 100644 --- a/docs/dialect.md +++ b/docs/dialect.md @@ -702,12 +702,16 @@ dest := 'path' | STDOUT | ( value-expression ) | sink(value, ...) its clock, a rate clock's own, the rate of the input a `rate-of` clock names, or its clock input's over its stride. A feed by port, a self-clocked node's output and data have none. A lateral's picture - whose body reads an input's own first picture with a declared size - (`f.video[1] AS video` over `input(url, shape => STRUCT(1920 AS - width, 1080 AS height, ...)) f`) carries that width and height, and - the sidecar conforms the feed to them; without one, the feed is - conformed to the size of the input its port follows, or the clock - input's. Each call carries the + carries the width and height its body makes, where the compile can + tell them, and the sidecar conforms the feed to them: an input's own + first picture whose `shape` gives its size (`f.video[1] AS video` + over `input(url, shape => STRUCT(1920 AS width, 1080 AS height, + ...)) f`), through `fps`, `setpts` or `format`, or a `scale` of it to + a width and a height, each a number or a parameter the call writes or + gives a `DEFAULT` (`fps(scale(f.video[1], width, height), rate)`). A + message binding another size has its picture conformed to that one. + Without a size, the feed is conformed to the size of the input its + port follows, or the clock input's. Each call carries the list it was asked with (`-bound =`), which the host asks the shape with again, so the shape a node runs with is the one the query compiled against. A node at a COPY's TO is asked once for its diff --git a/docs/known_gaps.md b/docs/known_gaps.md index 4b17ecd..dfd888a 100644 --- a/docs/known_gaps.md +++ b/docs/known_gaps.md @@ -73,14 +73,14 @@ from. of one ends with an upload. Only 8-bit and 10-bit 4:2:0 pictures, and 8-bit 4:4:4 HEVC on CUDA, stay on the card; 4:2:2, 10-bit 4:4:4 and an input whose pixel format the probe does not report run on the CPU. -- **A lateral's picture size is known only where its body declares it.** - A node input fed on a port by a run-time lateral is conformed to the - size the lateral's body declares (`f.video[1]` of an `input(...)` whose - `shape` gives `width` and `height`). A body that reads a probed file, or - scales its picture, declares none, and its feed is conformed to the size - of the input the port follows, or the clock input's, which for a small - clock is a blur. Declare the `shape` in the body to keep the picture's - own size. +- **A lateral's picture size is known only where its body says it.** A + node input fed on a port by a run-time lateral is conformed to the size + the lateral's body makes: its input's declared `shape`, or a `scale` to + numbers or to parameters with a value the call writes or a `DEFAULT`. + A body that reads a probed file, scales to a value each message binds + with no `DEFAULT`, or passes its picture through other filters says + none, and its feed is conformed to the size of the input the port + follows, or the clock input's, which for a small clock is a blur. - **The sidecar has no device filters on macOS.** Its bundled libavfilter carries none for VideoToolbox, so a filter between two modules holding their pictures on the GPU runs in ffmpeg there, and the