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;