From 0a247ec1c42c9a03fce6b78825796f679ea2392c Mon Sep 17 00:00:00 2001 From: Jon-Carlos Rivera Date: Thu, 15 Oct 2026 19:38:35 -0700 Subject: [PATCH] sidecar: hand a state input's earlier rows at the times they arrived An instance of a frame-parallel node with a state input is handed the rows of the ticks it did not process. Each tick's rows were recorded at the tick's pts, in the clock's time base, and a node reads them in the rows stream's own time base, as it reads the messages of its own tick. When the two bases differ, the rows another instance rendered were folded at the wrong time. blitz's compose over a picture decoded in the region (1/90000 from MPEG-TS) with change rows from a producer clocked by the same input's raw picture (1/61440 on the pipe) scheduled every row it was handed that way about 1.46 times later than it arrived, so each worker showed a different, older document, while --jobs 1 was right. A row stamped between two ticks was moved to the tick's time even on one base. Each row is now recorded at its own pts: a message's, or the frame's it rode, in its stream's time base, in the order the node reads them. shape_state now also says when its newest word arrived, and a fleet test reads cues on 1/48000 that fall between its ticks on 1/10, the same at -jobs 1, 2 and 4. An exec test runs blitz's emit_changes into compose over an MPEG-TS input at --jobs 1 and the default, frame for frame, where a blitz checkout sits beside this one. Co-Authored-By: Claude Fable 5.1 --- cli/tests/exec/test_exec_wasm.py | 89 +++++++++++++++++++++---- sidecar/NODE-CLI.md | 3 +- sidecar/ffrwd-wasm/src/tick.rs | 52 +++++++++------ sidecar/ffrwd-wasm/tests/node_shapes.rs | 52 +++++++++++++++ sidecar/modules/shape-state/src/lib.rs | 54 ++++++++++----- sidecar/wit/av.wit | 3 +- 6 files changed, 199 insertions(+), 54 deletions(-) diff --git a/cli/tests/exec/test_exec_wasm.py b/cli/tests/exec/test_exec_wasm.py index dd956b50..c1a322ff 100644 --- a/cli/tests/exec/test_exec_wasm.py +++ b/cli/tests/exec/test_exec_wasm.py @@ -1825,20 +1825,17 @@ def test_a_file_an_encoder_in_the_region_writes_says_the_colour_it_was_made_in( _EMIT_CHANGES = _built(_BLITZ, "emit_changes") -def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_beside_it( - tmp_path: Path, -) -> None: - """blitz's `compose` reads the programme's packets, decoded in the - region, and change rows from `emit_changes`, whose clock is the same - input's raw picture. Both cross on one pipe, whose ffmpeg writes them - ahead of its clock by the offset the sidecar takes off. A producer that - writes no rows still lets every frame through.""" +def _composed_changes( + emit: str, html: str, out_path: Path, source: Path = _SOURCE +) -> Compiled: + """blitz's `compose` over `source`'s picture, decoded in the region, + reading on `changes` the rows `emit_changes` writes from `emit`, clocked + by the same input's raw picture.""" for module in (_COMPOSE, _EMIT_CHANGES): if not module.exists(): pytest.skip(f"module missing: {module}") - out_path = tmp_path / "composed.mkv" changes = 'STRUCT(at number, "select" text, change text, text text, html text)' - compiled = compile_all( + return compile_all( f"CREATE FUNCTION emit_changes(v video_stream, emit text) RETURNS {changes}[]\n" f" AS '{_EMIT_CHANGES.as_posix()}', 'emit_changes' LANGUAGE wasm;\n" f"CREATE FUNCTION compose(v video_stream, changes {changes}[][] DEFAULT NULL,\n" @@ -1846,11 +1843,23 @@ def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_bes f" AS '{_COMPOSE.as_posix()}', 'compose' LANGUAGE wasm;\n" "COPY (SELECT compose(f.video[1],\n" " changes => emit_changes(ffmpeg.format(f.video[1], 'yuv420p'),\n" - " emit => '[]'),\n" - " html => '')\n" - f" FROM input('{_SOURCE.as_posix()}') f)\n" + f" emit => '{emit}'),\n" + f" html => '{html}')\n" + f" FROM input('{source.as_posix()}') f)\n" f"TO '{out_path.as_posix()}' WITH (video_codec 'ffv1')" ) + + +def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_beside_it( + tmp_path: Path, +) -> None: + """blitz's `compose` reads the programme's packets, decoded in the + region, and change rows from `emit_changes`, whose clock is the same + input's raw picture. Both cross on one pipe, whose ffmpeg writes them + ahead of its clock by the offset the sidecar takes off. A producer that + writes no rows still lets every frame through.""" + out_path = tmp_path / "composed.mkv" + compiled = _composed_changes("[]", '', out_path) assert compiled.plan is not None result = execute_plan( compiled.plan, sidecar_argv=wasm.sidecar_argv, overwrite=True, @@ -1862,3 +1871,57 @@ def test_a_reader_of_a_producers_rows_takes_every_frame_of_a_picture_decoded_bes for member in stage.members ) assert _video_stream(out_path)["nb_read_frames"] == _video_stream(_SOURCE)["nb_read_frames"] + + +def test_a_document_folding_a_producers_rows_is_the_same_at_any_worker_count( + tmp_path: Path, +) -> None: + """Rows from `emit_changes`, clocked by the input's raw picture, drive + blitz's document over the picture the region decodes from MPEG-TS, a + clock on another time base. Spread over the default workers, every + frame is the one `--jobs 1` makes: an instance is handed the rows of the + ticks it did not render at the times they arrived.""" + source = tmp_path / "in.ts" + subprocess.run( + [ + "ffmpeg", "-v", "error", "-y", "-f", "lavfi", "-i", "testsrc2=size=320x240:rate=30", + "-t", "4", "-c:v", "libx264", "-pix_fmt", "yuv420p", str(source), + ], + check=True, + timeout=_SUBPROCESS_TIMEOUT, + ) + emit = json.dumps([ + [0.3 * k, [{"select": "#n", "text": f"row {k}"}, + {"select": "#box", "change": f"~c{k % 3}"}]] + for k in range(1, 13) + ]) + html = ( + "" + '
' + ) + frames: dict[int | None, list[str]] = {} + for jobs in (1, None): + out_path = tmp_path / f"composed-{jobs}.mkv" + compiled = _composed_changes(emit, html, out_path, source) + assert compiled.plan is not None + result = execute_plan( + compiled.plan, + sidecar_argv=functools.partial(wasm.sidecar_argv, jobs=jobs), + overwrite=True, + timeout=_SUBPROCESS_TIMEOUT, + ) + assert result.exit_code == 0, "\n".join( + f"{member.id} exited {member.exit_code}: {member.stderr_tail}" + for stage in result.stages + for member in stage.members + ) + frames[jobs] = _stream_md5(out_path, 0) + assert len(frames[1]) == int(str(_video_stream(source)["nb_read_frames"])) + assert len(set(frames[1])) > 1, "the rows change the document" + differ = [n for n, (one, many) in enumerate(zip(frames[1], frames[None], strict=True)) + if one != many] + assert not differ, f"frames {differ} differ from --jobs 1's" diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md index 11aac113..fc43c947 100644 --- a/sidecar/NODE-CLI.md +++ b/sidecar/NODE-CLI.md @@ -415,7 +415,8 @@ tick that held them. - **Lanes.** A pure node runs on as many workers as `-jobs` allows, its results put back in tick order before they leave. A pure node with a state input opens every instance before its first tick, and each - instance is handed the rows of the ticks it did not process. Any other + instance is handed the rows of the ticks it did not process, each at + the pts it arrived at in its stream's time base. Any other node runs one tick at a time. - **Finishing.** A node that says it is finished gets its last call on the instance that said so, and its outputs end; a rate clock with inputs diff --git a/sidecar/ffrwd-wasm/src/tick.rs b/sidecar/ffrwd-wasm/src/tick.rs index d6396e03..e4f8465a 100644 --- a/sidecar/ffrwd-wasm/src/tick.rs +++ b/sidecar/ffrwd-wasm/src/tick.rs @@ -291,8 +291,10 @@ impl DataDrive { } /// The rows a state input's instance is owed: those of every tick it did -/// not process since its previous call, oldest first. A frame-parallel -/// node's instances each see every row, whichever ticks they were handed. +/// not process since its previous call, oldest first, each at the pts it +/// arrived at in its stream's time base, as the instance that processed the +/// tick was handed it. A frame-parallel node's instances each see every row +/// at the same time, whichever ticks they were handed. #[derive(Default)] pub struct EarlierRows { /// Per stream id, every tick's rows, by tick number. @@ -302,15 +304,14 @@ pub struct EarlierRows { } impl EarlierRows { - /// The rows stream `id` brought at tick `number`, at `pts`. - pub fn record(&mut self, id: u32, number: u64, pts: i64, rows: Vec) { - if rows.is_empty() { - return; - } - self.ticks - .entry(id) - .or_default() - .push((number, TimedRows { pts, rows })); + /// The rows stream `id` brought at tick `number`. + pub fn record(&mut self, id: u32, number: u64, rows: Vec) { + let ticks = self.ticks.entry(id).or_default(); + ticks.extend( + rows.into_iter() + .filter(|r| !r.rows.is_empty()) + .map(|r| (number, r)), + ); } /// What `instance`, about to process tick `number`, is owed on stream @@ -1126,18 +1127,21 @@ impl Assembler { if !state { continue; } - let rows: Vec = stream - .frames + // Each row at its own pts in the stream's time base, in the + // order the node reads them: messages, then the rows on frames. + let rows: Vec = stream + .messages .iter() - .flat_map(|f| f.rows.iter().cloned()) - .chain( - stream - .messages - .iter() - .map(|m| String::from_utf8_lossy(&m.data).into_owned()), - ) + .map(|m| TimedRows { + pts: m.pts, + rows: vec![String::from_utf8_lossy(&m.data).into_owned()], + }) + .chain(stream.frames.iter().map(|f| TimedRows { + pts: f.pts, + rows: f.rows.clone(), + })) .collect(); - self.earlier.record(stream.id, number, pts, rows); + self.earlier.record(stream.id, number, rows); } Tick { pts, @@ -2124,7 +2128,11 @@ mod tests { fn earlier_rows_are_what_an_instance_missed() { let mut earlier = EarlierRows::default(); for n in 0..4u64 { - earlier.record(1, n, n as i64 * 10, vec![format!("r{n}")]); + let rows = TimedRows { + pts: n as i64 * 10, + rows: vec![format!("r{n}")], + }; + earlier.record(1, n, vec![rows]); } assert!(earlier.owed(0, 1, 0).is_empty()); earlier.processed(0, 0); diff --git a/sidecar/ffrwd-wasm/tests/node_shapes.rs b/sidecar/ffrwd-wasm/tests/node_shapes.rs index a1626c30..c734bf3e 100644 --- a/sidecar/ffrwd-wasm/tests/node_shapes.rs +++ b/sidecar/ffrwd-wasm/tests/node_shapes.rs @@ -326,6 +326,58 @@ fn every_instance_of_a_state_node_sees_every_row() { assert_eq!(ticks[19]["last"], true); } +#[test] +fn every_instance_of_a_state_node_is_handed_each_earlier_row_at_the_time_it_arrived() { + // Cues every 0.25 s at 1/48000 from a producer clocked by the sound, + // read by a node clocked by pictures at 1/10: most cues fall between + // two of the reader's ticks, and none is on the reader's time base. + let dir = scratch("state-times"); + let audio = Stream::audio("f32", 48000, 1).expect("f32 is carried"); + let mut items: Vec<(usize, Write)> = Vec::new(); + for k in 0..40i64 { + items.push((1, Write::Frame(k * 4800, vec![0u8; 4800 * 4]))); + items.push((0, Write::Frame(k, vec![0u8; 16]))); + } + write_nut( + &dir.join("in.nut"), + &[video(2, 2, TENTHS, (10, 1)), audio], + &items, + ); + let input = dir.join("in.nut").display().to_string(); + let out = at_every_jobs( + "state-times", + &args(&[ + "-f", + "nut", + "-i", + &input, + "-m", + &module("shape_window"), + "-m", + &module("shape_state"), + "-filter_complex", + "[a=0:a]shape_window=window=12000[cues=c];[v=0:v][words=c]shape_state[seen=s]", + "-map", + "[s]", + "-f", + "ndjson", + "{dir}/seen.ndjson", + ]), + &["seen.ndjson"], + ); + let ticks = lines(&out[0]); + assert_eq!(ticks.len(), 40); + for (n, tick) in ticks.iter().enumerate() { + // Cue k starts at k * 0.25 s and lands on tick floor(2.5 k). + let cues = (2 * n + 1) / 5; + assert_eq!( + (tick["seen"].as_u64(), tick["newest"].as_f64()), + (Some(cues as u64 + 1), Some(cues as f64 * 0.25)), + "tick {n}" + ); + } +} + #[test] fn an_output_sized_by_the_params_is_redrawn_at_that_size() { let dir = scratch("canvas"); diff --git a/sidecar/modules/shape-state/src/lib.rs b/sidecar/modules/shape-state/src/lib.rs index 3f058362..74b351c5 100644 --- a/sidecar/modules/shape-state/src/lib.rs +++ b/sidecar/modules/shape-state/src/lib.rs @@ -1,7 +1,9 @@ //! A picture clock and words read by interval and kept as state: every tick //! says how many words this instance has seen, its own and those of the -//! ticks it did not process, so a count that matches the serial run's at -//! every worker count is the host handing each instance its earlier rows. +//! ticks it did not process, and when the newest of them arrived, read in +//! the words' time base. A count and a time that match the serial run's at +//! every worker count are the host handing each instance its earlier rows, +//! each at the time it arrived. // Every shape module carries the same few helpers, and not every one uses // each of them. @@ -143,11 +145,17 @@ const SCHEMA: &str = r#"{"type":"object","properties":{"latency":{"type":"number struct State { words: Option, + /// The words' time base. + base: (i32, i32), seen: u64, + /// When the newest word arrived, in the words' time base. + newest: Option, } thread_local! { - static STATE: RefCell = const { RefCell::new(State { words: None, seen: 0 }) }; + static STATE: RefCell = const { + RefCell::new(State { words: None, base: (1, 1), seen: 0, newest: None }) + }; } struct Node; @@ -184,8 +192,14 @@ impl Guest for Node { } fn init(bound: Vec, _latched: Vec, _params: String) -> Result<(), String> { - let words = bound_ids(&bound, "words").first().copied(); - STATE.with(|s| *s.borrow_mut() = State { words, seen: 0 }); + let words = bound.iter().find(|b| b.port == "words"); + let state = State { + words: words.map(|b| b.id), + base: words.map_or((1, 1), |b| (b.info.time_base.num, b.info.time_base.den)), + seen: 0, + newest: None, + }; + STATE.with(|s| *s.borrow_mut() = state); Ok(()) } @@ -196,21 +210,27 @@ impl Guest for Node { fn process(tick: &Tick) -> Result { STATE.with(|s| { let mut state = s.borrow_mut(); - let (earlier, now) = match state.words { - Some(id) => ( - tick.earlier_rows(id) - .iter() - .map(|r| r.rows.len() as u64) - .sum::(), - tick.messages(id).len() as u64, - ), - None => (0, 0), - }; - state.seen += earlier + now; + let mut now = 0; + if let Some(id) = state.words { + for rows in tick.earlier_rows(id) { + state.seen += rows.rows.len() as u64; + state.newest = Some(rows.pts); + } + for m in tick.messages(id) { + now += 1; + state.seen += 1; + state.newest = Some(m.pts); + } + } + let (num, den) = state.base; + let newest = state + .newest + .map(|pts| pts as f64 * f64::from(num) / f64::from(den)); let item = message( "seen", tick.pts(), - json!({"pts": tick.pts(), "seen": state.seen, "now": now, "last": tick.last()}), + json!({"pts": tick.pts(), "seen": state.seen, "now": now, "newest": newest, + "last": tick.last()}), ); Ok(emitted(vec![item], false)) }) diff --git a/sidecar/wit/av.wit b/sidecar/wit/av.wit index 43d6a8e1..470765bc 100644 --- a/sidecar/wit/av.wit +++ b/sidecar/wit/av.wit @@ -660,7 +660,8 @@ interface node-tick { /// For an input whose `rows` is `state`: the rows of every tick this /// instance did not process since its previous call, oldest first, - /// each beside the pts it arrived at, all earlier than this call's. An + /// each beside the pts it arrived at in its stream's time base (a + /// message's, or the frame's it rode), all earlier than this call's. An /// instance's first call is handed every one before it. A data input's /// rows are its messages. Empty on every other input. earlier-rows: func(id: u32) -> list;