From 4b5a80c7583b893c5b64bd698d2a294cd5c4e7a1 Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Mon, 19 Oct 2026 21:34:05 -0700 Subject: [PATCH 1/2] sidecar: a port feed is conformed to the size the compile declares for it A hold input fed on a port was conformed to the size of the input it follows (accepts.like) or, failing that, the clock input's. blitz's compose_av follows nothing, so a 1080p feed beside a 16x16 clock reached it at 16x16. A stream of a port-served input in -bound may now carry width and height. Where it does, the host tells the node that size at init and conforms the feed to it; like and then the clock are the fallbacks as before. The pixel format still comes from the port's accepts, else from the input it follows. gpu-box gains w=own, a port-fed w with no like whose output takes w's size, and the tests run it on both memories with a 640x360 feed against a 16x16 clock. Co-Authored-By: Claude Fable 5.1 --- sidecar/NODE-CLI.md | 19 ++- sidecar/NODE-SHAPE.md | 15 ++- sidecar/ffrwd-wasm/src/node_graph.rs | 160 +++++++++++++++++++++++-- sidecar/ffrwd-wasm/src/shape_json.rs | 39 ++++-- sidecar/ffrwd-wasm/tests/gpu_frames.rs | 158 ++++++++++++++++++++++++ sidecar/ffrwd-wasm/tests/node_shape.rs | 4 + sidecar/modules/gpu-box/src/lib.rs | 92 ++++++++++++-- sidecar/runtime/src/node.rs | 6 +- 8 files changed, 457 insertions(+), 36 deletions(-) diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md index ba55ab98..448e5952 100644 --- a/sidecar/NODE-CLI.md +++ b/sidecar/NODE-CLI.md @@ -119,6 +119,15 @@ chains takes its `-bound` flags in turn: the k-th given for the name goes to the k-th chain calling it. A node source binds nothing and carries `[]`. +A stream of an input a port serves may also carry `width` and `height`: +the picture size the compile declared for that feed, which the host +conforms the feed to (see "Feeds by port"). The module is not handed +them. The k-th stream of such an input is the one on the k-th port of its +param's list, so the compiler writes an input's streams in the order of +its ports there: + + -bound 'compose=[{"input":"v","streams":[{"rate":{"num":60,"den":1}}]},{"input":"inputs","streams":[{"rate":null,"width":1920,"height":1080}]}]' + The list is checked against the pads, and a refusal names the module and the input: @@ -256,9 +265,13 @@ or every output's reader has closed, and a reader closing is a clean end. ## Feeds by port A hold input the call gives a port rather than a stream is bound a stream -of the host's own at `init`, in the port's format: the size and colour of -the input it follows (`accepts.like`, or the clock input), in the first -pixel or sample format the port accepts. The host listens on the port +of the host's own at `init`, in the port's format. A picture takes the +size `-bound` gives that stream (`width` and `height`, the size the +compile declared for the feed), else the size of the input it follows +(`accepts.like`), else the clock input's; sound takes the rate and layout +of the input it follows or of the clock input. Either is in the first +pixel or sample format the port accepts, else that input's, with that +input's colour where there is one. The host listens on the port from the moment the run starts, and whatever connects and writes a NUT of video and PCM is the input's source for as long as it stays: its picture is conformed to that format on the way in (rgba and yuv420p, at any size; diff --git a/sidecar/NODE-SHAPE.md b/sidecar/NODE-SHAPE.md index f2c8c390..cc68e5f2 100644 --- a/sidecar/NODE-SHAPE.md +++ b/sidecar/NODE-SHAPE.md @@ -35,11 +35,15 @@ rows or data-filter interface is refused. stream bound to it, in the order the call names them. `rate` is a video stream's frame rate or an audio stream's sample rate over 1, `null` (or absent) where nothing settles it before the run; `streams` absent is - one stream with no rate. Or the comma list of the names alone, each one + one stream with no rate. A stream of an input a port serves may also + give `width` and `height`, both or neither, the size its feed is + conformed to (`NODE-CLI.md`, "Feeds by port"); the module's `shape` is + handed the rate alone. Or the comma list of the names alone, each one stream with no rate, a name written again one stream more (`v,v,v` binds three to `v`). Absent or empty binds none. A name the shape does not declare is refused, and so is an input named twice in the JSON, an - empty `streams`, a rate that is not positive and a key not listed here. + empty `streams`, a rate that is not positive, a width without a height + (or the reverse) or either of them 0, and a key not listed here. The run is handed the list each call was asked with as `-bound` (`NODE-CLI.md`, "The shape a call was planned with"), and asks the shape with it again. @@ -334,7 +338,10 @@ video. A module of the node world alone, or of an older world, has An input held on the GPU that follows another input (`accepts.like`) is told that input's size at `init`, and the host resizes its frames on the device as they arrive, with a bilinear filter; an input in host memory -bound to a stream is handed it at the stream's size. A frame reaches a +bound to a stream is handed it at the stream's size. An input fed on a +port is told the size `-bound` declares for its feed, else that of the +input it follows, else the clock input's (`NODE-CLI.md`, "Feeds by +port"), in either memory, and each feed is conformed to that size. A frame reaches a port in the port's format and size whichever way it arrived: put on the GPU as it reaches the node, left there by a decoder (`AV-EDGE.md`), or fed on a port (`NODE-CLI.md`). Each kind of conversion the host makes on @@ -361,6 +368,8 @@ frame: its stream, its pts, and its feed's start (`feed-start.at` and `v` and `out` on the GPU, `{"memory":"host"}` holds neither, and the two write the same bytes. `{"w":"pad"}` adds a second input, `w`, that follows `v`, and `{"w":"port","port":N}` feeds it on a port. +`{"w":"own","port":N}` feeds it on a port with no `like`, and `out` takes +`w`'s size. A module that runs a model on its frames includes `ffrwd:gpu/gpu-nn-module` instead and asks `ffrwd:gpu/nn.frame-tensor` for a frame as the model's diff --git a/sidecar/ffrwd-wasm/src/node_graph.rs b/sidecar/ffrwd-wasm/src/node_graph.rs index 7571215c..1302c719 100644 --- a/sidecar/ffrwd-wasm/src/node_graph.rs +++ b/sidecar/ffrwd-wasm/src/node_graph.rs @@ -1950,7 +1950,9 @@ fn listen_on( feeds: &mut Vec, groups: &mut Vec<(Option, usize)>, ) -> Result<()> { - let format = conformed(name, shape, input, bound)?; + let k = bound.iter().filter(|b| b.port == input.name).count(); + let hint = hint_of(told, &input.name, k); + let format = conformed(name, shape, input, bound, &hint)?; let id = defs.len() as u32; let micros = TimeBase { num: 1, @@ -1979,13 +1981,7 @@ fn listen_on( rendition: RenditionMeta::default(), row: None, }; - let k = bound.iter().filter(|b| b.port == input.name).count(); - bound.push(bound_stream( - &input.name, - id, - &def, - hint_of(told, &input.name, k), - )); + bound.push(bound_stream(&input.name, id, &def, hint)); let timing = input.accepts.wants == Wants::Timing; let member = FeedMember { id, @@ -2121,12 +2117,15 @@ fn port_param( /// The format a port feed is conformed to: the input it follows (`like`), /// or the clock input's where that is the same kind, with the port's own -/// pixel or sample format where it names one. +/// pixel or sample format where it names one. A picture whose size the +/// compile declared (`hint.size`) takes that size instead, and needs a +/// followed input only for its pixel format where the port names none. fn conformed( name: &str, shape: &NodeShape, input: &ffrwd_wasm_runtime::node::InputPort, bound: &[BoundStream], + hint: &StreamHint, ) -> Result { let followed = input.accepts.like.clone().or_else(|| match &shape.clock { Clock::Input(clock) => Some(clock.clone()), @@ -2137,6 +2136,29 @@ fn conformed( .and_then(|port| bound.iter().find(|b| &b.port == port)) .filter(|b| b.format.kind() == input.kind) .map(|b| b.format.clone()); + if let (PortKind::Video, Some((width, height))) = (input.kind, hint.size) { + let followed = match &model { + Some(StreamFormat::Video(v)) => Some(v), + _ => None, + }; + let pix_fmt = match (input.accepts.pixel_formats.first(), followed) { + (Some(first), _) => intern(first, KNOWN_PIX_FMTS, "pixel format")?, + (None, Some(v)) => v.pix_fmt, + (None, None) => bail!( + "{name} input '{}' is held on a port, accepts any pixel format and follows no \ + bound video input (accepts.like), so the host cannot say what a feed is \ + conformed to", + input.name + ), + }; + return Ok(StreamFormat::Video(VideoFormat { + width, + height, + pix_fmt, + frame_len: crate::frame_len_for(pix_fmt, width, height)?, + color: followed.and_then(|v| v.color), + })); + } let Some(model) = model else { bail!( "{name} input '{}' is held on a port and follows no bound {} input (accepts.like), \ @@ -2979,6 +3001,126 @@ mod tests { ); } + #[test] + fn a_port_feed_takes_its_declared_size_else_its_like_else_the_clocks() { + use ffrwd_wasm_runtime::node::{Accepts, Anchor, Hold, InputPort}; + let video = |width, height, pix_fmt| { + StreamFormat::Video(VideoFormat { + width, + height, + pix_fmt, + frame_len: crate::frame_len_for(pix_fmt, width, height).unwrap(), + color: None, + }) + }; + let stream = |port: &str, format: StreamFormat| BoundStream { + port: port.to_string(), + id: 0, + info: info_for(&format), + time_base: TimeBase { num: 1, den: 25 }, + format, + rendition: RenditionMeta::default(), + row: None, + decode_delay: 0, + latency: None, + hint: StreamHint::default(), + }; + let port = + |name: &str, pairing: Pairing, pixel_formats: &[&str], like: Option<&str>| InputPort { + name: name.to_string(), + kind: PortKind::Video, + required: false, + many: false, + pairing, + rows: RowsUse::Ignore, + window: 1, + stride: 1, + accepts: Accepts { + pixel_formats: pixel_formats.iter().map(|p| p.to_string()).collect(), + like: like.map(str::to_string), + ..Accepts::default() + }, + schema: None, + }; + let held = Pairing::Hold(Hold { + anchor: Anchor::FirstFrame, + lead: 0.0, + linger: None, + timeout: None, + group: None, + port_param: Some("port".to_string()), + }); + let shape = |clock: Clock, w: InputPort| NodeShape { + inputs: vec![ + port("v", Pairing::Lockstep, &[], None), + port("u", Pairing::Lockstep, &[], None), + w, + ], + outputs: Vec::new(), + clock, + pure: true, + one_to_one: true, + bounded: true, + relation: Vec::new(), + }; + let bound = [ + stream("v", video(16, 16, "rgba")), + stream("u", video(320, 180, "yuv420p")), + ]; + let clocked = || Clock::Input("v".to_string()); + let declared = StreamHint { + rate: None, + size: Some((640, 360)), + }; + let none = StreamHint::default(); + let conform = |clock, w: InputPort, hint: &StreamHint| { + let shape = shape(clock, w.clone()); + conformed("n", &shape, &w, &bound, hint) + }; + + let w = port("w", held.clone(), &["rgba"], None); + assert_eq!( + conform(clocked(), w.clone(), &declared).unwrap(), + video(640, 360, "rgba"), + "the declared size" + ); + assert_eq!( + conform(clocked(), w, &none).unwrap(), + video(16, 16, "rgba"), + "nothing declared: the clock's size" + ); + + let like = port("w", held.clone(), &["rgba"], Some("u")); + assert_eq!( + conform(clocked(), like.clone(), &none).unwrap(), + video(320, 180, "rgba"), + "the size of the input it follows" + ); + assert_eq!( + conform(clocked(), like, &declared).unwrap(), + video(640, 360, "rgba"), + "a declared size wins over like" + ); + + let any = port("w", held.clone(), &[], None); + assert_eq!( + conform(clocked(), any.clone(), &declared).unwrap(), + video(640, 360, "rgba"), + "no pixel format of its own: the clock's" + ); + let rate = || Clock::Rate(Rational { num: 25, den: 1 }); + assert_eq!( + conform(rate(), port("w", held, &["yuv420p"], None), &declared).unwrap(), + video(640, 360, "yuv420p"), + "a declared size needs no input to follow where the port names a pixel format" + ); + let refused = conform(rate(), any, &declared).expect_err("no pixel format anywhere"); + assert!( + refused.to_string().contains("accepts any pixel format"), + "{refused:#}" + ); + } + /// shape-probe, built for wasm32-wasip2 once per test binary; none /// where that target is not installed (CI's module-free job). fn probe() -> Option { diff --git a/sidecar/ffrwd-wasm/src/shape_json.rs b/sidecar/ffrwd-wasm/src/shape_json.rs index 4f80ffe3..87fd9a4a 100644 --- a/sidecar/ffrwd-wasm/src/shape_json.rs +++ b/sidecar/ffrwd-wasm/src/shape_json.rs @@ -43,6 +43,11 @@ struct BindingJson { struct HintJson { #[serde(default)] rate: Option, + /// A picture fed on a port: the size the compile declared for it. + #[serde(default)] + width: Option, + #[serde(default)] + height: Option, } #[derive(Deserialize)] @@ -79,17 +84,31 @@ pub fn parse_bound(text: &str) -> Result> { } Some(streams) => streams .into_iter() - .map(|hint| match hint.rate { - Some(RationalJson { num, den }) if num <= 0 || den <= 0 => bail!( - "--bound gives input '{}' the rate {num}/{den}, which is no rate", - binding.input - ), - rate => Ok(StreamHint { - rate: rate.map(|r| Rational { - num: r.num, - den: r.den, + .map(|hint| { + let size = match (hint.width, hint.height) { + (None, None) => None, + (Some(w), Some(h)) if w > 0 && h > 0 => Some((w, h)), + (w, h) => bail!( + "--bound gives input '{}' the size {}x{}; a size is a width and a \ + height, both above 0", + binding.input, + w.map_or("?".to_string(), |w| w.to_string()), + h.map_or("?".to_string(), |h| h.to_string()) + ), + }; + match hint.rate { + Some(RationalJson { num, den }) if num <= 0 || den <= 0 => bail!( + "--bound gives input '{}' the rate {num}/{den}, which is no rate", + binding.input + ), + rate => Ok(StreamHint { + rate: rate.map(|r| Rational { + num: r.num, + den: r.den, + }), + size, }), - }), + } }) .collect::>()?, }; diff --git a/sidecar/ffrwd-wasm/tests/gpu_frames.rs b/sidecar/ffrwd-wasm/tests/gpu_frames.rs index e43b3314..c16c4ce0 100644 --- a/sidecar/ffrwd-wasm/tests/gpu_frames.rs +++ b/sidecar/ffrwd-wasm/tests/gpu_frames.rs @@ -1937,6 +1937,164 @@ fn a_feed_held_on_the_gpu_is_conformed_on_the_device() { assert!(max <= 1 && mean < 0.05, "{max} apart, mean {mean}"); } +/// A run's pictures, each with its pts, and its stderr. +type Written = (Vec<(i64, Vec)>, String); + +/// Runs gpu_box with `w=own` and `memory` over a 16x16 programme, its `w` +/// fed on a port with `feed`-sized pattern pictures and given the size +/// `declared` by `-bound`, as the compile gives a lateral's declared shape: +/// the output's pictures and the run's stderr, or None where `memory` is +/// gpu and the machine has no adapter. +fn declared_feed_run(memory: &str, feed: (u32, u32), declared: (u32, u32)) -> Option { + use std::net::{TcpListener, TcpStream}; + use std::time::{Duration, Instant}; + let module = module(); + let clock = 16u32; + let port = { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind a free port"); + listener.local_addr().expect("its address").port() + }; + let wiring = format!("[v=0:v]gpu_box=memory={memory}:w=own:port={port}:size=3[out=o]"); + let bound = format!( + r#"gpu_box=[{{"input":"v"}},{{"input":"w","streams":[{{"width":{},"height":{}}}]}}]"#, + declared.0, declared.1 + ); + let mut child = Command::new(env!("CARGO_BIN_EXE_ffrwd-wasm")) + .args(["-gpu", &module, "-f", "nut", "-i", "-", "-m"]) + .arg(format!("gpu_box={module}")) + .args(["-bound", &bound]) + .args(["-filter_complex", &wiring, "-map", "[o]", "-f", "nut", "-"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn ffrwd-wasm"); + let (listening, stderr_reader) = watch_for_listening(&mut child); + let mut stdin = child.stdin.take().expect("stdin is piped"); + let programme = std::thread::spawn(move || { + let mut stream = Stream::video("rgba", clock, clock, TIME_BASE).expect("rgba"); + stream.frame_rate = Some((25, 1)); + let mut muxer = Muxer::new(&mut stdin, &stream).expect("headers"); + if muxer.write_frame(0, &green(clock, clock)).is_err() { + return; + } + let _ = muxer.flush(); + let started = listening + .recv_timeout(LISTENING_BOUND) + .unwrap_or_else(|_| Instant::now()); + for k in 1..50i64 { + let due = started + Duration::from_millis(40 * k as u64); + if let Some(wait) = due.checked_duration_since(Instant::now()) { + std::thread::sleep(wait); + } + if muxer.write_frame(k, &green(clock, clock)).is_err() { + return; + } + let _ = muxer.flush(); + } + let _ = muxer.finish(); + }); + let feeder = std::thread::spawn(move || { + let sent = pattern(feed.0, feed.1); + let until = Instant::now() + Duration::from_secs(10); + let socket = loop { + std::thread::sleep(Duration::from_millis(100)); + if let Ok(socket) = TcpStream::connect(("127.0.0.1", port)) { + break socket; + } + if Instant::now() > until { + return; + } + }; + let mut stream = Stream::video("rgba", feed.0, feed.1, TIME_BASE).expect("rgba"); + stream.frame_rate = Some((25, 1)); + let Ok(mut muxer) = Muxer::new(socket, &stream) else { + return; + }; + for k in 0..25i64 { + if muxer.write_frame(k, &sent).is_err() || muxer.flush().is_err() { + return; + } + std::thread::sleep(Duration::from_millis(40)); + } + let _ = muxer.finish(); + }); + let output = child.wait_with_output().expect("wait for ffrwd-wasm"); + let _ = (programme.join(), feeder.join()); + let stderr = stderr_reader.join().expect("read stderr"); + if memory == "gpu" && no_gpu(&output) { + eprintln!("skipped: this machine has no GPU adapter"); + return None; + } + assert!( + output.status.success(), + "exited {:?}:\n{stderr}", + output.status.code() + ); + Some((read_nut(&output.stdout), stderr)) +} + +/// The output's pictures the feed reached, every one at `size`: the rest +/// are transparent. +fn fed_pictures(pictures: &[(i64, Vec)], size: (u32, u32), stderr: &str) -> Vec> { + assert_eq!(pictures.len(), 50, "a picture a programme frame:\n{stderr}"); + for (pts, picture) in pictures { + assert_eq!( + picture.len(), + (size.0 * size.1 * 4) as usize, + "picture {pts} is the feed's declared size" + ); + } + let fed: Vec> = pictures + .iter() + .map(|(_, p)| p.clone()) + .filter(|p| p.iter().any(|&b| b != 0)) + .collect(); + assert!( + fed.len() >= 10, + "the feed shows on {} pictures of 50:\n{stderr}", + fed.len() + ); + fed +} + +#[test] +fn a_port_feed_is_handed_at_the_size_the_compile_declared_not_the_clocks() { + let size = (640u32, 360u32); + let mut theirs = pattern(size.0, size.1); + for y in 0..3usize { + for x in 0..3usize { + let at = (y * size.0 as usize + x) * 4; + theirs[at..at + 4].copy_from_slice(&[255, 0, 0, 255]); + } + } + for memory in ["host", "gpu"] { + let Some((pictures, stderr)) = declared_feed_run(memory, size, size) else { + continue; + }; + assert!( + !stderr.contains("conformed to"), + "{memory}: a feed at its declared size is not converted:\n{stderr}" + ); + for ours in fed_pictures(&pictures, size, &stderr) { + assert!( + ours == theirs, + "{memory}: a fed picture is the feed's own, with the box" + ); + } + } + + // A feed of another size is scaled to the declared one, not the clock's. + let small = (320u32, 180u32); + let (pictures, stderr) = declared_feed_run("host", small, size).expect("a host run"); + let conformed = + "feed: gpu_box 'w': the feeder's picture is 320x180 rgba, conformed to 640x360 rgba"; + assert!(stderr.contains(conformed), "{stderr}"); + for ours in fed_pictures(&pictures, size, &stderr) { + assert!(has_the_box(&ours, size.0 as usize, 3)); + } +} + #[test] fn a_coded_feed_held_on_the_gpu_is_decoded_on_the_device() { use std::net::{TcpListener, TcpStream}; diff --git a/sidecar/ffrwd-wasm/tests/node_shape.rs b/sidecar/ffrwd-wasm/tests/node_shape.rs index 31f8fe8d..af16cd58 100644 --- a/sidecar/ffrwd-wasm/tests/node_shape.rs +++ b/sidecar/ffrwd-wasm/tests/node_shape.rs @@ -187,6 +187,10 @@ fn shape_takes_the_bound_list_as_json_and_hands_each_streams_rate_to_the_module( r#"[{"input":"v","streams":[{"rate":{"num":0,"den":1}}]}]"#, "which is no rate", ), + ( + r#"[{"input":"v","streams":[{"width":640}]}]"#, + "the size 640x?; a size is a width and a height", + ), (r#"[{"input":"v","nope":1}]"#, "unknown field"), ] { let run = sidecar(&["--shape", &probe, "--bound", bound]); diff --git a/sidecar/modules/gpu-box/src/lib.rs b/sidecar/modules/gpu-box/src/lib.rs index 36e34d18..fb30bef0 100644 --- a/sidecar/modules/gpu-box/src/lib.rs +++ b/sidecar/modules/gpu-box/src/lib.rs @@ -21,6 +21,9 @@ //! paired in lockstep, on the GPU only, where the host resizes it), or fed //! on the loopback port `port` names (`w=port`, a hold input). A tick that //! has a frame of `w` draws the box on that frame rather than on `v`'s. +//! `w=own` feeds `w` on the port too, with no `like`: it is told the size +//! the host conforms its feed to, and `out` takes `w`'s size (`like` `w`). +//! A tick with no frame of `w` writes a transparent frame of that size. wit_bindgen::generate!({ path: ["../../wit", "../../wit-gpu", "wit"], @@ -48,7 +51,7 @@ use wasi::webgpu::webgpu::{ GpuTextureDescriptor, GpuTextureFormat, GpuTextureUsage, }; -const SCHEMA: &str = r#"{"type":"object","properties":{"memory":{"type":"string","enum":["gpu","host"]},"out":{"type":"string","enum":["gpu","host"]},"size":{"type":"integer","minimum":1},"w":{"type":"string","enum":["pad","port"]},"port":{"type":"integer"}},"additionalProperties":false}"#; +const SCHEMA: &str = r#"{"type":"object","properties":{"memory":{"type":"string","enum":["gpu","host"]},"out":{"type":"string","enum":["gpu","host"]},"size":{"type":"integer","minimum":1},"w":{"type":"string","enum":["pad","port","own"]},"port":{"type":"integer"}},"additionalProperties":false}"#; /// The box's colour, rgba. const COLOUR: [u8; 4] = [255, 0, 0, 255]; @@ -118,8 +121,8 @@ fn parse(text: &str) -> Result { if params.w.as_deref() == Some("pad") && params.memory != "gpu" { return Err("w=pad takes memory=gpu, where the host resizes w".to_string()); } - if params.w.as_deref() == Some("port") && params.port.is_none() { - return Err("w=port names the port w is fed on with port".to_string()); + if matches!(params.w.as_deref(), Some("port" | "own")) && params.port.is_none() { + return Err("w=port and w=own name the port w is fed on with port".to_string()); } Ok(params) } @@ -130,6 +133,10 @@ struct State { w: Option, width: u32, height: u32, + /// The output's size: `v`'s, or `w`'s with `w=own`. + out: (u32, u32), + /// `w=own`: a tick with no frame of `w` writes a transparent frame. + own: bool, size: u32, yuv: bool, on_gpu: bool, @@ -260,7 +267,14 @@ fn draw_on_gpu( .ok_or("the device is opened at init")?; let handed = planes(id, index)?; let source = handed.first().ok_or("an rgba frame has one plane")?; - let (width, height) = (state.width, state.height); + let (width, height) = state.out; + if (source.width(), source.height()) != (width, height) { + return Err(format!( + "frame {index} of stream {id} is a {}x{} texture, and the output is {width}x{height}", + source.width(), + source.height() + )); + } let extent = |w: u32, h: u32| GpuExtent3D { width: w, height: Some(h), @@ -312,6 +326,32 @@ fn draw_on_gpu( emit("out", pts, duration, out) } +/// Hands the host a transparent frame on `out`, on the GPU, at `pts`. +fn blank_on_gpu(state: &State, pts: i64, duration: Option) -> Result<(), String> { + let device = state + .device + .as_ref() + .ok_or("the device is opened at init")?; + let (width, height) = state.out; + let out = device.create_texture(&GpuTextureDescriptor { + size: GpuExtent3D { + width, + height: Some(height), + depth_or_array_layers: None, + }, + mip_level_count: None, + sample_count: None, + dimension: None, + format: GpuTextureFormat::Rgba8unorm, + usage: GpuTextureUsage::copy_src() + | GpuTextureUsage::copy_dst() + | GpuTextureUsage::texture_binding(), + view_formats: None, + label: None, + }); + emit("out", pts, duration, out) +} + struct Node; impl Guest for Node { @@ -351,11 +391,12 @@ impl Guest for Node { }, schema: None, }; + let own = params.w.as_deref() == Some("own"); let out = OutputPort { name: "out".to_string(), kind: PortKind::Video, format: Some(OutputFormat::Like(LikeInput { - port: "v".to_string(), + port: if own { "w" } else { "v" }.to_string(), pixel_format: None, sample_format: None, })), @@ -369,8 +410,8 @@ impl Guest for Node { let mut w = inputs[0].clone(); w.name = "w".to_string(); w.accepts.pixel_formats = vec!["rgba".to_string()]; - w.accepts.like = Some("v".to_string()); - if bound == "port" { + w.accepts.like = (!own).then(|| "v".to_string()); + if bound != "pad" { w.required = false; w.pairing = Pairing::Hold(Hold { anchor: Anchor::FirstFrame, @@ -408,12 +449,17 @@ impl Guest for Node { "a texture handed back is rgba; a yuv420p input takes out=host".to_string(), ); } + let own = params.w.as_deref() == Some("own"); let w = bound.iter().find(|b| b.port == "w"); + let mut out = (format.width, format.height); if let Some(w) = w { let Some(OutputFormat::Video(told)) = &w.format else { return Err("w is video".to_string()); }; - if yuv || (told.width, told.height) != (format.width, format.height) { + if own { + out = (told.width, told.height); + } + if yuv || (!own && (told.width, told.height) != (format.width, format.height)) { return Err(format!( "w is told {}x{} {}, and follows v's {}x{} rgba", told.width, told.height, told.pix_fmt, format.width, format.height @@ -438,6 +484,8 @@ impl Guest for Node { w: w.map(|w| w.id), width: format.width, height: format.height, + out, + own, size: params.size, yuv, on_gpu, @@ -463,6 +511,22 @@ impl Guest for Node { Some((w, held)) if !held.is_empty() => (w, held[held.len() - 1].index), _ => (state.v, frame.index), }; + let (width, height) = state.out; + if state.own && id == state.v { + if state.out_on_gpu { + blank_on_gpu(state, frame.pts, frame.duration)?; + } else { + items.push(Emission { + port: "out".to_string(), + payload: Payload::Frame(RawFrame { + pts: frame.pts, + duration: frame.duration, + data: vec![0; (width * height * 4) as usize], + }), + }); + } + continue; + } if state.out_on_gpu { draw_on_gpu(state, frame.pts, frame.duration, id, index)?; continue; @@ -475,9 +539,17 @@ impl Guest for Node { tick.fetch(id, index) }; if state.yuv { - draw_yuv(&mut pixels, state.width, state.height, state.size); + draw_yuv(&mut pixels, width, height, state.size); } else { - draw(&mut pixels, state.width, state.height, state.size); + if pixels.len() != (width * height * 4) as usize { + return Err(format!( + "frame {index} of stream {id} is {} bytes, and an rgba frame of the \ + output's {width}x{height} is {}", + pixels.len(), + width * height * 4 + )); + } + draw(&mut pixels, width, height, state.size); } items.push(Emission { port: "out".to_string(), diff --git a/sidecar/runtime/src/node.rs b/sidecar/runtime/src/node.rs index 864a5383..ff60250a 100644 --- a/sidecar/runtime/src/node.rs +++ b/sidecar/runtime/src/node.rs @@ -96,10 +96,14 @@ impl Interval { } /// What the compiler knows of one stream before the run: the frame rate, -/// or the sample rate over 1. +/// or the sample rate over 1; and, for a picture fed on a port, the size +/// the compile declared for it, which the host conforms the feed to. The +/// size is the host's alone: a module's `shape` and `init` are handed the +/// rate. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] pub struct StreamHint { pub rate: Option, + pub size: Option<(u32, u32)>, } /// One input a call binds, as `shape` is told it: one hint per stream, in From 13af234b347d32f7190ec8b68221a811222ff1d2 Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Mon, 19 Oct 2026 21:34:07 -0700 Subject: [PATCH 2/2] cli: a lateral's declared picture size is its stream's hint in -bound A run-time lateral whose body selects an input's own first picture, with a shape giving width and height, now passes that size on to the node input it feeds on a port, so the sidecar conforms the feed to it. A body that declares no size passes none, and the feed is conformed as before. Co-Authored-By: Claude Fable 5.1 --- cli/ffrwd/functions.py | 72 ++++++++++++++++++++++++++++++++++++++ cli/ffrwd/lower.py | 24 +++++++++++-- cli/ffrwd/shapes.py | 26 +++++++++----- cli/tests/test_laterals.py | 35 ++++++++++++++++++ docs/dialect.md | 8 ++++- docs/known_gaps.md | 8 +++++ 6 files changed, 161 insertions(+), 12 deletions(-) diff --git a/cli/ffrwd/functions.py b/cli/ffrwd/functions.py index bbfc4077..371a0791 100644 --- a/cli/ffrwd/functions.py +++ b/cli/ffrwd/functions.py @@ -126,6 +126,7 @@ is_annotation_argument, kwarg_name, parse, + subscript_index, ) from .project import ( RESERVED_NAMESPACES, @@ -201,6 +202,9 @@ 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)``. """ function: str @@ -212,6 +216,7 @@ class RuntimeLateral: columns: tuple[Parameter, ...] written: Mapping[str, str] needs: tuple[str, ...] + sizes: tuple[tuple[str, int, int], ...] = () # Names a definition may not claim: the dialect's own FROM item, the two @@ -3689,6 +3694,72 @@ 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. + """ + body = function.body + items: list[exp.Expr] = [] + found = body.args.get("from_") + if isinstance(found, exp.From): + items.append(found.this) + items += [join.this for join in body.args.get("joins") or []] + shapes: dict[str, tuple[int, int]] = {} + for item in items: + call = item.this if isinstance(item, exp.Table) else None + alias = item.args.get("alias") if isinstance(item, exp.Table) else None + if not isinstance(call, exp.Anonymous) or str(call.this).lower() != "input": + continue + if not isinstance(alias, exp.TableAlias) or alias.this is None: + continue + for argument in call.expressions: + if not isinstance(argument, exp.Kwarg) or kwarg_name(argument) != "shape": + continue + 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 + ): + shapes[_ident_name(alias.this)] = (width, height) + 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)) + return tuple(sizes) + + def _check_runs_per_message(function: _Function, anchor: exp.Expr) -> None: """A run-time lateral's declaration: the streams a feeder takes out, one value per parameter after the data stream, and a body that reads no @@ -5920,6 +5991,7 @@ def _expand_runtime_lateral( columns=function.columns or (), written=written, needs=self._needs(function), + sizes=_declared_sizes(function), ) line, col = _pos(item) columns: dict[str, exp.Expr] = {} diff --git a/cli/ffrwd/lower.py b/cli/ffrwd/lower.py index 10002183..ed60c20d 100644 --- a/cli/ffrwd/lower.py +++ b/cli/ffrwd/lower.py @@ -16525,12 +16525,16 @@ def _node_instance( and all(_lateral_parts(stream.ref) is not None for stream in value.streams) } # A lateral's port is bound, to the host's own stream, so the shape - # is asked with it; what it binds in the network is nothing. + # is asked with it; what it binds in the network is nothing. Its + # picture's hint carries the size the lateral's body declares, which + # the host conforms the feed to. bound = [ Binding( param.name, tuple( - StreamHint(None if param.name in fed else self._stream_rate(stream.ref)) + StreamHint(None, self._lateral_size(stream.ref)) + if param.name in fed + else StreamHint(self._stream_rate(stream.ref)) for stream in streams[param.name].streams ), ) @@ -16695,6 +16699,22 @@ def _node_instance( self._node_refs[key] = instance return instance + def _lateral_size(self, ref: FrameRef) -> tuple[int, int] | None: + """The picture size the body of the run-time lateral making `ref` + declares for it, where it declares one.""" + parts = _lateral_parts(ref) + if parts is None: + return None + key, column = parts + return next( + ( + (width, height) + for named, width, height in self._laterals[key].declared.sizes + if named == column + ), + None, + ) + def _stream_rate(self, ref: FrameRef) -> Fraction | None: """The rate a node's shape is told the stream `ref` runs at.""" return stream_rate(self.graph, self.probes, self._made_shape, ref) diff --git a/cli/ffrwd/shapes.py b/cli/ffrwd/shapes.py index 6dd3101e..def23bb0 100644 --- a/cli/ffrwd/shapes.py +++ b/cli/ffrwd/shapes.py @@ -231,9 +231,11 @@ def to_dict(self) -> dict[str, object]: class StreamHint: """What the compiler knows of one bound stream before the run: a video stream's frame rate or an audio stream's sample rate, None where nothing - settles it.""" + settles it; and, for a picture fed on a port, the width and height its + source declares, which the sidecar conforms the feed to.""" rate: Fraction | None = None + size: tuple[int, int] | None = None @dataclass(frozen=True) @@ -251,14 +253,7 @@ def bound_json(bound: Sequence[Binding]) -> str: [ { "input": binding.input, - "streams": [ - { - "rate": None - if hint.rate is None - else {"num": hint.rate.numerator, "den": hint.rate.denominator} - } - for hint in binding.streams - ], + "streams": [_hint_json(hint) for hint in binding.streams], } for binding in bound ], @@ -266,6 +261,19 @@ def bound_json(bound: Sequence[Binding]) -> str: ) +def _hint_json(hint: StreamHint) -> dict[str, object]: + """One stream's hint as ``--bound`` takes it: its rate, and its size + where it has one.""" + found: dict[str, object] = { + "rate": None + if hint.rate is None + else {"num": hint.rate.numerator, "den": hint.rate.denominator} + } + if hint.size is not None: + found["width"], found["height"] = hint.size + return found + + def _reject(message: str, hint: str = _UNKNOWN_HINT) -> FfrwdError: return FfrwdError(ErrorCode.UNSUPPORTED_SQL, message, hint=hint) diff --git a/cli/tests/test_laterals.py b/cli/tests/test_laterals.py index 28bc9f9d..73df3080 100644 --- a/cli/tests/test_laterals.py +++ b/cli/tests/test_laterals.py @@ -1010,6 +1010,41 @@ def test_a_port_param_taking_one_or_many_is_handed_a_list_for_two_laterals() -> assert _compose_node(plan).args["port"] == ports +_CAM = """CREATE FUNCTION cam(launch data_stream, url text) +RETURNS TABLE(video video_stream) AS $$ + SELECT c.video[1] AS video + FROM input(url, format => 'mpegts', shape => STRUCT(1920 AS width, 1080 AS height, 60 AS fps)) c +$$ LANGUAGE sql;""" + + +def test_a_laterals_declared_picture_size_is_its_streams_hint_in_port_order() -> None: + """The cam's body reads an input whose shape declares 1920x1080, so its + stream's hint carries that size; the ad's body scales to what each + message binds, so its hint has none. The hints are in the order of the + ports the module is handed, which is how the host pairs them.""" + query = _TWO_ADS.replace("ARRAY[ad.video, lbar.video]", "ARRAY[m.video, ad.video]").replace( + "LATERAL play(prog.d) ad, LATERAL play(prog.d) lbar", + "LATERAL cam(prog.d) m, LATERAL play(prog.d) ad", + ) + modules = _compose_modules(_ONE_OR_MANY_PORTS) + plan = compile_all( + _CAM + "\n" + _declared(query).replace("COPY (", _COMPOSE + "\nCOPY (", 1), + describe=lambda path: modules[path], + shape=_compose_shape, + ).plan + assert plan is not None + compose = _compose_node(plan) + by_function = {lateral.function: lateral.connections[0].port for lateral in plan.laterals} + assert compose.args["port"] == [by_function["cam"], by_function["play"]] + assert json.loads(compose.bound) == [ + {"input": "v", "streams": [{"rate": {"num": 25, "den": 1}}]}, + { + "input": "inputs", + "streams": [{"rate": None, "width": 1920, "height": 1080}, {"rate": None}], + }, + ] + + 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 19d14f8a..d3a9b919 100644 --- a/docs/dialect.md +++ b/docs/dialect.md @@ -701,7 +701,13 @@ dest := 'path' | STDOUT | ( value-expression ) | sink(value, ...) conforms it to). An input's is its probe's; a node's picture runs at 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. Each call carries the + 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 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 23a8e1fe..4b17ecd7 100644 --- a/docs/known_gaps.md +++ b/docs/known_gaps.md @@ -73,6 +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. - **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