sim_lib_audio_graph_live/
site.rs

1use std::thread;
2
3use sim_kernel::{Error, Result, Symbol};
4use sim_lib_audio_graph_core::{Processor, Transport};
5use sim_lib_stream_host::{ProcessRingPush, ProcessRingSnapshot, ProcessSharedRing};
6
7use crate::{
8    LiveGraphConfig, LiveGraphRunner, LiveProcessReport, LiveSteadyStateSnapshot,
9    realtime_local_audio_profile,
10};
11
12/// Local placement sites supported by the live audio graph backbone.
13#[derive(Clone, Copy, Debug, PartialEq, Eq)]
14pub enum LivePlacementSite {
15    /// Runs inline on the caller (cooperative coroutine).
16    Coroutine,
17    /// Runs on a dedicated worker thread.
18    Thread,
19    /// Runs directly in the host audio callback (realtime, audio-clock thread).
20    HostCallback,
21    /// Runs in a separate process behind shared rings.
22    Process,
23}
24
25/// Capacity snapshot for a placed live node.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub struct LivePlacementSnapshot {
28    site: LivePlacementSite,
29    runner: LiveSteadyStateSnapshot,
30    process_request_ring: Option<ProcessRingSnapshot>,
31    process_response_ring: Option<ProcessRingSnapshot>,
32}
33
34/// Live graph runner bound to one local placement site.
35#[derive(Debug)]
36pub struct LivePlacedNode<P> {
37    site: LivePlacementSite,
38    runner: LiveGraphRunner<P>,
39    process_request_ring: Option<ProcessSharedRing<ProcessSiteMessage>>,
40    process_response_ring: Option<ProcessSharedRing<ProcessSiteMessage>>,
41}
42
43#[derive(Clone, Copy, Debug, PartialEq)]
44enum ProcessSiteMessage {
45    Render { frames: usize, transport: Transport },
46    Rendered { frames: u32 },
47}
48
49impl LivePlacementSite {
50    /// Returns every placement site in a stable order.
51    pub const fn all() -> [Self; 4] {
52        [
53            Self::Coroutine,
54            Self::Thread,
55            Self::HostCallback,
56            Self::Process,
57        ]
58    }
59
60    /// Returns the stable wire name for this site.
61    pub const fn name(self) -> &'static str {
62        match self {
63            Self::Coroutine => "coroutine",
64            Self::Thread => "thread",
65            Self::HostCallback => "host-callback",
66            Self::Process => "process",
67        }
68    }
69
70    /// Returns the namespaced symbol for this site.
71    pub fn symbol(self) -> Symbol {
72        Symbol::qualified("stream/site", self.name())
73    }
74
75    /// Returns whether this site runs on the audio-clock thread.
76    pub const fn runs_on_audio_clock_thread(self) -> bool {
77        matches!(self, Self::HostCallback)
78    }
79
80    /// Returns whether this site communicates over a shared process ring.
81    pub const fn uses_process_ring(self) -> bool {
82        matches!(self, Self::Process)
83    }
84
85    /// Returns whether realtime-pinned processors may run at this site.
86    pub const fn allows_realtime_pin(self) -> bool {
87        self.runs_on_audio_clock_thread()
88    }
89
90    fn validate_processor(self, realtime_pin: bool) -> Result<()> {
91        if realtime_pin && !self.allows_realtime_pin() {
92            return Err(Error::Eval(format!(
93                "realtime-pinned live audio node cannot run at {} site",
94                self.name()
95            )));
96        }
97        Ok(())
98    }
99}
100
101impl<P: Processor> LivePlacedNode<P> {
102    /// Places a processor at a site, validating realtime pinning and creating
103    /// any process rings the site requires.
104    pub fn new(processor: P, config: LiveGraphConfig, site: LivePlacementSite) -> Result<Self> {
105        site.validate_processor(processor.realtime_pin())?;
106        let runner = if site.runs_on_audio_clock_thread() {
107            LiveGraphRunner::new_realtime(processor, config, &realtime_local_audio_profile())?
108        } else {
109            LiveGraphRunner::new(processor, config)?
110        };
111        let process_request_ring = site
112            .uses_process_ring()
113            .then(|| ProcessSharedRing::with_capacity(1))
114            .transpose()?;
115        let process_response_ring = site
116            .uses_process_ring()
117            .then(|| ProcessSharedRing::with_capacity(1))
118            .transpose()?;
119        Ok(Self {
120            site,
121            runner,
122            process_request_ring,
123            process_response_ring,
124        })
125    }
126
127    /// Returns the placement site.
128    pub fn site(&self) -> LivePlacementSite {
129        self.site
130    }
131
132    /// Returns a shared reference to the underlying runner.
133    pub fn runner(&self) -> &LiveGraphRunner<P> {
134        &self.runner
135    }
136
137    /// Returns a mutable reference to the underlying runner.
138    pub fn runner_mut(&mut self) -> &mut LiveGraphRunner<P> {
139        &mut self.runner
140    }
141
142    /// Processes one interleaved block, dispatching to the site's execution
143    /// strategy (inline, worker thread, or process ring).
144    pub fn process_interleaved_f32(
145        &mut self,
146        input: Option<&[f32]>,
147        output: &mut [f32],
148        frames: usize,
149        transport: Transport,
150    ) -> Result<LiveProcessReport> {
151        match self.site {
152            LivePlacementSite::Coroutine | LivePlacementSite::HostCallback => self
153                .runner
154                .process_interleaved_f32(input, output, frames, transport),
155            LivePlacementSite::Thread => {
156                run_on_worker_thread(&mut self.runner, input, output, frames, transport)
157            }
158            LivePlacementSite::Process => {
159                self.process_via_process_ring(input, output, frames, transport)
160            }
161        }
162    }
163
164    /// Captures the placed node's steady-state capacity snapshot.
165    pub fn steady_state_snapshot(&self) -> LivePlacementSnapshot {
166        LivePlacementSnapshot {
167            site: self.site,
168            runner: self.runner.steady_state_snapshot(),
169            process_request_ring: self
170                .process_request_ring
171                .as_ref()
172                .map(|ring| ring.snapshot()),
173            process_response_ring: self
174                .process_response_ring
175                .as_ref()
176                .map(|ring| ring.snapshot()),
177        }
178    }
179
180    fn process_via_process_ring(
181        &mut self,
182        input: Option<&[f32]>,
183        output: &mut [f32],
184        frames: usize,
185        transport: Transport,
186    ) -> Result<LiveProcessReport> {
187        let work = ProcessSiteMessage::Render { frames, transport };
188        {
189            let request_ring = self.process_request_ring.as_mut().ok_or_else(|| {
190                Error::Eval("process site is missing its request ring".to_owned())
191            })?;
192            accept_process_push(request_ring.try_push(work), "request")?;
193        }
194        let work = self
195            .process_request_ring
196            .as_mut()
197            .and_then(ProcessSharedRing::try_pop)
198            .ok_or_else(|| Error::Eval("process site request ring lost work".to_owned()))?;
199        let ProcessSiteMessage::Render { frames, transport } = work else {
200            return Err(Error::Eval(
201                "process site request ring received a response".to_owned(),
202            ));
203        };
204
205        let report = run_on_worker_thread(&mut self.runner, input, output, frames, transport)?;
206        {
207            let response_ring = self.process_response_ring.as_mut().ok_or_else(|| {
208                Error::Eval("process site is missing its response ring".to_owned())
209            })?;
210            accept_process_push(
211                response_ring.try_push(ProcessSiteMessage::Rendered {
212                    frames: report.frames(),
213                }),
214                "response",
215            )?;
216        }
217        let response = self
218            .process_response_ring
219            .as_mut()
220            .and_then(ProcessSharedRing::try_pop)
221            .ok_or_else(|| Error::Eval("process site response ring lost output".to_owned()))?;
222        match response {
223            ProcessSiteMessage::Rendered { frames } if frames == report.frames() => Ok(report),
224            ProcessSiteMessage::Rendered { .. } => Err(Error::Eval(
225                "process site response ring returned the wrong frame count".to_owned(),
226            )),
227            ProcessSiteMessage::Render { .. } => Err(Error::Eval(
228                "process site response ring received a request".to_owned(),
229            )),
230        }
231    }
232}
233
234impl LivePlacementSnapshot {
235    /// Returns the placement site captured by the snapshot.
236    pub fn site(&self) -> LivePlacementSite {
237        self.site
238    }
239
240    /// Returns the runner's steady-state capacity snapshot.
241    pub fn runner(&self) -> &LiveSteadyStateSnapshot {
242        &self.runner
243    }
244
245    /// Returns the request ring snapshot, if this site uses a process ring.
246    pub fn process_request_ring(&self) -> Option<ProcessRingSnapshot> {
247        self.process_request_ring
248    }
249
250    /// Returns the response ring snapshot, if this site uses a process ring.
251    pub fn process_response_ring(&self) -> Option<ProcessRingSnapshot> {
252        self.process_response_ring
253    }
254}
255
256fn run_on_worker_thread<P: Processor>(
257    runner: &mut LiveGraphRunner<P>,
258    input: Option<&[f32]>,
259    output: &mut [f32],
260    frames: usize,
261    transport: Transport,
262) -> Result<LiveProcessReport> {
263    match thread::scope(|scope| {
264        scope
265            .spawn(move || runner.process_interleaved_f32(input, output, frames, transport))
266            .join()
267    }) {
268        Ok(result) => result,
269        Err(_) => Err(Error::Eval(
270            "live audio worker thread panicked while processing".to_owned(),
271        )),
272    }
273}
274
275fn accept_process_push<T>(push: ProcessRingPush<T>, role: &str) -> Result<()> {
276    match push {
277        ProcessRingPush::Accepted => Ok(()),
278        ProcessRingPush::DroppedNewest(_) => {
279            Err(Error::Eval(format!("process site {role} ring is full")))
280        }
281        ProcessRingPush::Closed(_) => {
282            Err(Error::Eval(format!("process site {role} ring is closed")))
283        }
284    }
285}
sim_lib_audio_graph_live/site.rs

sim_lib_audio_graph_live/
site.rs
