diff --git a/plans/gpu-resident-frames.md b/plans/gpu-resident-frames.md index 03bf066..2d380cb 100644 --- a/plans/gpu-resident-frames.md +++ b/plans/gpu-resident-frames.md @@ -2924,18 +2924,51 @@ probe. The real findings from that work: ### What is left -- A lane's task queue can grow past the scheduler's cap. On foundmeat's - libx264 run, a node of the foundmeat package in compose's sidecar reached - 1023 tasks against a cap of 10, while the reader of the stream it reads - waited for room. How it got past the cap is not known yet. The cap - (`lane.queue.len() < cap`) is checked only where a producer waits for - room, before a reader delivers and before a lane starts a task, so tasks - queued any other way would not be counted. The check predates the - branch's change to the bound, since a6a649c counts only the arrivals - queued at a self-clocked reader. While such a queue stays above the cap, - every reader of the streams that lane reads is held back with it. Next: - find which path queued the tasks, with the `room:` trace and the lane's - intake, and bound it. +- The 1023-task queue and 0.32.0 falling short of real time (2026-10-19). + The pace fix shipped in 0.32.1; the waker fix (the scheduler's waker + set after its workers started, on main too, the cause of the hangs + seen while testing) shipped in 0.32.2; the lane change below, held + back at first since the queue did not keep 0.31.0 from real time, + went in once the CPU profile showed it was the idle broadcast's + cost. The lane was `listen_audio` in + compose's sidecar. Its clock is the programme's sound, an `asplit` copy + that rides one input with compose's own `a`, and the module declares + a window of one sample, so each 1024-sample packet settled 1024 ticks. + An input clock's cut queued every settled tick at once (the cap was + checked only where a reader delivers), and every tick cost a turn of the + scheduler. On top of that, 3dad921's pace check walked the whole quarter + second of ticks (12000 of them) on every tick whenever the clock ran at + real time. The lane fell behind and held back the reader that carries + compose's sound, so the programme ran at 50 fps: 8.3 media seconds per + 10 s of wall on 0.32.0 with the main feed alone, against 10.15 on + 0.31.0. Bisected in a container on the box. a6a649c and 912bf25 + reverted: 8.34. bcce48e: 10.15. 8f6b9d9: 8.34. 0.32.0 with the pace + check off: 10.15. Fixed: the pace keeps running extremes and is asked + only where a port-fed hold uses it; the cut stops at the cap and holds + the next tick back; a one-at-a-time lane takes up to 1024 settled ticks + in a turn; a turn's progress marks leave once; and an audio clock's + reader is held back by packets waiting, not by ticks. With two cams the + query keeps real time (10.15 against 8.02 on 0.32.0). Left for + foundmeat: `listen_audio` could declare a window of a packet or more, + since it reads only the times. +- The Linux route's CPU was measured on 2026-10-20 (perf, strace, gdb on + the box): 1.34 cores idle on 0.32.1, 61% of it the kernel, the four + workers making about 30,000 context switches a second each because a + node clocked a sample a tick (foundmeat's listener on the programme's + sound) was cut one task a sample and every dispatch and completion + woke every worker. The host's waits on the device cost under 1%, and + CUDA's waits already block (libav sets `CU_CTX_SCHED_BLOCKING_SYNC` + itself). The lane change (the cut stops at the cap, ticks taken in + turns) brought it to 0.99 and keeping the host's plane textures to + 0.92, the Windows figure; the node's own window of a packet rather + than a sample (foundmeat 0.3.0) brought their idle broadcast to 0.46 + on 0.32.2 alone. Two further commits, the decoder's copy and the + encoder's feed submitted on the device after the picture's semaphore + with no host wait (branch `fix/linux-route-cpu`, 91c0689 and + 6a87172), measured within noise and rest on wgpu-core's submit + holding its device fence lock for writing, which is not documented; + they stay unmerged until a wgpu release documents that or the saving + is worth the dependence. - The host waits for the decoder's write before each copy is submitted, and the decode thread waits with it. A wait on the queue would need a submit with wait semaphores, which wgpu-core does not have; a raw submit diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md index 342aaa0..ba55ab9 100644 --- a/sidecar/NODE-CLI.md +++ b/sidecar/NODE-CLI.md @@ -369,6 +369,28 @@ tick that held them. the clock reads: an `arrival` input's pts may sit on any origin (a feed at the epoch while the clock counts from 0), so they never hold the clock. +- **Queues.** A node's ticks wait in a queue of at most `2 * jobs + 2` + until a worker runs them, and a reader waits while the queue of + anything reading its stream is full. Every clock is cut no further than + that. One arrival can settle many ticks: an audio clock ticks every + `window` samples, so a node with a window of one sample on a clock + bound to sound is cut 1024 ticks by a packet of 1024 samples, 48000 a + second. Those past the bound stay with the clock and are cut as the + queue drains, and while any are left the node counts as full to its + readers, except to the reader of that audio clock, which waits only + once the bound's worth of packets waits on the clock. A self-clocked node is handed one arrival an input a call, and + the arrivals waiting on its input count against the same bound, so a + decoder ahead of a slower encoder holds its reader back rather than + queue every picture it made. A node that runs one tick at a time takes + the ticks queued behind the one it starts, and those its clock has + settled past the bound, up to 1024, and runs them in order in one turn, + and the progress they send on each port leaves once, as the last of + them: a clock of small ticks costs a node call a tick, not a turn of + the scheduler a tick. A node with ticks held back goes first. A node + slower than its clock holds back the reader of its clock's input, and + with it every stream that input carries (two `asplit` copies of one + sound, one for a listener and one for the programme, ride one input), + so it can slow the whole programme. - **Hold inputs.** A hold input bound to a stream holds the tick until its source has arrived past it (its picture, or with none its sound, past `lead` beyond the tick), or ended, for no longer than its diff --git a/sidecar/ffrwd-wasm/src/lanes.rs b/sidecar/ffrwd-wasm/src/lanes.rs index ce5d851..2d98e79 100644 --- a/sidecar/ffrwd-wasm/src/lanes.rs +++ b/sidecar/ffrwd-wasm/src/lanes.rs @@ -167,6 +167,16 @@ enum LaneIntake { }, } +/// What a worker takes from a lane in one turn: the tasks it runs in +/// order, the instance that runs them, and that instance or the means to +/// open it. +struct Turn { + tasks: Vec, + instance: usize, + runner: Option>, + opener: Option, +} + struct Task { ordinal: u64, tick: Tick, @@ -229,6 +239,10 @@ struct Lane { started: Option, /// When a rate clock whose inputs all arrive unpaired last cut a tick. turned: Option, + /// The next tick of an input clock whose cut stopped at the cap, and + /// where its interval ends: while one is held back the lane counts as + /// full to everything it reads. + withheld: Option<(Tick, Option)>, unbuffered: bool, } @@ -390,7 +404,13 @@ impl State { /// taking one arrival an input, so what waits for it is the arrivals /// queued on its input, which count against the same bound as a task /// queue: without that, a decoder ahead of a slower encoder would queue - /// every picture it made. + /// every picture it made. An input clock is cut no further than the + /// cap, and while its cut stopped there the lane counts as full, but + /// to the reader of its audio clock: one packet of sound settles as + /// many ticks as it has samples over the stride, so while ticks are + /// held back what waits there is counted in packets, against the same + /// bound, and that reader is held back only once the lane is that many + /// packets behind. fn room(&self, stream: u32, consumers: &[Consumer]) -> bool { consumers.iter().all(|c| match c { Consumer::Lane(j) => { @@ -401,7 +421,21 @@ impl State { } _ => 0, }; - lane.stopped || (lane.queue.len() < self.cap && queued < self.cap) + let sound = match &lane.intake { + LaneIntake::Assembled(a) if lane.clock == ClockKind::Input => { + a.audio_backlog(stream) + } + _ => None, + }; + lane.stopped + || match sound { + Some(packets) if lane.withheld.is_some() => packets < self.cap, + _ => { + lane.queue.len() < self.cap + && lane.withheld.is_none() + && queued < self.cap + } + } } Consumer::Writer(w, _) => self.writers[*w].len() < self.cap, }) @@ -648,14 +682,36 @@ impl State { let LaneIntake::Assembled(a) = &mut lane.intake else { unreachable!("an input clock is assembled") }; - let mut cut = Vec::new(); - while let Some(tick) = a.next_clocked()? { - let last = tick.last; - cut.push((tick, a.tick_end())); + // One arrival can settle many ticks (an audio clock cuts + // `window` samples a tick, so a packet of 1024 makes 1024 + // ticks of one), so the cut stops at the cap as every other + // clock's does, holding the next tick back if one is + // settled, and the rest is cut as the queue drains. + let mut cut: Vec<(Tick, Option)> = Vec::new(); + let mut last = false; + while lane.queue.len() + cut.len() < cap { + let next = match lane.withheld.take() { + Some(held) => held, + None => match a.next_clocked()? { + Some(tick) => { + let end = a.tick_end(); + (tick, end) + } + None => break, + }, + }; + last = next.0.last; + cut.push(next); if last { break; } } + if !last && lane.withheld.is_none() && lane.queue.len() + cut.len() >= cap { + if let Some(tick) = a.next_clocked()? { + let end = a.tick_end(); + lane.withheld = Some((tick, end)); + } + } for (tick, end) in cut { Self::push_task(lane, tick, end); } @@ -844,31 +900,44 @@ impl State { } } - /// A rate clock's ticks are cut as its queue has room, and a - /// self-clocked node's as it returns, so both are cut here, where a - /// worker looks for work, as well as on arrivals. + /// A rate clock's ticks are cut as its queue has room, a self-clocked + /// node's as it returns, and an input clock's that the cap withheld as + /// its queue drains, so all are cut here, where a worker looks for + /// work, as well as on arrivals. A lane holding a tick back holds back + /// the readers of what it reads, so it goes first; then the lane with + /// the longest queue. fn pick(&mut self) -> Result> { for j in 0..self.lanes.len() { - if matches!( - self.lanes[j].clock, - ClockKind::Rate { .. } | ClockKind::SelfClocked - ) { + let lane = &self.lanes[j]; + if matches!(lane.clock, ClockKind::Rate { .. } | ClockKind::SelfClocked) + || (lane.withheld.is_some() && lane.queue.len() < self.cap) + { self.cut(j)?; } } Ok((0..self.lanes.len()) .filter(|i| self.dispatchable(*i)) - .max_by_key(|i| self.lanes[*i].queue.len())) + .max_by_key(|i| { + ( + self.lanes[*i].withheld.is_some(), + self.lanes[*i].queue.len(), + ) + })) } /// Takes lane `i`'s front task with an instance, or the means to open - /// one, and hands a state input its earlier rows. - fn dispatch(&mut self, i: usize) -> (Task, usize, Option>, Option) { + /// one, and hands a state input its earlier rows. A lane that runs one + /// tick at a time takes the tasks queued behind the front one too, and + /// the ticks its clock has settled past the cap, up to [`TURN`] and + /// short of its last, and runs them in order in one turn: a clock of + /// small ticks (an audio clock cut a sample a tick, 48000 a second) + /// would otherwise wait for a worker, and pay a turn of the scheduler + /// and a wake of every thread waiting on it, for each. + fn dispatch(&mut self, i: usize) -> Result { let lane = &mut self.lanes[i]; - let mut task = lane.queue.pop_front().expect("picked lanes have a task"); - task.tick.ordinal = task.ordinal; + let front = lane.queue.pop_front().expect("picked lanes have a task"); lane.in_flight += 1; - let wanted = match (task.tick.last, lane.finisher) { + let wanted = match (front.tick.last, lane.finisher) { (true, Some(finisher)) => lane.idle.iter().position(|(i, _)| *i == finisher), _ => lane.idle.len().checked_sub(1), }; @@ -880,55 +949,89 @@ impl State { (instance, None, lane.opener.clone()) } }; - if !lane.state.is_empty() { - if let LaneIntake::Assembled(a) = &mut lane.intake { - for stream in &mut task.tick.streams { - if lane.state.contains(&stream.id) { - stream.earlier_rows = a.earlier.owed(instance, stream.id, task.ordinal); - } + let mut tasks = vec![front]; + if lane.width == 1 && !lane.unbuffered && !tasks[0].tick.last { + loop { + let lane = &mut self.lanes[i]; + while tasks.len() < TURN && lane.queue.front().is_some_and(|t| !t.tick.last) { + tasks.extend(lane.queue.pop_front()); } - a.earlier.processed(instance, task.ordinal); - if lane.created >= lane.width { - a.earlier.forget_told(lane.created); + if tasks.len() >= TURN || !lane.queue.is_empty() || lane.withheld.is_none() { + break; } + self.cut(i)?; } } - if let LaneIntake::Assembled(a) = &mut lane.intake { - for stream in &mut task.tick.streams { - stream.ended_feeds = a.ended.owed(instance, stream.id, task.ordinal); + let lane = &mut self.lanes[i]; + for task in &mut tasks { + task.tick.ordinal = task.ordinal; + if !lane.state.is_empty() { + if let LaneIntake::Assembled(a) = &mut lane.intake { + for stream in &mut task.tick.streams { + if lane.state.contains(&stream.id) { + stream.earlier_rows = a.earlier.owed(instance, stream.id, task.ordinal); + } + } + a.earlier.processed(instance, task.ordinal); + if lane.created >= lane.width { + a.earlier.forget_told(lane.created); + } + } } - a.ended.processed(instance, task.ordinal); - if lane.created >= lane.width { - a.ended.forget_told(lane.created); + if let LaneIntake::Assembled(a) = &mut lane.intake { + for stream in &mut task.tick.streams { + stream.ended_feeds = a.ended.owed(instance, stream.id, task.ordinal); + } + a.ended.processed(instance, task.ordinal); + if lane.created >= lane.width { + a.ended.forget_told(lane.created); + } } } - (task, instance, runner, opener) + Ok(Turn { + tasks, + instance, + runner, + opener, + }) } + /// The results of one turn of lane `i`, by ordinal, in the order they + /// ran. fn complete( &mut self, i: usize, - ordinal: u64, + results: Vec<(u64, Result)>, instance: usize, runner: Option>, - done: Result, ) { let lane = &mut self.lanes[i]; lane.in_flight -= 1; - let last = done.as_ref().is_ok_and(|d| d.last); + let last = results + .iter() + .any(|(_, done)| done.as_ref().is_ok_and(|d| d.last)); if let Some(runner) = runner { if !last { lane.idle.push((instance, runner)); } } - let flushed = match done { - Ok(done) => { - if ordinal >= lane.next_flush { - lane.done.insert(ordinal, done); + let mut failed = None; + for (ordinal, done) in results { + match done { + Ok(done) => { + if ordinal >= lane.next_flush { + lane.done.insert(ordinal, done); + } + } + Err(e) => { + failed = Some(e.context(format!("in {}", lane.name))); + break; } - self.flush(i) } - Err(e) => Err(e.context(format!("in {}", self.lanes[i].name))), + } + let flushed = match failed { + Some(e) => Err(e), + None => self.flush(i), }; if let Err(e) = flushed { if self.error.is_none() { @@ -941,6 +1044,7 @@ impl State { /// port's readers, the node's progress down every port, and on the last /// tick the end of every port. fn flush(&mut self, i: usize) -> Result<()> { + let mut leaving: Vec<(u32, Delivery)> = Vec::new(); loop { let lane = &mut self.lanes[i]; let Some(done) = lane.done.remove(&lane.next_flush) else { @@ -1098,6 +1202,7 @@ impl State { lane.stopped = true; lane.finisher = Some(instance); lane.queue.clear(); + lane.withheld = None; lane.done.clear(); lane.next_flush = lane.next_ordinal; let streams = lane @@ -1123,17 +1228,18 @@ impl State { }); lane.next_ordinal += 1; } - for (stream, delivery) in deliveries { - match delivery { - Delivery::Item(item) => self.arrive(stream, item)?, - Delivery::Progress(at) => self.progress(stream, at)?, - Delivery::End => self.end(stream)?, - Delivery::Trailing(rows) => self.trailing(stream, &rows), - Delivery::Batch => { - for consumer in self.consumers(stream) { - if let Consumer::Writer(w, s) = consumer { - self.writers[w].push(s, Out::Batch); - } + leaving.extend(deliveries); + } + for (stream, delivery) in superseded_progress_dropped(leaving) { + match delivery { + Delivery::Item(item) => self.arrive(stream, item)?, + Delivery::Progress(at) => self.progress(stream, at)?, + Delivery::End => self.end(stream)?, + Delivery::Trailing(rows) => self.trailing(stream, &rows), + Delivery::Batch => { + for consumer in self.consumers(stream) { + if let Consumer::Writer(w, s) = consumer { + self.writers[w].push(s, Out::Batch); } } } @@ -1162,6 +1268,28 @@ enum Delivery { Batch, } +/// What several ticks of one lane hand on, less each progress mark that a +/// later one on the same stream says again with nothing between them on +/// that stream: a turn of many small ticks moves its readers' progress once. +/// A mark is kept when it is the last on its stream before the stream's +/// next item, end or rows, or the last on its stream at all. +fn superseded_progress_dropped(deliveries: Vec<(u32, Delivery)>) -> Vec<(u32, Delivery)> { + let mut marked: HashSet = HashSet::new(); + let mut kept: Vec<(u32, Delivery)> = deliveries + .into_iter() + .rev() + .filter(|(stream, delivery)| match delivery { + Delivery::Progress(_) => marked.insert(*stream), + _ => { + marked.remove(stream); + true + } + }) + .collect(); + kept.reverse(); + kept +} + fn out_of(item: &Item) -> Out { match item { Item::Frame(f) => Out::Frame { @@ -1180,6 +1308,10 @@ fn out_of(item: &Item) -> Out { /// core from the ffmpeg processes and encoders beside it. pub const DEFAULT_WORKERS: usize = 4; +/// The most ticks one turn of a lane that runs a tick at a time takes: a +/// packet of sound's worth on a clock cut a sample a tick. +const TURN: usize = 1024; + /// How many worker threads a run gets: `-jobs` when it was given, else /// `DEFAULT_WORKERS`, never more than the machine's effective core count. /// `-jobs 1` is the serial escape hatch. @@ -1294,6 +1426,7 @@ impl Scheduler { finisher: None, started: None, turned: None, + withheld: None, unbuffered: spec.unbuffered, }); } @@ -1360,7 +1493,12 @@ impl Scheduler { .map(|c| match c { Consumer::Lane(j) => { let lane = &state.lanes[*j]; - format!("{} queue {}", lane.name, lane.queue.len()) + let withheld = if lane.withheld.is_some() { + " and a tick withheld" + } else { + "" + }; + format!("{} queue {}{withheld}", lane.name, lane.queue.len()) } Consumer::Writer(w, _) => format!("writer {w} {}", state.writers[*w].len()), }) @@ -1542,7 +1680,19 @@ fn worker(shared: &Shared) { }; continue; }; - let (task, instance, runner, opener) = state.dispatch(i); + let Turn { + tasks, + instance, + runner, + opener, + } = match state.dispatch(i) { + Ok(taken) => taken, + Err(e) => { + state.error = Some(e); + shared.notify(); + return; + } + }; drop(state); shared.notify(); @@ -1553,27 +1703,43 @@ fn worker(shared: &Shared) { None => Err(anyhow!("a lane ran out of instances")), }, }; - let Task { ordinal, tick, end } = task; - let (pts, last) = (tick.pts, tick.last); - let (runner, done) = match opened { + let mut results = Vec::with_capacity(tasks.len()); + let runner = match opened { Ok(mut runner) => { - let started = Instant::now(); - let result = runner.process(tick); - PROFILE.process.add(0, started); - let result = result.map(|emitted| Done { - emitted, - pts, - end, - last, - instance, - }); - (Some(runner), result) + for Task { ordinal, tick, end } in tasks { + let (pts, last) = (tick.pts, tick.last); + let started = Instant::now(); + let result = runner.process(tick); + PROFILE.process.add(0, started); + // A tick the node says it finished on, or failed on, is + // the last of the turn: what was taken after it is not + // run, as a lane that stops drops its queue. Its flush + // skips their ordinals, drops what is queued (a last + // tick behind them too) and queues the last call for + // the instance that finished; a failure stops the run. + let stop = result.as_ref().map_or(true, |emitted| emitted.finished); + let result = result.map(|emitted| Done { + emitted, + pts, + end, + last, + instance, + }); + results.push((ordinal, result)); + if stop { + break; + } + } + Some(runner) + } + Err(e) => { + results.push((tasks[0].ordinal, Err(e))); + None } - Err(e) => (None, Err(e)), }; state = shared.lock(); - state.complete(i, ordinal, instance, runner, done); + state.complete(i, results, instance, runner); shared.notify(); } } @@ -1592,8 +1758,8 @@ mod tests { assert_eq!(worker_count(Some(1)), 1); assert_eq!(worker_count(Some(usize::MAX)), cores); } - use ffrwd_wasm_runtime::runtime::{StreamInfo, VideoFormat}; - use std::sync::mpsc; + use ffrwd_wasm_runtime::runtime::{AudioFormat, StreamInfo, VideoFormat}; + use std::sync::{mpsc, OnceLock}; /// What a publisher's calls brought: frames per call, and whether each /// was the last. @@ -1847,4 +2013,531 @@ mod tests { writer.join(); let _ = std::fs::remove_dir_all(&dir); } + + /// Opens when told to, for a node to wait on in its first call. + #[derive(Default)] + struct Gate { + open: Mutex, + opened: Condvar, + } + + impl Gate { + fn wait(&self) { + let mut open = self.open.lock().unwrap(); + while !*open { + open = self.opened.wait(open).unwrap(); + } + } + + fn open(&self) { + *self.open.lock().unwrap() = true; + self.opened.notify_all(); + } + } + + /// A node that notes the pts of each tick and, at each call, how long + /// its lane's queue is; it waits on `gate` in its first call. + struct Noting { + shape: NodeShape, + seen: Arc>>, + queues: Arc>>, + gate: Arc, + shared: Arc>>, + } + + impl Node for Noting { + fn name(&self) -> &str { + "noting" + } + + fn shape(&self) -> &NodeShape { + &self.shape + } + + fn set_params(&mut self, _params: &str) -> Result<()> { + Ok(()) + } + + fn process(&mut self, tick: Tick) -> Result { + if let Some(shared) = self.shared.get() { + let queue = shared.lock().lanes[0].queue.len(); + self.queues.lock().unwrap().push(queue); + } + let first = { + let mut seen = self.seen.lock().unwrap(); + if !tick.last { + seen.push(tick.pts); + } + seen.len() == 1 + }; + if first { + self.gate.wait(); + } + Ok(Emitted::default()) + } + } + + /// One lane running `node`, clocked by its one input `port` on stream + /// 0, and the scheduler running it on one worker (a cap of 4). + fn one_lane(shape: NodeShape, bound: BoundStream, node: Noting) -> Scheduler { + let assembler = Assembler::new(&shape, &[bound], "noting", &[]).expect("an assembler"); + let plan = Plan { + lanes: vec![LaneSpec { + gpu: HashMap::new(), + name: "noting".to_string(), + shape, + intake: Intake::Assembled(Box::new(assembler)), + bound: vec![0], + state: Vec::new(), + tick_base: TimeBase { + num: 1, + den: 48_000, + }, + runners: vec![Box::new(node)], + opener: None, + ports: Vec::new(), + rows: None, + unbuffered: false, + }], + consumers: HashMap::from([(0, vec![Consumer::Lane(0)])]), + writers: Vec::new(), + mirrors: HashMap::new(), + }; + Scheduler::start(plan, 1, &WakeSlot::new()).expect("the lanes") + } + + fn wait_until(what: &str, done: impl Fn() -> bool) { + let until = Instant::now() + Duration::from_secs(5); + while !done() { + assert!(Instant::now() < until, "{what}"); + thread::sleep(Duration::from_millis(5)); + } + } + + fn input_port(name: &str, kind: PortKind, pairing: Pairing) -> InputPort { + InputPort { + name: name.to_string(), + kind, + required: true, + many: false, + pairing, + rows: RowsUse::Ignore, + window: 1, + stride: 1, + accepts: Accepts::default(), + schema: None, + } + } + + /// A node clocked by its sound a sample a tick (a listener written for + /// pictures, bound to sound) is cut 1024 ticks by one packet of 1024 + /// samples. The cut stops at the cap, the reader waits while ticks are + /// withheld, every tick still runs once and in order, and the ticks + /// queued behind the first are taken in one turn. + #[test] + fn an_audio_clock_a_sample_a_tick_queues_no_more_than_the_cap_and_runs_every_tick() { + let shape = NodeShape { + inputs: vec![input_port("a", PortKind::Audio, Pairing::Lockstep)], + outputs: Vec::new(), + clock: Clock::Input("a".to_string()), + pure: false, + one_to_one: false, + bounded: true, + relation: Vec::new(), + }; + let samples = TimeBase { + num: 1, + den: 48_000, + }; + let bound = BoundStream { + port: "a".to_string(), + id: 0, + info: StreamInfo::default(), + time_base: samples, + format: StreamFormat::Audio(AudioFormat { + sample_rate: 48_000, + channels: 1, + sample_fmt: "f32", + channel_layout: None, + }), + rendition: Default::default(), + row: None, + decode_delay: 0, + latency: None, + hint: Default::default(), + }; + let seen = Arc::new(Mutex::new(Vec::new())); + let queues = Arc::new(Mutex::new(Vec::new())); + let gate = Arc::new(Gate::default()); + let shared = Arc::new(OnceLock::new()); + let node = Noting { + shape: shape.clone(), + seen: Arc::clone(&seen), + queues: Arc::clone(&queues), + gate: Arc::clone(&gate), + shared: Arc::clone(&shared), + }; + let scheduler = one_lane(shape, bound, node); + let _ = shared.set(Arc::clone(&scheduler.shared)); + let packet = |pts: i64| { + Item::Frame(TickFrame { + pts, + duration: Some(1024), + data: Arc::new(vec![0u8; 1024 * 4]).into(), + rows: Vec::new(), + }) + }; + assert!(scheduler.arrive(0, packet(0))); + wait_until("the first tick ran", || !seen.lock().unwrap().is_empty()); + // The first packet's ticks went in one turn; the second's wait, and + // the reader goes on until the cap's worth of packets waits. + assert!(scheduler.arrive(0, packet(1024))); + { + let state = scheduler.shared.lock(); + let lane = &state.lanes[0]; + assert_eq!( + lane.queue.len(), + state.cap, + "the queue holds the cap's worth of the second packet's ticks" + ); + assert!( + lane.withheld.is_some(), + "the rest of the packet's ticks wait" + ); + assert!( + state.room(0, &state.consumers(0)), + "one packet behind, the reader goes on" + ); + } + for k in 2..5 { + assert!(scheduler.arrive(0, packet(k * 1024))); + } + { + let state = scheduler.shared.lock(); + assert!( + state.lanes[0].queue.len() <= state.cap, + "the queue holds no more than the cap" + ); + assert!( + !state.room(0, &state.consumers(0)), + "the reader waits once the cap's worth of packets waits" + ); + } + gate.open(); + wait_until("every sample's tick ran", || { + seen.lock().unwrap().len() == 5 * 1024 + }); + assert!(scheduler.end(0)); + let (done, finished) = mpsc::channel(); + let shared_queues = Arc::clone(&queues); + thread::spawn(move || done.send(scheduler.finish())); + finished + .recv_timeout(Duration::from_secs(5)) + .expect("the lane ended with its input") + .expect("the run"); + assert_eq!(*seen.lock().unwrap(), (0..5 * 1024).collect::>()); + let queues = shared_queues.lock().unwrap(); + assert!( + queues[1..1024].iter().all(|q| *q == 4), + "the first packet's ticks ran in one turn, the second's queued behind it: {:?}", + &queues[..12] + ); + assert!( + queues.iter().all(|q| *q <= 4), + "no call saw more than the cap" + ); + } + + /// A node that waits on `gate` in its first call and finishes, or + /// fails, on its third; it notes each call's pts and whether it was + /// the last. + struct Stopping { + shape: NodeShape, + calls: Arc>>, + gate: Arc, + fail: bool, + } + + impl Node for Stopping { + fn name(&self) -> &str { + "stopping" + } + + fn shape(&self) -> &NodeShape { + &self.shape + } + + fn set_params(&mut self, _params: &str) -> Result<()> { + Ok(()) + } + + fn process(&mut self, tick: Tick) -> Result { + let count = { + let mut calls = self.calls.lock().unwrap(); + calls.push((tick.pts, tick.last)); + calls.len() + }; + if count == 1 { + self.gate.wait(); + } + if count == 3 && !tick.last { + if self.fail { + bail!("the third call fails"); + } + return Ok(Emitted { + finished: true, + ..Emitted::default() + }); + } + Ok(Emitted::default()) + } + } + + /// What a run of [`a_turn_cut_short`] came to: the node's calls, the + /// run's end, and the lane's next ordinal, its front task's and whether + /// that is the last, while the first call waited. + struct CutShort { + calls: Vec<(i64, bool)>, + result: Result<()>, + queued: Vec>, + } + + /// One lane of a node clocked by its sound a sample a tick, its first + /// turn holding six ticks and the last tick queued behind them, the node + /// finishing (or failing) on the third: the run ends, and nothing waits. + fn a_turn_cut_short(fail: bool) -> CutShort { + let shape = NodeShape { + inputs: vec![input_port("a", PortKind::Audio, Pairing::Lockstep)], + outputs: Vec::new(), + clock: Clock::Input("a".to_string()), + pure: false, + one_to_one: false, + bounded: true, + relation: Vec::new(), + }; + let bound = BoundStream { + port: "a".to_string(), + id: 0, + info: StreamInfo::default(), + time_base: TimeBase { + num: 1, + den: 48_000, + }, + format: StreamFormat::Audio(AudioFormat { + sample_rate: 48_000, + channels: 1, + sample_fmt: "f32", + channel_layout: None, + }), + rendition: Default::default(), + row: None, + decode_delay: 0, + latency: None, + hint: Default::default(), + }; + let assembler = Assembler::new(&shape, &[bound], "stopping", &[]).expect("an assembler"); + let calls = Arc::new(Mutex::new(Vec::new())); + let gate = Arc::new(Gate::default()); + let node = Stopping { + shape: shape.clone(), + calls: Arc::clone(&calls), + gate: Arc::clone(&gate), + fail, + }; + let plan = Plan { + lanes: vec![LaneSpec { + gpu: HashMap::new(), + name: "stopping".to_string(), + shape, + intake: Intake::Assembled(Box::new(assembler)), + bound: vec![0], + state: Vec::new(), + tick_base: TimeBase { + num: 1, + den: 48_000, + }, + runners: vec![Box::new(node)], + opener: None, + ports: Vec::new(), + rows: None, + unbuffered: false, + }], + consumers: HashMap::from([(0, vec![Consumer::Lane(0)])]), + writers: Vec::new(), + mirrors: HashMap::new(), + }; + let scheduler = Scheduler::start(plan, 1, &WakeSlot::new()).expect("the lanes"); + let packet = TickFrame { + pts: 0, + duration: Some(6), + data: Arc::new(vec![0u8; 6 * 4]).into(), + rows: Vec::new(), + }; + assert!(scheduler.arrive(0, Item::Frame(packet))); + wait_until("the first tick ran", || !calls.lock().unwrap().is_empty()); + assert!(scheduler.end(0)); + let queued = { + let state = scheduler.shared.lock(); + let lane = &state.lanes[0]; + vec![ + Some(lane.next_ordinal), + lane.queue.front().map(|t| t.ordinal), + Some(u64::from(lane.queue.front().is_some_and(|t| t.tick.last))), + ] + }; + gate.open(); + let (done, finished) = mpsc::channel(); + thread::spawn(move || done.send(scheduler.finish())); + let result = finished + .recv_timeout(Duration::from_secs(5)) + .expect("the run ended, nothing waiting"); + let calls = calls.lock().unwrap().clone(); + CutShort { + calls, + result, + queued, + } + } + + #[test] + fn a_node_that_finishes_in_a_turn_skips_the_rest_of_it_and_takes_its_last_call() { + let CutShort { + calls, + result, + queued, + } = a_turn_cut_short(false); + assert_eq!( + queued, + [Some(7), Some(6), Some(1)], + "six ticks in the turn, the last queued behind them" + ); + result.expect("the run"); + assert_eq!( + calls, + [(0, false), (1, false), (2, false), (3, true)], + "the ticks after the one it finished on are skipped, and the last call follows" + ); + } + + #[test] + fn a_node_that_fails_in_a_turn_stops_the_run() { + let CutShort { calls, result, .. } = a_turn_cut_short(true); + let error = result.expect_err("the run failed"); + assert!( + format!("{error:#}").contains("the third call fails"), + "{error:#}" + ); + assert_eq!(calls, [(0, false), (1, false), (2, false)]); + } + + /// A self-clocked reader (a decoder) slower than what reaches it holds + /// its writer back once the cap's worth of arrivals waits on its input, + /// whatever else the scheduler bounds. + #[test] + fn a_self_clocked_reader_holds_its_writer_at_the_cap() { + let shape = NodeShape { + inputs: vec![input_port("v", PortKind::Video, Pairing::Arrival)], + outputs: Vec::new(), + clock: Clock::SelfClocked, + pure: false, + one_to_one: false, + bounded: true, + relation: Vec::new(), + }; + let bound = BoundStream { + port: "v".to_string(), + id: 0, + info: StreamInfo::default(), + time_base: TimeBase { num: 1, den: 30 }, + format: StreamFormat::Video(VideoFormat { + width: 2, + height: 2, + pix_fmt: "rgba", + frame_len: 16, + color: None, + }), + rendition: Default::default(), + row: None, + decode_delay: 0, + latency: None, + hint: Default::default(), + }; + let seen = Arc::new(Mutex::new(Vec::new())); + let gate = Arc::new(Gate::default()); + let node = Noting { + shape: shape.clone(), + seen: Arc::clone(&seen), + queues: Arc::default(), + gate: Arc::clone(&gate), + shared: Arc::default(), + }; + let scheduler = Arc::new(one_lane(shape, bound, node)); + let delivered = Arc::new(Mutex::new(0usize)); + let writer = { + let scheduler = Arc::clone(&scheduler); + let delivered = Arc::clone(&delivered); + thread::spawn(move || { + for k in 0..40 { + let frame = TickFrame { + pts: k, + duration: None, + data: Arc::new(vec![0; 16]).into(), + rows: Vec::new(), + }; + if !scheduler.arrive(0, Item::Frame(frame)) { + return; + } + *delivered.lock().unwrap() += 1; + } + scheduler.end(0); + }) + }; + wait_until("the first call started", || { + !seen.lock().unwrap().is_empty() + }); + thread::sleep(Duration::from_millis(200)); + let held = *delivered.lock().unwrap(); + assert!( + held <= 4 + 1, + "the writer waits once the cap's worth is queued ({held} delivered)" + ); + gate.open(); + writer.join().expect("the writer"); + scheduler.finish().expect("the run"); + assert_eq!(*delivered.lock().unwrap(), 40); + } + + #[test] + fn a_progress_mark_a_later_one_says_again_is_dropped() { + let marks = |d: &[(u32, Delivery)]| -> Vec { + d.iter() + .map(|(s, d)| match d { + Delivery::Progress(at) => format!("{s}:p{at}"), + Delivery::End => format!("{s}:end"), + _ => format!("{s}:other"), + }) + .collect() + }; + let message = || { + Delivery::Item(Item::Message(Message { + pts: 0, + data: Vec::new(), + })) + }; + let kept = superseded_progress_dropped(vec![ + (1, Delivery::Progress(1)), + (1, Delivery::Progress(2)), + (2, Delivery::Progress(2)), + (1, message()), + (1, Delivery::Progress(3)), + (2, Delivery::Progress(3)), + (1, Delivery::Progress(4)), + (1, Delivery::End), + ]); + assert_eq!( + marks(&kept), + ["1:p2", "1:other", "2:p3", "1:p4", "1:end"], + "each stream's marks up to the next thing on it, or the end, are the last of them" + ); + } } diff --git a/sidecar/ffrwd-wasm/src/tick.rs b/sidecar/ffrwd-wasm/src/tick.rs index 2e11b36..eaf9e30 100644 --- a/sidecar/ffrwd-wasm/src/tick.rs +++ b/sidecar/ffrwd-wasm/src/tick.rs @@ -571,6 +571,9 @@ pub struct Assembler { /// how many, and their bytes, which a timing clock carries none of. audio_buffer: Vec, audio_buffered: usize, + /// How many samples the packet merged last brought, for how many + /// packets the samples not yet consumed are. + audio_packet: usize, audio_pts: Option, /// Whether the last tick has been made. done: bool, @@ -893,6 +896,7 @@ impl Assembler { packets_time: None, audio_buffer: Vec::new(), audio_buffered: 0, + audio_packet: 0, audio_pts: None, done: false, made_end: None, @@ -1007,6 +1011,21 @@ impl Assembler { Some(self.inputs.iter().find(|i| i.id == id)?.queue.len()) } + /// How many packets wait on an audio clock bound to stream `id`, those + /// not yet looked at and those whose samples no tick has taken yet; + /// None where `id` is not an audio clock. + pub fn audio_backlog(&self, id: u32) -> Option { + let ClockState::Input(clock) = self.clock else { + return None; + }; + let input = &self.inputs[clock]; + if input.id != id || input.kind != PortKind::Audio { + return None; + } + let buffered = self.audio_buffered.div_ceil(self.audio_packet.max(1)); + Some(input.queue.len() + buffered) + } + /// How many more frames the hold input bound to stream `id` takes before /// its buffer is full, whoever feeds it; None where `id` is not held. pub fn held_room(&self, id: u32) -> Option { @@ -1354,7 +1373,9 @@ impl Assembler { if self.audio_buffered == 0 { self.audio_pts = Some(frame.pts); } - self.audio_buffered += sample_count(frame, audio.sample_len()); + let samples = sample_count(frame, audio.sample_len()); + self.audio_buffered += samples; + self.audio_packet = samples; self.audio_buffer.extend_from_slice(frame.data.host()); input.queue.pop_front(); } diff --git a/sidecar/ffrwd-wasm/tests/live.rs b/sidecar/ffrwd-wasm/tests/live.rs index 3577330..f557798 100644 --- a/sidecar/ffrwd-wasm/tests/live.rs +++ b/sidecar/ffrwd-wasm/tests/live.rs @@ -1588,6 +1588,50 @@ fn a_programme_without_a_feeder_passes_through_untouched() { assert_eq!(out.clock.len(), 3); } +/// A node clocked by the programme's sound a sample a tick (48000 ticks a +/// second, 1024 of them settled by each packet), on the same input as the +/// programme's picture, as foundmeat's listener on the programme's sound +/// is: every one of its ticks runs, and the programme leaves in its own +/// time, the reader of the input not held back by the node's ticks. +#[test] +fn a_node_clocked_a_sample_a_tick_on_the_programmes_sound_keeps_the_programme_in_real_time() { + let _serial = serial(); + let program = Program::new(60, 3.0); + let chain = "[v=0:v]inset=port={port}:lead=0.5[v=o];[a=0:a]shape_window=window=1[cues=c]"; + let (mut run, port) = start( + "sample-clock", + &program, + "4", + &["inset", "shape_window"], + chain, + ); + wait_for_port(&mut run, port); + let started = Instant::now(); + let out = finish(run); + let took = started.elapsed().as_secs_f64(); + assert_eq!(out.frames.len() as i64, program.frames(), "{}", out.stderr); + // The programme's sound is written in whole packets of 1024. + let samples = (program.seconds * RATE as f64) as usize / 1024 * 1024; + assert_eq!( + out.clock.len(), + samples, + "a tick for every sample of the programme's sound" + ); + assert!( + out.clock + .iter() + .enumerate() + .all(|(k, cue)| cue["text"] == format!("window {k}")), + "the ticks ran in order" + ); + assert!( + took < program.seconds + 1.5, + "the programme's {} s left in {took:.2} s\n{}", + program.seconds, + out.stderr + ); +} + #[test] fn recipe_149_runs_with_a_feed_and_without_one_at_every_jobs() { let _serial = serial(); diff --git a/sidecar/runtime/src/gpu/convert.rs b/sidecar/runtime/src/gpu/convert.rs index 6604fe3..0108781 100644 --- a/sidecar/runtime/src/gpu/convert.rs +++ b/sidecar/runtime/src/gpu/convert.rs @@ -43,7 +43,7 @@ use wgpu_core::id::{BindGroupId, BindGroupLayoutId, BufferId, TextureViewId}; use wgpu_core::id::{PipelineLayoutId, RenderPipelineId, ShaderModuleId}; use wgpu_types::TextureFormat; -use super::frames::{create_plane, Plane}; +use super::frames::{create_plane, release_plane, Plane}; use crate::profile::PROFILE; const SHADER: &str = r#" @@ -834,7 +834,7 @@ fn render( Ok(()) => Ok(planes), Err(e) => { for plane in &planes { - global.texture_drop(plane.texture); + release_plane(global, plane.texture); } Err(e) } diff --git a/sidecar/runtime/src/gpu/frames.rs b/sidecar/runtime/src/gpu/frames.rs index e9a4c8b..8b44a20 100644 --- a/sidecar/runtime/src/gpu/frames.rs +++ b/sidecar/runtime/src/gpu/frames.rs @@ -208,7 +208,7 @@ impl GpuFrame { Ok(plane) => plane, Err(e) => { for made in &planes { - global.texture_drop(made.texture); + release_plane(&global, made.texture); } return Err(e); } @@ -233,7 +233,7 @@ impl GpuFrame { planes.push(plane); if let Err(e) = written { for made in &planes { - global.texture_drop(made.texture); + release_plane(&global, made.texture); } bail!("uploading a {w}x{h} {format:?} plane: {e}"); } @@ -505,6 +505,54 @@ fn read_plane( result } +/// The size and format of a texture [`create_plane`] made. +type PlaneKey = (u32, u32, TextureFormat); + +/// The textures [`create_plane`] made, and those of them no frame holds, +/// kept for the next plane of the same size and format. Making and +/// destroying a texture costs the driver an allocation each time, at every +/// frame of a conversion or an upload. A texture used again is written +/// after every use queued before it, since wgpu orders the uses of a +/// texture on its queue. Only textures made here are kept, so a module's +/// texture a frame owns is never handed out as a plane. +#[derive(Default)] +struct Recycled { + made: std::collections::HashMap, + free: std::collections::HashMap>, +} + +/// How many free textures of one size and format are kept. +const KEPT: usize = 4; + +fn recycled() -> std::sync::MutexGuard<'static, Recycled> { + static RECYCLED: OnceLock> = OnceLock::new(); + RECYCLED + .get_or_init(Mutex::default) + .lock() + .unwrap_or_else(|e| e.into_inner()) +} + +/// Lets go of `texture`, a plane a frame owned: kept for a later plane +/// where [`create_plane`] made it and fewer than [`KEPT`] of its kind are +/// free, dropped otherwise. +pub(super) fn release_plane(global: &Global, texture: TextureId) { + { + let mut recycled = recycled(); + if let Some(key) = recycled.made.get(&texture).copied() { + let free = recycled.free.entry(key).or_default(); + if free.len() < KEPT { + free.push(texture); + return; + } + recycled.made.remove(&texture); + } + } + global.texture_drop(texture); +} + +/// A texture for one plane of a frame: a free one of the same size and +/// format where one is kept ([`release_plane`]), else a new one. Every +/// caller writes each texel before anything reads it. pub(super) fn create_plane( global: &Global, device: &SharedDevice, @@ -513,6 +561,18 @@ pub(super) fn create_plane( format: TextureFormat, bytes_per_pixel: u32, ) -> Result { + let key = (width, height, format); + let plane = |texture| Plane { + texture, + aspect: TextureAspect::All, + width, + height, + format, + bytes_per_pixel, + }; + if let Some(texture) = recycled().free.get_mut(&key).and_then(Vec::pop) { + return Ok(plane(texture)); + } let desc = wgpu_types::TextureDescriptor { label: Some(Cow::Borrowed("ffrwd frame")), size: Extent3d { @@ -532,14 +592,8 @@ pub(super) fn create_plane( global.texture_drop(texture); bail!("a {width}x{height} {format:?} texture: {e}"); } - Ok(Plane { - texture, - aspect: TextureAspect::All, - width, - height, - format, - bytes_per_pixel, - }) + recycled().made.insert(texture, key); + Ok(plane(texture)) } /// Copies one plane into `buffer`, waits for it, and appends its rows, @@ -630,7 +684,7 @@ fn copy_out( impl Drop for GpuFrame { fn drop(&mut self) { for texture in &self.owned { - self.global.texture_drop(*texture); + release_plane(&self.global, *texture); } #[cfg(any(windows, target_os = "macos", target_os = "linux"))] if let Some(wrapped) = self