Skip to main content

alpha2_audio/
graph.rs

1#![allow(clippy::needless_range_loop)]
2
3use crate::{ChannelLayout, ProcessCtx, Processor, StateBlob, StateError};
4
5/// Why a [`ProcessorChain`] refuses a processor.
6#[derive(Copy, Clone, Debug, Eq, PartialEq)]
7#[non_exhaustive]
8pub enum ChainError {
9    /// The processor reads a second bus ([`Processor::inputs_per_output`]), which a chain cannot
10    /// supply.
11    MultipleInputs,
12}
13
14impl core::fmt::Display for ChainError {
15    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
16        formatter.write_str(match self {
17            Self::MultipleInputs => {
18                "a processor chain takes only processors that read one input per output"
19            }
20        })
21    }
22}
23
24impl std::error::Error for ChainError {}
25
26/// Where and why a [`ProcessorChain`] refused a saved state. The chain is unchanged.
27#[derive(Copy, Clone, Debug, Eq, PartialEq)]
28#[non_exhaustive]
29pub struct ChainStateError {
30    /// The refusing node's index; `None` for the chain's own node count, framing or bypass
31    /// record.
32    pub node: Option<usize>,
33    /// Why.
34    pub error: StateError,
35}
36
37impl ChainStateError {
38    const fn chain(error: StateError) -> Self {
39        Self { node: None, error }
40    }
41}
42
43impl core::fmt::Display for ChainStateError {
44    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
45        match self.node {
46            Some(node) => write!(formatter, "node {node}: {}", self.error),
47            None => write!(formatter, "the chain: {}", self.error),
48        }
49    }
50}
51
52impl std::error::Error for ChainStateError {
53    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
54        Some(&self.error)
55    }
56}
57
58/// Where and why an [`AudioGraph`] refused a saved state. The graph is unchanged.
59#[derive(Copy, Clone, Debug, Eq, PartialEq)]
60#[non_exhaustive]
61pub enum GraphStateError {
62    /// The state was saved by a graph with another number of buses.
63    Topology,
64    /// A bus's clip chain refused its part.
65    ClipChain {
66        /// The bus.
67        bus: usize,
68        /// Where in the chain, and why.
69        error: ChainStateError,
70    },
71    /// A track bus refused its part.
72    TrackBus {
73        /// The bus.
74        bus: usize,
75        /// Where in the chain, and why.
76        error: ChainStateError,
77    },
78    /// The master chain refused its part.
79    Master(ChainStateError),
80    /// A bus's alignment state does not fit because bus latencies differ from the saving graph.
81    Alignment {
82        /// The bus.
83        bus: usize,
84        /// Why.
85        error: StateError,
86    },
87}
88
89impl core::fmt::Display for GraphStateError {
90    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
91        match self {
92            Self::Topology => {
93                formatter.write_str("the saved state was made by a graph with other buses")
94            }
95            Self::ClipChain { bus, error } => write!(formatter, "clip chain {bus}, {error}"),
96            Self::TrackBus { bus, error } => write!(formatter, "track bus {bus}, {error}"),
97            Self::Master(error) => write!(formatter, "master, {error}"),
98            Self::Alignment { bus, error } => write!(formatter, "bus {bus} alignment: {error}"),
99        }
100    }
101}
102
103impl std::error::Error for GraphStateError {}
104
105/// How far a chain has been prepared since its processors last changed.
106#[derive(Copy, Clone, Debug, Eq, PartialEq)]
107enum Prepared {
108    /// Not prepared since the last push; processing passes input through.
109    No,
110    /// By [`ProcessorChain::prepare_nodes`]; only [`ProcessorChain::process_in_place`] runs
111    /// nodes.
112    Nodes,
113    /// By [`ProcessorChain::prepare`].
114    Full,
115}
116
117/// Ordered processor chain with all realtime scratch allocated by `prepare`.
118///
119/// Until prepared (initially and after each `push`), a chain passes its input through.
120///
121/// Each node's context is moved back by the latency of the nodes before it, so sample-addressed
122/// nodes (fades, automation) after a lookahead act on the audio they receive.
123pub struct ProcessorChain {
124    processors: Vec<Box<dyn Processor>>,
125    layout: ChannelLayout,
126    scratch_a: [Vec<f32>; 2],
127    scratch_b: [Vec<f32>; 2],
128    bypass: bool,
129    prepared: Prepared,
130    max_block: usize,
131    /// Whether [`Self::dry`] is kept, which starts with the first [`Self::set_bypass`].
132    bypassable: bool,
133    /// The last [`Self::latency_samples`] of input, played while bypassed.
134    dry: Delay,
135}
136
137impl ProcessorChain {
138    /// Creates an empty chain for a validated layout.
139    #[must_use]
140    pub fn new(layout: ChannelLayout) -> Self {
141        Self {
142            processors: Vec::new(),
143            layout,
144            scratch_a: [Vec::new(), Vec::new()],
145            scratch_b: [Vec::new(), Vec::new()],
146            bypass: false,
147            prepared: Prepared::No,
148            max_block: 0,
149            bypassable: false,
150            dry: Delay::default(),
151        }
152    }
153    /// Adds a processor outside the realtime path. The chain passes input through until prepared
154    /// again.
155    ///
156    /// # Errors
157    /// Returns [`ChainError::MultipleInputs`] for a processor that reads more than one input per
158    /// output.
159    pub fn push(&mut self, processor: impl Processor + 'static) -> Result<(), ChainError> {
160        if processor.inputs_per_output() != 1 {
161            return Err(ChainError::MultipleInputs);
162        }
163        self.processors.push(Box::new(processor));
164        self.prepared = Prepared::No;
165        Ok(())
166    }
167    /// Sets or clears bypass, outside the realtime path.
168    ///
169    /// A bypassed chain plays its input delayed by [`Self::latency_samples`] so hosts stay
170    /// aligned, and keeps running its nodes so ending the bypass resumes on current audio. The
171    /// delayed input record starts with the first call; until it fills, the gap plays silence.
172    /// Chains never bypassed keep no record.
173    pub fn set_bypass(&mut self, bypass: bool) {
174        self.bypass = bypass;
175        if !self.bypassable {
176            self.bypassable = true;
177            if self.prepared != Prepared::No {
178                self.dry
179                    .resize(self.delay_samples(), self.layout.channels());
180            }
181        }
182        // `process_in_place` needs the chain's own scratch while bypassed.
183        if bypass && self.prepared == Prepared::Nodes {
184            self.allocate_scratch(false);
185        }
186    }
187    /// Whether [`Self::process`] would only copy input to output (empty or unprepared), so a host
188    /// may skip it.
189    #[must_use]
190    pub fn passes_through(&self) -> bool {
191        self.processors.is_empty() || self.prepared == Prepared::No
192    }
193    /// Whether the chain has been prepared since its processors last changed.
194    #[must_use]
195    pub fn is_prepared(&self) -> bool {
196        self.prepared != Prepared::No
197    }
198    /// Allocates scratch and prepares every node.
199    pub fn prepare(&mut self, sample_rate: u32, max_block: usize) {
200        for processor in &mut self.processors {
201            processor.prepare(sample_rate, max_block);
202        }
203        // At least one, so any block length can run in pieces.
204        self.max_block = max_block.max(1);
205        self.prepared = Prepared::Full;
206        if self.processors.is_empty() {
207            self.scratch_a.iter_mut().for_each(Vec::clear);
208            self.scratch_b.iter_mut().for_each(Vec::clear);
209            self.dry.resize(0, 0);
210            return;
211        }
212        self.allocate_scratch(true);
213        self.dry
214            .resize(self.delay_samples(), self.layout.channels());
215    }
216    /// Prepares the nodes without allocating chain scratch, for chains run only through
217    /// [`Self::process_in_place`]. [`Self::process`] passes input through until [`Self::prepare`].
218    pub fn prepare_nodes(&mut self, sample_rate: u32, max_block: usize) {
219        for processor in &mut self.processors {
220            processor.prepare(sample_rate, max_block);
221        }
222        self.max_block = max_block.max(1);
223        for channel in self.scratch_a.iter_mut().chain(&mut self.scratch_b) {
224            *channel = Vec::new();
225        }
226        self.prepared = Prepared::Nodes;
227        if self.bypass {
228            self.allocate_scratch(false);
229        }
230        self.dry
231            .resize(self.delay_samples(), self.layout.channels());
232    }
233    fn allocate_scratch(&mut self, both: bool) {
234        for channel in 0..self.layout.channels() {
235            self.scratch_a[channel].resize(self.max_block, 0.0);
236            if both {
237                self.scratch_b[channel].resize(self.max_block, 0.0);
238            }
239        }
240    }
241    fn delay_samples(&self) -> usize {
242        if self.bypassable {
243            usize::try_from(self.latency_samples()).unwrap_or(usize::MAX)
244        } else {
245            0
246        }
247    }
248    /// Processes the first `frames` samples of each `buffer` channel in place without
249    /// allocating. `spare` holds intermediate node output and may swap allocations with `buffer`;
250    /// both need `frames` samples per layout channel.
251    pub fn process_in_place(
252        &mut self,
253        ctx: &ProcessCtx,
254        buffer: &mut [Vec<f32>; 2],
255        spare: &mut [Vec<f32>; 2],
256        frames: usize,
257    ) {
258        let channels = self.layout.channels();
259        if self.processors.is_empty() || self.prepared == Prepared::No {
260            return;
261        }
262        let frames = buffer
263            .iter()
264            .take(channels)
265            .chain(spare.iter().take(channels))
266            .map(Vec::len)
267            .fold(frames, usize::min);
268        if !self.bypass {
269            for channel in 0..channels {
270                self.dry.remember(channel, &buffer[channel][..frames]);
271            }
272            self.dry.advance(frames);
273            let mut upstream = 0_u32;
274            for processor in &mut self.processors {
275                let node_ctx = ctx.offset_by(-i64::from(upstream));
276                process_processor(
277                    processor.as_mut(),
278                    &node_ctx,
279                    channels,
280                    (buffer, 0..frames),
281                    (spare, 0..frames),
282                );
283                core::mem::swap(buffer, spare);
284                upstream = upstream.saturating_add(processor.latency_samples());
285            }
286            return;
287        }
288        // Bypassed: nodes alternate between `spare` and the chain scratch, in scratch-sized
289        // pieces, and the buffer becomes the delayed input.
290        let piece = self.scratch_a[0].len().max(1);
291        let mut start = 0;
292        while start < frames {
293            let end = (start + piece).min(frames);
294            let count = end - start;
295            let piece_ctx = ctx.offset_by(i64::try_from(start).unwrap_or(i64::MAX));
296            let mut upstream = 0_u32;
297            for (index, processor) in self.processors.iter_mut().enumerate() {
298                let node_ctx = piece_ctx.offset_by(-i64::from(upstream));
299                if index == 0 {
300                    process_processor(
301                        processor.as_mut(),
302                        &node_ctx,
303                        channels,
304                        (buffer, start..end),
305                        (spare, start..end),
306                    );
307                } else if index % 2 == 1 {
308                    process_processor(
309                        processor.as_mut(),
310                        &node_ctx,
311                        channels,
312                        (spare, start..end),
313                        (&mut self.scratch_a, 0..count),
314                    );
315                } else {
316                    process_processor(
317                        processor.as_mut(),
318                        &node_ctx,
319                        channels,
320                        (&self.scratch_a, 0..count),
321                        (spare, start..end),
322                    );
323                }
324                upstream = upstream.saturating_add(processor.latency_samples());
325            }
326            for channel in 0..channels {
327                self.dry
328                    .delay_in_place(channel, &mut buffer[channel][start..end]);
329            }
330            self.dry.advance(count);
331            start = end;
332        }
333    }
334    /// Sum of serial node latencies; zero while unprepared.
335    #[must_use]
336    pub fn latency_samples(&self) -> u32 {
337        if self.prepared == Prepared::No {
338            return 0;
339        }
340        self.processors.iter().fold(0_u32, |total, processor| {
341            total.saturating_add(processor.latency_samples())
342        })
343    }
344    /// Sum of serial processor settling horizons, excluding their delays.
345    #[must_use]
346    pub fn history_samples(&self) -> u32 {
347        self.processors.iter().fold(0_u32, |total, processor| {
348            total.saturating_add(processor.history_samples())
349        })
350    }
351    /// Clears all histories.
352    pub fn reset(&mut self) {
353        for processor in &mut self.processors {
354            processor.reset();
355        }
356        for channel in &mut self.scratch_a {
357            channel.fill(0.0);
358        }
359        for channel in &mut self.scratch_b {
360            channel.fill(0.0);
361        }
362        self.dry.reset();
363    }
364    /// Processes one block without allocating, covering the shortest input or output channel.
365    /// Blocks longer than `max_block` run in pieces. With missing layout channels the outputs are
366    /// silenced.
367    pub fn process(&mut self, ctx: &ProcessCtx, inputs: &[&[f32]], outputs: &mut [&mut [f32]]) {
368        let channels = self.layout.channels();
369        if inputs.len() < channels || outputs.len() < channels {
370            for output in outputs.iter_mut() {
371                output.fill(0.0);
372            }
373            return;
374        }
375        let frames = inputs
376            .iter()
377            .take(channels)
378            .map(|v| v.len())
379            .chain(outputs.iter().take(channels).map(|v| v.len()))
380            .min()
381            .unwrap_or(0);
382        if self.processors.is_empty() || self.prepared != Prepared::Full {
383            for channel in 0..channels {
384                outputs[channel][..frames].copy_from_slice(&inputs[channel][..frames]);
385            }
386            return;
387        }
388        let piece = self.scratch_a[0].len();
389        let mut start = 0;
390        while start < frames {
391            let end = (start + piece).min(frames);
392            let count = end - start;
393            for channel in 0..channels {
394                self.scratch_a[channel][..count].copy_from_slice(&inputs[channel][start..end]);
395            }
396            let piece_ctx = ctx.offset_by(i64::try_from(start).unwrap_or(i64::MAX));
397            let mut upstream = 0_u32;
398            for processor in &mut self.processors {
399                let node_ctx = piece_ctx.offset_by(-i64::from(upstream));
400                process_processor(
401                    processor.as_mut(),
402                    &node_ctx,
403                    channels,
404                    (&self.scratch_a, 0..count),
405                    (&mut self.scratch_b, 0..count),
406                );
407                core::mem::swap(&mut self.scratch_a, &mut self.scratch_b);
408                upstream = upstream.saturating_add(processor.latency_samples());
409            }
410            for channel in 0..channels {
411                if self.bypass {
412                    self.dry.delay(
413                        channel,
414                        &inputs[channel][start..end],
415                        &mut outputs[channel][start..end],
416                    );
417                } else {
418                    self.dry.remember(channel, &inputs[channel][start..end]);
419                    outputs[channel][start..end].copy_from_slice(&self.scratch_a[channel][..count]);
420                }
421            }
422            self.dry.advance(count);
423            start = end;
424        }
425    }
426    /// Serializes each processor state with length framing, then the bypass record.
427    #[must_use]
428    pub fn save_state(&self) -> StateBlob {
429        let mut bytes = Vec::new();
430        bytes.extend_from_slice(
431            &(u32::try_from(self.processors.len()).unwrap_or(u32::MAX)).to_le_bytes(),
432        );
433        for state in self
434            .processors
435            .iter()
436            .map(|processor| processor.save_state())
437            .chain(core::iter::once(self.dry.save_state()))
438        {
439            bytes.extend_from_slice(
440                &(u64::try_from(state.as_bytes().len()).unwrap_or(u64::MAX)).to_le_bytes(),
441            );
442            bytes.extend_from_slice(state.as_bytes());
443        }
444        StateBlob::new(bytes)
445    }
446    /// Restores a state saved by a chain of the same nodes and preparation, atomically: if a node
447    /// refuses, earlier nodes are rolled back.
448    ///
449    /// A missing bypass record leaves this chain's empty; a record this chain does not keep is
450    /// dropped.
451    ///
452    /// # Errors
453    /// Returns where and why the state was refused.
454    pub fn load_state(&mut self, state: &StateBlob) -> Result<(), ChainStateError> {
455        let (nodes, record) = Self::split_state(state.as_bytes(), self.processors.len())
456            .ok_or(ChainStateError::chain(StateError::Mismatch))?;
457        let record = self
458            .dry
459            .check_record(&record)
460            .map_err(ChainStateError::chain)?;
461        let snapshot = self
462            .processors
463            .iter()
464            .map(|processor| processor.save_state())
465            .collect::<Vec<_>>();
466        for (index, part) in nodes.iter().enumerate() {
467            if let Err(error) = self.processors[index].load_state(part) {
468                undo(&mut self.processors[..index], &snapshot);
469                return Err(ChainStateError {
470                    node: Some(index),
471                    error,
472                });
473            }
474        }
475        self.dry.load_record(record);
476        Ok(())
477    }
478    /// The [`Processor::save_state`] of the node at `index`, for carrying history across a chain
479    /// rebuild with [`Self::load_node_state`].
480    #[must_use]
481    pub fn node_state(&self, index: usize) -> Option<StateBlob> {
482        self.processors
483            .get(index)
484            .map(|processor| processor.save_state())
485    }
486    /// Restores the node at `index` from [`Self::node_state`], atomically.
487    ///
488    /// # Errors
489    /// Returns [`StateError::Mismatch`] past the last node, or the node's own refusal.
490    pub fn load_node_state(&mut self, index: usize, state: &StateBlob) -> Result<(), StateError> {
491        self.processors
492            .get_mut(index)
493            .ok_or(StateError::Mismatch)?
494            .load_state(state)
495    }
496    /// Splits a chain state into `count` node states and the bypass record.
497    fn split_state(bytes: &[u8], count: usize) -> Option<(Vec<StateBlob>, StateBlob)> {
498        let saved = u32::from_le_bytes(bytes.get(0..4)?.try_into().ok()?);
499        if usize::try_from(saved).ok()? != count {
500            return None;
501        }
502        let mut offset = 4;
503        let mut parts = Vec::with_capacity(count + 1);
504        for _ in 0..=count {
505            let length = usize::try_from(u64::from_le_bytes(
506                bytes.get(offset..offset + 8)?.try_into().ok()?,
507            ))
508            .ok()?;
509            let end = (offset + 8).checked_add(length)?;
510            parts.push(StateBlob::new(bytes.get(offset + 8..end)?.to_vec()));
511            offset = end;
512        }
513        let record = parts.pop()?;
514        (offset == bytes.len()).then_some((parts, record))
515    }
516}
517
518/// Rolls `processors` back to their own `snapshot`; a processor that refuses its own state is
519/// reset.
520fn undo(processors: &mut [Box<dyn Processor>], snapshot: &[StateBlob]) {
521    for (processor, state) in processors.iter_mut().zip(snapshot) {
522        if processor.load_state(state).is_err() {
523            processor.reset();
524        }
525    }
526}
527
528/// Runs `processor` over `range` of `input` into `output` on the first `channels`.
529fn process_processor(
530    processor: &mut dyn Processor,
531    ctx: &ProcessCtx,
532    channels: usize,
533    (input, from): (&[Vec<f32>; 2], core::ops::Range<usize>),
534    (output, to): (&mut [Vec<f32>; 2], core::ops::Range<usize>),
535) {
536    match channels {
537        1 => processor.process(ctx, &[&input[0][from]], &mut [&mut output[0][to]]),
538        2 => {
539            let (left, right) = output.split_at_mut(1);
540            processor.process(
541                ctx,
542                &[&input[0][from.clone()], &input[1][from]],
543                &mut [&mut left[0][to.clone()], &mut right[0][to]],
544            );
545        }
546        _ => {}
547    }
548}
549
550/// A fixed per-channel ring delay with a shared cursor: process each channel, then call
551/// [`Self::advance`] once.
552#[derive(Clone, Debug, Default)]
553struct Delay {
554    lines: [Vec<f32>; 2],
555    cursor: usize,
556}
557
558impl Delay {
559    /// Empties the delay and makes it `samples` long on `channels` channels.
560    fn resize(&mut self, samples: usize, channels: usize) {
561        for (channel, line) in self.lines.iter_mut().enumerate() {
562            line.clear();
563            if channel < channels {
564                line.resize(samples, 0.0);
565            } else {
566                line.shrink_to_fit();
567            }
568        }
569        self.cursor = 0;
570    }
571    fn reset(&mut self) {
572        for line in &mut self.lines {
573            line.fill(0.0);
574        }
575        self.cursor = 0;
576    }
577    fn delay(&mut self, channel: usize, input: &[f32], output: &mut [f32]) {
578        let line = &mut self.lines[channel];
579        if line.is_empty() {
580            output.copy_from_slice(input);
581            return;
582        }
583        let mut cursor = self.cursor;
584        for (input, output) in input.iter().zip(output.iter_mut()) {
585            *output = line[cursor];
586            line[cursor] = *input;
587            cursor += 1;
588            if cursor == line.len() {
589                cursor = 0;
590            }
591        }
592    }
593    fn delay_in_place(&mut self, channel: usize, samples: &mut [f32]) {
594        let line = &mut self.lines[channel];
595        if line.is_empty() {
596            return;
597        }
598        let mut cursor = self.cursor;
599        for sample in samples {
600            core::mem::swap(&mut line[cursor], sample);
601            cursor += 1;
602            if cursor == line.len() {
603                cursor = 0;
604            }
605        }
606    }
607    /// Records `input` as [`Self::delay`] would, without writing it out.
608    fn remember(&mut self, channel: usize, input: &[f32]) {
609        let line = &mut self.lines[channel];
610        let length = line.len();
611        if length == 0 {
612            return;
613        }
614        let skip = input.len().saturating_sub(length);
615        let mut cursor = (self.cursor + skip % length) % length;
616        for &sample in &input[skip..] {
617            line[cursor] = sample;
618            cursor += 1;
619            if cursor == length {
620                cursor = 0;
621            }
622        }
623    }
624    /// Moves every channel on by `frames` samples.
625    fn advance(&mut self, frames: usize) {
626        let length = self.lines[0].len();
627        if length != 0 {
628            self.cursor = (self.cursor + frames % length) % length;
629        }
630    }
631    fn save_state(&self) -> StateBlob {
632        let mut bytes = Vec::with_capacity(8 + self.lines.iter().map(Vec::len).sum::<usize>() * 4);
633        bytes.extend_from_slice(&(u64::try_from(self.cursor).unwrap_or(0)).to_le_bytes());
634        for value in self.lines.iter().flatten() {
635            bytes.extend_from_slice(&value.to_le_bytes());
636        }
637        StateBlob::new(bytes)
638    }
639    /// The cursor of a state saved by a delay of the same length and channels. Any sample values
640    /// are accepted.
641    fn check(&self, state: &StateBlob) -> Result<usize, StateError> {
642        let bytes = state.as_bytes();
643        let samples = self.lines.iter().map(Vec::len).sum::<usize>();
644        if bytes.len() != 8 + samples * 4 {
645            return Err(StateError::Mismatch);
646        }
647        let cursor = usize::try_from(u64::from_le_bytes(
648            bytes[..8].try_into().expect("fixed range"),
649        ))
650        .map_err(|_| StateError::Invalid)?;
651        if cursor >= self.lines[0].len().max(1) {
652            return Err(StateError::Invalid);
653        }
654        Ok(cursor)
655    }
656    /// Restores a state [`Self::check`] took, at the cursor it returned.
657    fn load(&mut self, state: &StateBlob, cursor: usize) {
658        let values = state.as_bytes()[8..]
659            .as_chunks::<4>()
660            .0
661            .iter()
662            .map(|word| f32::from_le_bytes(*word));
663        for (target, value) in self.lines.iter_mut().flatten().zip(values) {
664            *target = value;
665        }
666        self.cursor = cursor;
667    }
668    /// How a saved bypass record loads ([`Self::load_record`]): whole when both keep one, emptied
669    /// when only this chain does, skipped when this chain keeps none.
670    fn check_record<'a>(&self, state: &'a StateBlob) -> Result<Record<'a>, StateError> {
671        if self.lines[0].is_empty() {
672            Ok(Record::Ignore)
673        } else if state.as_bytes().len() == 8 {
674            Ok(Record::Empty)
675        } else {
676            Ok(Record::Load(state, self.check(state)?))
677        }
678    }
679    fn load_record(&mut self, record: Record<'_>) {
680        match record {
681            Record::Ignore => {}
682            Record::Empty => self.reset(),
683            Record::Load(state, cursor) => self.load(state, cursor),
684        }
685    }
686}
687
688/// What a chain's bypass record does with a saved one.
689#[derive(Copy, Clone)]
690enum Record<'a> {
691    Ignore,
692    Empty,
693    Load(&'a StateBlob, usize),
694}
695
696/// Clip chain -> track bus -> master bus graph with deterministic latency aggregation.
697///
698/// Each bus is delayed to the longest clip-and-track latency before the mix, so output is input
699/// delayed by exactly [`Self::latency_samples`] regardless of where lookahead sits.
700pub struct AudioGraph {
701    layout: ChannelLayout,
702    clip_chains: Vec<ProcessorChain>,
703    track_buses: Vec<ProcessorChain>,
704    master: ProcessorChain,
705    alignment: Vec<Delay>,
706    mix: [Vec<f32>; 2],
707    bus_output: [Vec<f32>; 2],
708    track_output: [Vec<f32>; 2],
709    max_block: usize,
710}
711
712/// Complete graph history captured between offline chunks.
713#[derive(Clone, Debug, Default, Eq, PartialEq)]
714pub struct AudioGraphState {
715    clip_chains: Vec<StateBlob>,
716    track_buses: Vec<StateBlob>,
717    master: StateBlob,
718    alignment: Vec<StateBlob>,
719}
720
721impl AudioGraph {
722    /// Creates one clip chain and one track bus per input bus.
723    #[must_use]
724    pub fn new(layout: ChannelLayout, buses: usize) -> Self {
725        Self {
726            layout,
727            clip_chains: (0..buses).map(|_| ProcessorChain::new(layout)).collect(),
728            track_buses: (0..buses).map(|_| ProcessorChain::new(layout)).collect(),
729            master: ProcessorChain::new(layout),
730            alignment: (0..buses).map(|_| Delay::default()).collect(),
731            mix: [Vec::new(), Vec::new()],
732            bus_output: [Vec::new(), Vec::new()],
733            track_output: [Vec::new(), Vec::new()],
734            max_block: 0,
735        }
736    }
737    /// Channel layout every bus and the master run in.
738    #[must_use]
739    pub const fn layout(&self) -> ChannelLayout {
740        self.layout
741    }
742    /// Number of input buses, each with its own clip chain and track bus.
743    #[must_use]
744    pub fn buses(&self) -> usize {
745        self.clip_chains.len()
746    }
747    /// Mutable clip effect chain of `bus`; `None` past the last bus.
748    pub fn clip_chain_mut(&mut self, bus: usize) -> Option<&mut ProcessorChain> {
749        self.clip_chains.get_mut(bus)
750    }
751    /// Mutable track effect chain of `bus`; `None` past the last bus.
752    pub fn track_bus_mut(&mut self, bus: usize) -> Option<&mut ProcessorChain> {
753        self.track_buses.get_mut(bus)
754    }
755    /// Mutable master effect chain.
756    pub fn master_mut(&mut self) -> &mut ProcessorChain {
757        &mut self.master
758    }
759    /// Prepares every allocation before realtime processing.
760    pub fn prepare(&mut self, sample_rate: u32, max_block: usize) {
761        // At least one, so any block length can run in pieces.
762        self.max_block = max_block.max(1);
763        for chain in self
764            .clip_chains
765            .iter_mut()
766            .chain(&mut self.track_buses)
767            .chain(core::iter::once(&mut self.master))
768        {
769            chain.prepare(sample_rate, self.max_block);
770        }
771        for channel in 0..2 {
772            self.mix[channel].resize(self.max_block, 0.0);
773            self.bus_output[channel].resize(self.max_block, 0.0);
774            self.track_output[channel].resize(self.max_block, 0.0);
775        }
776        let longest = self.bus_latency();
777        for bus in 0..self.alignment.len() {
778            let pad = longest - self.path_latency(bus);
779            self.alignment[bus].resize(
780                usize::try_from(pad).unwrap_or(usize::MAX),
781                self.layout.channels(),
782            );
783        }
784    }
785    /// The latency of bus `bus` through its clip chain and track bus.
786    fn path_latency(&self, bus: usize) -> u32 {
787        self.clip_chains[bus]
788            .latency_samples()
789            .saturating_add(self.track_buses[bus].latency_samples())
790    }
791    /// The longest latency of any bus through its clip chain and track bus.
792    fn bus_latency(&self) -> u32 {
793        (0..self.clip_chains.len())
794            .map(|bus| self.path_latency(bus))
795            .max()
796            .unwrap_or(0)
797    }
798    /// Maximum bus latency plus master latency.
799    #[must_use]
800    pub fn latency_samples(&self) -> u32 {
801        self.bus_latency()
802            .saturating_add(self.master.latency_samples())
803    }
804    /// Mixes flattened planar bus inputs (`bus0-L, bus0-R, bus1-L, ...`) over the shortest output
805    /// channel. Short buses are padded with silence and long blocks run in pieces. With missing
806    /// channels or no [`Self::prepare`], the outputs are silenced.
807    pub fn process(&mut self, ctx: &ProcessCtx, inputs: &[&[f32]], outputs: &mut [&mut [f32]]) {
808        let channels = self.layout.channels();
809        if inputs.len() < self.clip_chains.len() * channels
810            || outputs.len() < channels
811            || self.max_block == 0
812        {
813            for output in outputs.iter_mut() {
814                output.fill(0.0);
815            }
816            return;
817        }
818        let frames = outputs
819            .iter()
820            .take(channels)
821            .map(|v| v.len())
822            .min()
823            .unwrap_or(0);
824        let mut start = 0;
825        while start < frames {
826            let end = (start + self.max_block).min(frames);
827            let ctx = ctx.offset_by(i64::try_from(start).unwrap_or(i64::MAX));
828            self.process_piece(&ctx, inputs, outputs, start, end);
829            start = end;
830        }
831    }
832
833    /// Mixes samples `start..end` of the block, at most `max_block` of them.
834    fn process_piece(
835        &mut self,
836        ctx: &ProcessCtx,
837        inputs: &[&[f32]],
838        outputs: &mut [&mut [f32]],
839        start: usize,
840        end: usize,
841    ) {
842        let channels = self.layout.channels();
843        let frames = end - start;
844        for channel in 0..channels {
845            self.mix[channel][..frames].fill(0.0);
846        }
847        for bus in 0..self.clip_chains.len() {
848            let base = bus * channels;
849            let piece = |channel: usize| {
850                let input = inputs[base + channel];
851                &input[start.min(input.len())..end.min(input.len())]
852            };
853            let bus_inputs: [&[f32]; 2] = [piece(0), piece(channels - 1)];
854            process_chain_to_internal(
855                &mut self.clip_chains[bus],
856                ctx,
857                channels,
858                frames,
859                &bus_inputs[..channels],
860                &mut self.bus_output,
861            );
862            let bus_refs: [&[f32]; 2] =
863                [&self.bus_output[0][..frames], &self.bus_output[1][..frames]];
864            // The track bus receives audio delayed by the clip chain.
865            let track_ctx = ctx.offset_by(-i64::from(self.clip_chains[bus].latency_samples()));
866            process_chain_to_internal(
867                &mut self.track_buses[bus],
868                &track_ctx,
869                channels,
870                frames,
871                &bus_refs[..channels],
872                &mut self.track_output,
873            );
874            let alignment = &mut self.alignment[bus];
875            for channel in 0..channels {
876                alignment.delay_in_place(channel, &mut self.track_output[channel][..frames]);
877            }
878            alignment.advance(frames);
879            for channel in 0..channels {
880                for frame in 0..frames {
881                    self.mix[channel][frame] += self.track_output[channel][frame];
882                }
883            }
884        }
885        let mix_refs: [&[f32]; 2] = [&self.mix[0][..frames], &self.mix[1][..frames]];
886        // Every bus reaches the master as late as the longest.
887        let ctx = ctx.offset_by(-i64::from(self.bus_latency()));
888        let (left, right) = outputs.split_at_mut(1);
889        match channels {
890            1 => self
891                .master
892                .process(&ctx, &mix_refs[..1], &mut [&mut left[0][start..end]]),
893            _ => self.master.process(
894                &ctx,
895                &mix_refs[..2],
896                &mut [&mut left[0][start..end], &mut right[0][start..end]],
897            ),
898        }
899    }
900    /// Clears every graph node.
901    pub fn reset(&mut self) {
902        for chain in self
903            .clip_chains
904            .iter_mut()
905            .chain(&mut self.track_buses)
906            .chain(core::iter::once(&mut self.master))
907        {
908            chain.reset();
909        }
910        for alignment in &mut self.alignment {
911            alignment.reset();
912        }
913    }
914
915    /// Captures every chain's history and the alignment delays.
916    #[must_use]
917    pub fn save_state(&self) -> AudioGraphState {
918        AudioGraphState {
919            clip_chains: self
920                .clip_chains
921                .iter()
922                .map(ProcessorChain::save_state)
923                .collect(),
924            track_buses: self
925                .track_buses
926                .iter()
927                .map(ProcessorChain::save_state)
928                .collect(),
929            master: self.master.save_state(),
930            alignment: self.alignment.iter().map(Delay::save_state).collect(),
931        }
932    }
933
934    /// Restores a state saved by a graph of the same buses, chains and preparation, atomically.
935    ///
936    /// # Errors
937    /// Returns where and why the state was refused.
938    pub fn load_state(&mut self, state: &AudioGraphState) -> Result<(), GraphStateError> {
939        let buses = self.clip_chains.len();
940        if state.clip_chains.len() != buses
941            || state.track_buses.len() != buses
942            || state.alignment.len() != buses
943        {
944            return Err(GraphStateError::Topology);
945        }
946        let mut cursors = Vec::with_capacity(buses);
947        for (bus, (alignment, saved)) in self.alignment.iter().zip(&state.alignment).enumerate() {
948            cursors.push(
949                alignment
950                    .check(saved)
951                    .map_err(|error| GraphStateError::Alignment { bus, error })?,
952            );
953        }
954        let snapshot = self
955            .chains()
956            .map(ProcessorChain::save_state)
957            .collect::<Vec<_>>();
958        let saved = state
959            .clip_chains
960            .iter()
961            .chain(&state.track_buses)
962            .chain(core::iter::once(&state.master));
963        let refused =
964            self.chains_mut()
965                .zip(saved)
966                .enumerate()
967                .find_map(|(index, (chain, saved))| {
968                    chain.load_state(saved).err().map(|error| (index, error))
969                });
970        if let Some((index, error)) = refused {
971            for (chain, state) in self.chains_mut().zip(&snapshot).take(index) {
972                if chain.load_state(state).is_err() {
973                    chain.reset();
974                }
975            }
976            return Err(if index < buses {
977                GraphStateError::ClipChain { bus: index, error }
978            } else if index < 2 * buses {
979                GraphStateError::TrackBus {
980                    bus: index - buses,
981                    error,
982                }
983            } else {
984                GraphStateError::Master(error)
985            });
986        }
987        for ((alignment, saved), cursor) in
988            self.alignment.iter_mut().zip(&state.alignment).zip(cursors)
989        {
990            alignment.load(saved, cursor);
991        }
992        Ok(())
993    }
994
995    /// Every chain: the clip chains, the track buses, then the master.
996    fn chains(&self) -> impl Iterator<Item = &ProcessorChain> {
997        self.clip_chains
998            .iter()
999            .chain(&self.track_buses)
1000            .chain(core::iter::once(&self.master))
1001    }
1002
1003    fn chains_mut(&mut self) -> impl Iterator<Item = &mut ProcessorChain> {
1004        self.clip_chains
1005            .iter_mut()
1006            .chain(&mut self.track_buses)
1007            .chain(core::iter::once(&mut self.master))
1008    }
1009}
1010
1011fn process_chain_to_internal(
1012    chain: &mut ProcessorChain,
1013    ctx: &ProcessCtx,
1014    channels: usize,
1015    frames: usize,
1016    inputs: &[&[f32]],
1017    output: &mut [Vec<f32>; 2],
1018) {
1019    // Clear the whole block so a short bus is padded with silence, not the previous bus's tail.
1020    for channel in output.iter_mut().take(channels) {
1021        channel[..frames].fill(0.0);
1022    }
1023    match channels {
1024        1 => chain.process(ctx, inputs, &mut [&mut output[0][..frames]]),
1025        2 => {
1026            let (left, right) = output.split_at_mut(1);
1027            chain.process(
1028                ctx,
1029                inputs,
1030                &mut [&mut left[0][..frames], &mut right[0][..frames]],
1031            );
1032        }
1033        _ => {}
1034    }
1035}