From fbdb7e301e6577142eec1b8071f8e115103237fd Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Sat, 17 Oct 2026 17:15:49 -0700 Subject: [PATCH 1/3] sidecar: a hold group is one per connection The assembler built one hold group per group name, so a node's many hold inputs on one group shared one source and one feed: with compose's picture inputs and a parallel sound input on one group, a second lateral connecting closed the first lateral's feed. A group is now keyed by its name and its connection: the port for a port-fed stream, and the stream's place among its input's streams for one bound in the plan. Each connection has its own source, feed and offset, a picture and a sound on one connection still share theirs, and a data input on the group follows the group of the connection its rows arrive on. shape-group is the test module: many pictures and many sounds held on one group, with a row per change in each stream's feed. The live test feeds it on two ports, drops and rejoins the second, and checks that the first stays one feed, each connection's sound shares its picture's offset, and the first connection's rows land with its own offset. Co-Authored-By: Claude Fable 5.1 --- sidecar/NODE-CLI.md | 17 ++- sidecar/ffrwd-wasm/src/hold.rs | 4 +- sidecar/ffrwd-wasm/src/node_graph.rs | 4 +- sidecar/ffrwd-wasm/src/tick.rs | 126 ++++++++++++++++++--- sidecar/ffrwd-wasm/tests/live.rs | 138 ++++++++++++++++++++++- sidecar/modules/Cargo.lock | 9 ++ sidecar/modules/Cargo.toml | 2 + sidecar/modules/shape-group/Cargo.toml | 13 +++ sidecar/modules/shape-group/src/lib.rs | 148 +++++++++++++++++++++++++ 9 files changed, 435 insertions(+), 26 deletions(-) create mode 100644 sidecar/modules/shape-group/Cargo.toml create mode 100644 sidecar/modules/shape-group/src/lib.rs diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md index 9eb561a..4e13910 100644 --- a/sidecar/NODE-CLI.md +++ b/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,7 +289,7 @@ 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 @@ -293,7 +297,8 @@ 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, diff --git a/sidecar/ffrwd-wasm/src/hold.rs b/sidecar/ffrwd-wasm/src/hold.rs index 7a1fc76..242eb42 100644 --- a/sidecar/ffrwd-wasm/src/hold.rs +++ b/sidecar/ffrwd-wasm/src/hold.rs @@ -1,7 +1,7 @@ //! Hold pairing: a source's frames paired by time onto a node's clock. //! -//! One [`Group`] serves one hold input, or every hold input of one `group` -//! (a feeder's picture and sound on one connection). Frames arrive as the +//! One [`Group`] serves one hold input, or the hold inputs of one `group` +//! on one connection (a feeder's picture and sound). Frames arrive as the //! source delivers them and queue per member; each tick, a video member //! hands the newest frame at or before the tick, repeating the last one //! while the source is behind and skipping forward when it catches up, and diff --git a/sidecar/ffrwd-wasm/src/node_graph.rs b/sidecar/ffrwd-wasm/src/node_graph.rs index 4345849..198a650 100644 --- a/sidecar/ffrwd-wasm/src/node_graph.rs +++ b/sidecar/ffrwd-wasm/src/node_graph.rs @@ -1619,9 +1619,9 @@ fn open_module( &schema, defs, )?; - let port_fed: Vec = feeds + let port_fed: Vec<(u32, u16)> = feeds .iter() - .flat_map(|f| f.members.iter().map(|m| m.id)) + .flat_map(|f| f.members.iter().map(|m| (m.id, f.port))) .collect(); let latched: Vec = call .outputs diff --git a/sidecar/ffrwd-wasm/src/tick.rs b/sidecar/ffrwd-wasm/src/tick.rs index e4f8465..f3b668b 100644 --- a/sidecar/ffrwd-wasm/src/tick.rs +++ b/sidecar/ffrwd-wasm/src/tick.rs @@ -591,14 +591,28 @@ pub struct Assembler { report: Report, } +/// A hold group's name and connection; none for a hold input of no group, +/// which is a group of its own. +type GroupKey = Option<(String, Connection)>; + +/// The connection a stream of a hold group arrives on: the port of a port +/// feed, or for a stream bound in the plan its place among its input's +/// streams. A group is one per name and connection. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Connection { + Port(u16), + Bound(usize), +} + impl Assembler { /// The ticks of `name`, a node of `shape` bound to `bound`. `port_fed` - /// names the hold inputs a port feed serves, which never hold a tick. + /// names the streams a port feed serves, each with its port; their hold + /// inputs never hold a tick. pub fn new( shape: &NodeShape, bound: &[BoundStream], name: &str, - port_fed: &[u32], + port_fed: &[(u32, u16)], ) -> Result { let clock_base = match &shape.clock { Clock::Input(name) => bound.iter().find(|b| &b.port == name).map(|b| b.time_base), @@ -610,9 +624,23 @@ impl Assembler { }; let mut inputs = Vec::with_capacity(bound.len()); let mut holds: Vec = Vec::new(); - let mut on_groups: Vec<(usize, String, Member, bool)> = Vec::new(); - let mut grouped: Vec<(Option, Hold, Vec, bool)> = Vec::new(); - for stream in bound { + let mut on_groups: Vec<(usize, String, Connection, Member, bool)> = Vec::new(); + let mut grouped: Vec<(GroupKey, Hold, Vec, bool)> = Vec::new(); + for (index, stream) in bound.iter().enumerate() { + let fed_on = port_fed + .iter() + .find(|(id, _)| *id == stream.id) + .map(|(_, port)| *port); + let connection = match fed_on { + Some(port) => Connection::Port(port), + None => Connection::Bound( + bound[..index] + .iter() + .filter(|b| b.port == stream.port) + .count(), + ), + }; + let fed = fed_on.is_some(); let port = shape .input(&stream.port) .ok_or_else(|| anyhow!("no input port '{}'", stream.port))?; @@ -637,8 +665,7 @@ impl Assembler { audio, ahead: 0.0, }; - let fed = port_fed.contains(&stream.id); - let key = hold.group.clone(); + let key = hold.group.clone().map(|g| (g, connection)); let group = match key .as_ref() .and_then(|k| grouped.iter().position(|(g, ..)| g.as_ref() == Some(k))) @@ -661,6 +688,7 @@ impl Assembler { on_groups.push(( inputs.len(), group.clone(), + connection, Member { name: port.name.clone(), kind: port.kind, @@ -669,7 +697,7 @@ impl Assembler { audio: None, ahead: *ahead, }, - port_fed.contains(&stream.id), + fed, )); None } @@ -700,11 +728,19 @@ impl Assembler { restamp, }); } - for (index, key, member, fed) in on_groups { - let Some(group) = grouped - .iter() - .position(|(g, ..)| g.as_deref() == Some(key.as_str())) - else { + for (index, key, connection, member, fed) in on_groups { + let named = |g: &GroupKey| g.as_ref().is_some_and(|(g, _)| *g == key); + let Some(group) = grouped.iter().position(|(g, ..)| { + g.as_ref() + .is_some_and(|(g, c)| *g == key && *c == connection) + }) else { + if grouped.iter().any(|(g, ..)| named(g)) { + bail!( + "input '{}' arrives on group '{key}', and none of that group's hold \ + inputs arrives on its connection", + member.name + ); + } bail!( "input '{}' arrives on group '{key}', and this call binds none of that \ group's hold inputs", @@ -1961,7 +1997,7 @@ mod tests { bound("feed_audio", 2, KHZ48, audio()), ], "m", - &[1, 2], + &[(1, 9000), (2, 9000)], ) .expect("assembler"); let source = |base| crate::hold::SourceInfo { @@ -1995,6 +2031,64 @@ mod tests { assert_eq!(sound.info.as_ref().map(|(_, base)| *base), Some(KHZ48)); } + #[test] + fn a_group_bound_in_the_plan_pairs_its_many_inputs_stream_by_stream() { + let hold = Pairing::Hold(ffrwd_wasm_runtime::node::Hold { + anchor: ffrwd_wasm_runtime::node::Anchor::FirstFrame, + lead: 0.0, + linger: None, + timeout: None, + group: Some("g".to_string()), + port_param: None, + }); + let many = |name: &str, kind: PortKind| InputPort { + many: true, + ..port(name, kind, hold.clone(), RowsUse::Ignore) + }; + let s = shape( + vec![ + port("v", PortKind::Video, Pairing::Lockstep, RowsUse::Ignore), + many("feed", PortKind::Video), + many("feed_audio", PortKind::Audio), + ], + Clock::Input("v".into()), + ); + let tb = TimeBase { num: 1, den: 30 }; + let mut a = Assembler::new( + &s, + &[ + bound("v", 0, tb, video()), + bound("feed", 1, tb, video()), + bound("feed", 2, tb, video()), + bound("feed_audio", 3, KHZ48, audio()), + bound("feed_audio", 4, KHZ48, audio()), + ], + "m", + &[], + ) + .expect("assembler"); + for k in 0..3 { + a.arrive(0, frame(k, &[])).expect("video"); + a.arrive(1, frame(k, &[])).expect("feed"); + a.arrive(2, frame(100 + k, &[])).expect("feed"); + a.arrive(3, samples(k * 1600, 1600)).expect("sound"); + a.arrive(4, samples((100 + k) * 1600, 1600)).expect("sound"); + } + let ticks = drain(&mut a); + let start = |id: usize| { + let feed = ticks[0].streams[id].feed.as_ref().expect("a feed"); + (feed.start.at, feed.start.first_pts) + }; + assert_eq!(start(1), (0, 0)); + assert_eq!( + start(2), + (0, 100), + "the second picture is a feed of its own, anchored on its own first frame" + ); + assert_eq!(start(3), (0, 0), "the first sound beside the first picture"); + assert_eq!(start(4), (0, 160_000), "the second sound beside the second"); + } + #[test] fn a_data_clock_ticks_once_per_pts_and_a_packets_clock_on_its_running_dts() { let s = shape( @@ -2162,8 +2256,8 @@ mod tests { Clock::Rate(Rational { num: 30, den: 1 }), ); let tb = TimeBase { num: 1, den: 30 }; - let mut a = - Assembler::new(&s, &[bound("feed", 4, tb, video())], "m", &[4]).expect("assembler"); + let mut a = Assembler::new(&s, &[bound("feed", 4, tb, video())], "m", &[(4, 9000)]) + .expect("assembler"); a.arrive(4, frame(0, &[])).expect("a frame"); let mut heard: Vec> = vec![Vec::new(); 3]; for _ in 0..10 { diff --git a/sidecar/ffrwd-wasm/tests/live.rs b/sidecar/ffrwd-wasm/tests/live.rs index 374c5ff..95f8e20 100644 --- a/sidecar/ffrwd-wasm/tests/live.rs +++ b/sidecar/ffrwd-wasm/tests/live.rs @@ -32,6 +32,7 @@ const MODULES: &[&str] = &[ "shape-window", "shape-hold", "shape-probe", + "shape-group", ]; const RATE: u32 = 48_000; @@ -676,7 +677,8 @@ fn start_padded( if chain.contains("[feeds=f]") || chain.contains("[spots=f]") { command.args(["-map", "[f]", "-f", "ndjson", &feeds]); } - if chain.contains("[clock=c]") { + // A second rows output of the node rides the same file as `clock`. + if chain.contains("[clock=c]") || chain.contains("[cues=c]") { command.args(["-map", "[c]", "-f", "ndjson", &clock]); } command.stdin(Stdio::piped()); @@ -1637,3 +1639,137 @@ fn rows_a_feeder_writes_beside_its_picture_and_sound_land_with_the_groups_offset out.stderr ); } + +/// The feeds `shape_group` said one stream of `input` had, in order: each +/// start as (at, first_pts), or None where its feed went off. +fn group_feeds(out: &Output, input: &str, k: u64) -> Vec<(i64, Option<(i64, i64)>)> { + out.feeds + .iter() + .filter(|r| r["member"] == input && r["k"].as_u64() == Some(k)) + .map(|r| { + let start = ( + r["at_pts"].as_i64().unwrap(), + r["first_pts"].as_i64().unwrap(), + ); + ( + r["pts"].as_i64().unwrap(), + r["live"].as_bool().unwrap().then_some(start), + ) + }) + .collect() +} + +#[test] +fn each_connection_of_a_many_hold_group_is_a_feed_of_its_own_with_its_sound_beside_its_picture() { + let _serial = serial(); + let program = Program::new(30, 9.0); + let (first, second) = (free_port(), free_port()); + let dir = scratch("group-many-params"); + let params = dir.join("params.json"); + fs::write(¶ms, format!(r#"{{"ports":[{first},{second}]}}"#)).expect("write the params"); + let from = format!("shape_group={}", params.display()); + let (mut run, _) = start_padded( + "group-many", + &program, + "1", + &["shape_group"], + "[c=0:v][a=0:a]shape_group[c=o][feeds=f][cues=c]", + &["-params-from", &from], + ); + wait_for_port(&mut run, first); + wait_for_port(&mut run, second); + // The first connection stays on throughout; the second comes, goes and + // comes back on another origin while it is. + let mut main = Feeder::new(30, 6.0); + main.start = 10.0; + main.rows = vec![ + (10_200, r#"{"text":"a"}"#.to_string()), + (12_000, r#"{"text":"b"}"#.to_string()), + ]; + let mut again = Feeder::new(30, 1.5); + again.start = 20.0; + let fed = [ + feed(first, main, Duration::from_millis(500)), + feed(second, Feeder::new(30, 1.5), Duration::from_millis(1500)), + feed(second, again, Duration::from_millis(4500)), + ]; + let out = finish(run); + for one in fed { + one.join().expect("a feeder"); + } + let _ = fs::remove_dir_all(&dir); + + let main_feeds = group_feeds(&out, "v", 0); + let second_feeds = group_feeds(&out, "v", 1); + assert!( + matches!( + main_feeds.as_slice(), + [(_, Some(_)), (_, None)] | [(_, Some(_))] + ), + "the first connection is one feed, on until it ends: {main_feeds:?}\n{}", + out.stderr + ); + let (main_at, main_first) = main_feeds[0].1.unwrap(); + assert_eq!(main_first, 300, "its own origin, ten seconds in"); + let starts: Vec = second_feeds + .iter() + .filter_map(|(_, start)| start.map(|s| s.1)) + .collect(); + assert_eq!( + starts, + vec![0, 600], + "the second connection is anchored again when it rejoins: {second_feeds:?}\n{}", + out.stderr + ); + let rejoined = second_feeds + .iter() + .find(|(_, start)| start.is_some_and(|s| s.1 == 600)) + .unwrap() + .0; + assert!( + main_feeds.get(1).is_none_or(|(off, _)| *off > rejoined), + "the first is still on when the second rejoins: {main_feeds:?} {second_feeds:?}" + ); + for k in 0..2 { + let picture = group_feeds(&out, "v", k); + let sound = group_feeds(&out, "s", k); + let on = |feeds: &[(i64, Option<(i64, i64)>)]| -> Vec<(i64, i64)> { + feeds.iter().filter_map(|(_, start)| *start).collect() + }; + let (picture, sound) = (on(&picture), on(&sound)); + assert_eq!( + picture.iter().map(|s| s.0).collect::>(), + sound.iter().map(|s| s.0).collect::>(), + "connection {k}'s picture and sound share each feed's offset\n{}", + out.stderr + ); + for (p, s) in picture.iter().zip(&sound) { + let apart = p.1 as f64 / 30.0 - s.1 as f64 / f64::from(RATE); + assert!( + apart.abs() <= 1.0 / 30.0, + "connection {k}'s sound starts where its picture does: {p:?} {s:?}" + ); + } + } + let cues: Vec<(i64, i64, String)> = out + .clock + .iter() + .map(|r| { + let text: Value = serde_json::from_str(r["text"].as_str().unwrap()).unwrap(); + ( + r["pts"].as_i64().unwrap(), + r["cue_pts"].as_i64().unwrap(), + text["text"].as_str().unwrap().to_string(), + ) + }) + .collect(); + assert_eq!( + cues, + vec![ + (main_at + 6, main_at + 6, "a".to_string()), + (main_at + 60, main_at + 60, "b".to_string()), + ], + "the first connection's rows land with its own offset, the second on or not:\n{}", + out.stderr + ); +} diff --git a/sidecar/modules/Cargo.lock b/sidecar/modules/Cargo.lock index 19e7577..05ac294 100644 --- a/sidecar/modules/Cargo.lock +++ b/sidecar/modules/Cargo.lock @@ -1955,6 +1955,15 @@ dependencies = [ "wit-bindgen 0.57.1", ] +[[package]] +name = "shape-group" +version = "0.1.0" +dependencies = [ + "ffrwd-node", + "serde", + "serde_json", +] + [[package]] name = "shape-hold" version = "0.1.0" diff --git a/sidecar/modules/Cargo.toml b/sidecar/modules/Cargo.toml index f5db833..328d99f 100644 --- a/sidecar/modules/Cargo.toml +++ b/sidecar/modules/Cargo.toml @@ -100,6 +100,8 @@ members = [ # What the live switch did, as a node: the host's hold pairing, feeds by # port and both anchors, with none of the switch's own code. "shape-switch", + # Many pictures and many sounds on one hold group, a connection each. + "shape-group", ] resolver = "2" diff --git a/sidecar/modules/shape-group/Cargo.toml b/sidecar/modules/shape-group/Cargo.toml new file mode 100644 index 0000000..6d60484 --- /dev/null +++ b/sidecar/modules/shape-group/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "shape-group" +version = "0.1.0" +edition = "2021" + +[lib] +name = "shape_group" +crate-type = ["cdylib"] + +[dependencies] +ffrwd-node = { git = "https://github.com/imbcmdth/ffrwd-node", tag = "v0.2.0" } +serde = { version = "1.0.229", features = ["derive"] } +serde_json = "1.0.151" diff --git a/sidecar/modules/shape-group/src/lib.rs b/sidecar/modules/shape-group/src/lib.rs new file mode 100644 index 0000000..b404052 --- /dev/null +++ b/sidecar/modules/shape-group/src/lib.rs @@ -0,0 +1,148 @@ +//! Many pictures and many sounds held on one group, as a compositor takes +//! each lateral's picture and sound: `v` and `s` take a stream per port +//! `ports` lists, the picture and the sound on one port arriving on one +//! connection. `cues` takes rows on the group's first connection. +//! +//! It passes the clock `c` through. `feeds` has a row whenever what the +//! host says of one stream's feed changes: its input (`member`), its place +//! among the input's streams, whether a feed is live and, while one is, +//! its start. `cues` has a row per message `cues` is handed, on the tick it is. + +use ffrwd_node::{Anchor, Bound, Init, Input, Node, Out, Output, Result, Shape, Tick}; +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize)] +struct Params { + lead: f64, + timeout: f64, +} + +#[derive(Serialize)] +struct FeedRow<'a> { + pts: i64, + member: &'a str, + k: usize, + live: bool, + at_pts: i64, + first_pts: i64, +} + +#[derive(Serialize)] +struct CueRow { + pts: i64, + cue_pts: i64, + text: String, +} + +/// One stream of `v` or `s`: its input, its place, its id and the start +/// said last. +struct Held { + input: &'static str, + k: usize, + id: u32, + said: Option<(i64, i64)>, +} + +struct Group { + c: u32, + held: Vec, + cues: Option, +} + +impl Node for Group { + const NAME: &'static str = "shape_group"; + const VERSION: &'static str = "0.1.0"; + const PARAMS_SCHEMA: &'static str = r#"{"type":"object","properties":{"ports":{"type":"array","items":{"type":"integer","minimum":1,"maximum":65535}},"lead":{"type":"number","minimum":0,"default":0.3},"timeout":{"type":"number","exclusiveMinimum":0,"default":1}},"additionalProperties":false}"#; + type Params = Params; + + fn shape(params: &Params, _: &Bound) -> Result { + let held = |input: Input| { + input + .optional() + .many() + .hold() + .anchor(Anchor::FirstFrame) + .lead(params.lead) + .timeout(params.timeout) + .group("lateral") + .port_param("ports") + }; + Ok(Shape::new() + .input(Input::video("c").clock().pixel_formats(&["rgba"])) + .input(Input::audio("a").optional().sample_formats(&["f32"])) + .input(held(Input::video("v")).like("c").pixel_formats(&["rgba"])) + .input(held(Input::audio("s")).like("a").sample_formats(&["f32"])) + .input(Input::rows("cues").optional().group("lateral").per_frame()) + .output(Output::like("c")) + .output(Output::rows("feeds").schema_json( + r#"{"type":"object","properties":{"pts":{"type":"integer"},"member":{"type":"string"},"k":{"type":"integer"},"live":{"type":"boolean"},"at_pts":{"type":"integer"},"first_pts":{"type":"integer"}}}"#, + )) + .output(Output::rows("cues").schema_json( + r#"{"type":"object","properties":{"pts":{"type":"integer"},"cue_pts":{"type":"integer"},"text":{"type":"string"}}}"#, + )) + .one_to_one()) + } + + fn init(_: Params, init: &Init) -> Result { + let mut held = Vec::new(); + for input in ["v", "s"] { + for (k, stream) in init.streams(input).iter().enumerate() { + held.push(Held { + input, + k, + id: stream.id, + said: None, + }); + } + } + Ok(Group { + c: init.stream("c")?.id, + held, + cues: init.optional("cues").map(|b| b.id), + }) + } + + fn process(&mut self, tick: &Tick, out: &mut Out) -> Result<()> { + let pts = tick.pts(); + let Some(frame) = tick.frame(self.c) else { + return Ok(()); + }; + out.pass("c", self.c, &frame)?; + for held in &mut self.held { + let start = tick + .feed(held.id) + .map(|feed| (feed.start.at, feed.start.first_pts)); + if start == held.said { + continue; + } + held.said = start; + let (at, first_pts) = start.unwrap_or_default(); + out.row( + "feeds", + pts, + &FeedRow { + pts, + member: held.input, + k: held.k, + live: start.is_some(), + at_pts: at, + first_pts, + }, + )?; + } + for message in self.cues.map(|id| tick.messages(id)).unwrap_or_default() { + out.row( + "cues", + pts, + &CueRow { + pts, + cue_pts: message.pts, + text: String::from_utf8_lossy(&message.data).into_owned(), + }, + )?; + } + Ok(()) + } +} + +ffrwd_node::export!(Group); From 3b2b8d70a8786752403865d597b68da18fa677c5 Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Sat, 17 Oct 2026 17:15:50 -0700 Subject: [PATCH 2/3] registry: load once when two compiles read it at once Two run-time laterals reading one data stream compile their instances on threads of their own at the same moment. The registry marked itself loaded before it had read ffmpeg's filters, so the second thread found none and refused its instance with "unknown function setpts()". The load now holds a lock, and a second reader waits for the first. Co-Authored-By: Claude Fable 5.1 --- cli/ffrwd/registry.py | 32 +++++++++++++++++++------------- cli/tests/test_registry.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 13 deletions(-) diff --git a/cli/ffrwd/registry.py b/cli/ffrwd/registry.py index adf4f92..8b24557 100644 --- a/cli/ffrwd/registry.py +++ b/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: diff --git a/cli/tests/test_registry.py b/cli/tests/test_registry.py index 832d58f..32caecd 100644 --- a/cli/tests/test_registry.py +++ b/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 From e924958bceed454fc3c80f7a50d4c9b8814ea9d2 Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Sat, 17 Oct 2026 17:15:51 -0700 Subject: [PATCH 3/3] compiler: a lateral may bring some of one call's held many inputs compose(k, ARRAY[m.video, c.video], audio => ARRAY[m.audio]) was refused: m's connection reached compose through `inputs` and `audio` and c's through `inputs` alone, and the group check wants every connection to reach the same calls. A group whose every connection reaches one and the same call is that call's many inputs, so a connection may bring some of them and not others. Groups whose connections reach different calls, the switch pair's case, are refused as before. Every many input of the group now lists the group's connections in its one port param, in the order first written, where the last input written used to overwrite the list. The exec test runs two laterals into shape-group through ffrwd run: m brings its picture and its sound, c its picture alone a second later; each is live beside the other on its own origin, and m's sound shares its picture's offset. Co-Authored-By: Claude Fable 5.1 --- cli/ffrwd/lower.py | 21 +++++++- cli/tests/exec/test_exec_laterals.py | 73 +++++++++++++++++++++++++ cli/tests/test_laterals.py | 80 ++++++++++++++++++++++++++++ docs/dialect.md | 6 +++ 4 files changed, 178 insertions(+), 2 deletions(-) diff --git a/cli/ffrwd/lower.py b/cli/ffrwd/lower.py index 143c3f6..1000218 100644 --- a/cli/ffrwd/lower.py +++ b/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 " diff --git a/cli/tests/exec/test_exec_laterals.py b/cli/tests/exec/test_exec_laterals.py index 7f12b1d..4a15771 100644 --- a/cli/tests/exec/test_exec_laterals.py +++ b/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}" diff --git a/cli/tests/test_laterals.py b/cli/tests/test_laterals.py index 57a1a12..28bc9f9 100644 --- a/cli/tests/test_laterals.py +++ b/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" diff --git a/docs/dialect.md b/docs/dialect.md index 95fa201..19d14f8 100644 --- a/docs/dialect.md +++ b/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( , ...)`, one field per output the module makes, read off the call