rill_lang/
graph_engine.rs

1//! Execution engine for CompiledGraph with a FixedBuffer pool.
2//!
3//! Runs a flat vector of NodeClosures in topological order. Zero heap
4//! allocation on the real-time signal path.
5
6use std::collections::HashMap;
7use std::sync::Arc;
8
9use rill_core::buffer::Buffer;
10use rill_core::math::Transcendental;
11use rill_core::queues::CommandEnum;
12use rill_core::traits::{Algorithm, MultichannelAlgorithm, ParamValue, ProcessResult};
13use rill_core_actor::{ActorRef, Mailbox};
14
15use crate::graph_compiler::CompiledGraph;
16
17#[cfg(feature = "debug")]
18use crate::debug::{CmdStr, CommandFrame, DebugControl, ProbeSlot};
19#[cfg(feature = "debug")]
20use rill_core::queues::spsc::SpscQueue;
21#[cfg(feature = "debug")]
22use std::sync::atomic::Ordering;
23
24/// Map from parameter name to its index in the node's parameter list.
25pub type ParamMap = HashMap<String, usize>;
26
27struct PendingParam {
28    node_idx: usize,
29    param_idx: usize,
30    value: ParamValue,
31    sample_pos: Option<u64>,
32}
33
34/// Graph execution engine running a [`CompiledGraph`].
35pub struct CompiledGraphEngine<T: Transcendental, const BUF_SIZE: usize> {
36    graph: CompiledGraph<T, BUF_SIZE>,
37    pending: Vec<PendingParam>,
38    param_maps: Vec<HashMap<String, usize>>,
39    anchor_map: HashMap<String, usize>,
40    mailbox: Arc<Mailbox<CommandEnum>>,
41    actor_ref: ActorRef<CommandEnum>,
42    #[cfg(feature = "debug")]
43    pub(crate) probe_slots: Vec<std::sync::Arc<ProbeSlot>>,
44    #[cfg(feature = "debug")]
45    pub(crate) command_queue: std::sync::Arc<SpscQueue<CommandFrame, 256>>,
46    #[cfg(feature = "debug")]
47    pub(crate) debug_control: DebugControl,
48}
49
50impl<T: Transcendental, const BUF_SIZE: usize> CompiledGraphEngine<T, BUF_SIZE> {
51    /// Create a new graph engine from a compiled graph and a shared mailbox.
52    pub fn new(graph: CompiledGraph<T, BUF_SIZE>, mailbox: Arc<Mailbox<CommandEnum>>) -> Self {
53        let actor_ref = mailbox.actor_ref();
54        let anchor_map: HashMap<String, usize> = graph
55            .node_names
56            .iter()
57            .enumerate()
58            .map(|(i, name)| (name.clone(), i))
59            .collect();
60        let param_maps: Vec<HashMap<String, usize>> = graph
61            .node_param_names
62            .iter()
63            .map(|names| {
64                names
65                    .iter()
66                    .enumerate()
67                    .map(|(i, n)| (n.clone(), i))
68                    .collect()
69            })
70            .collect();
71
72        Self {
73            graph,
74            pending: Vec::new(),
75            param_maps,
76            anchor_map,
77            mailbox,
78            actor_ref,
79            #[cfg(feature = "debug")]
80            probe_slots: Vec::new(),
81            #[cfg(feature = "debug")]
82            command_queue: std::sync::Arc::new(SpscQueue::new()),
83            #[cfg(feature = "debug")]
84            debug_control: DebugControl::new(),
85        }
86    }
87
88    /// Returns the actor handle for sending control commands to the engine.
89    pub fn handle(&self) -> ActorRef<CommandEnum> {
90        self.actor_ref.clone()
91    }
92
93    /// Returns the merged parameter map for all nodes in the engine.
94    ///
95    /// The map is the first node's mapping, which covers all parameters
96    /// for single-node engines. For multi-node graphs, this returns
97    /// the first node's map only.
98    pub fn param_map(&self) -> HashMap<String, usize> {
99        self.param_maps.first().cloned().unwrap_or_default()
100    }
101
102    #[cfg(feature = "debug")]
103    /// Allocate `count` probe slots for the engine.
104    pub fn allocate_probe_slots(&mut self, count: usize) {
105        self.probe_slots = (0..count)
106            .map(|_| std::sync::Arc::new(ProbeSlot::default()))
107            .collect();
108    }
109
110    #[cfg(feature = "debug")]
111    /// Return debug state handles for external collector/debugger threads.
112    pub fn debug_state(
113        &self,
114    ) -> (
115        &[std::sync::Arc<ProbeSlot>],
116        DebugControl,
117        std::sync::Arc<SpscQueue<CommandFrame, 256>>,
118    ) {
119        (
120            &self.probe_slots,
121            self.debug_control.clone(),
122            self.command_queue.clone(),
123        )
124    }
125
126    #[cfg(feature = "debug")]
127    /// Clone probe slots, debug control, and command queue for sharing with a
128    /// collector or debugger thread.
129    pub fn clone_debug_state(
130        &self,
131    ) -> (
132        Vec<std::sync::Arc<ProbeSlot>>,
133        DebugControl,
134        std::sync::Arc<SpscQueue<CommandFrame, 256>>,
135    ) {
136        (
137            self.probe_slots.clone(),
138            self.debug_control.clone(),
139            self.command_queue.clone(),
140        )
141    }
142
143    fn drain_mailbox(&mut self) {
144        #[cfg(feature = "debug")]
145        let block_idx = self.debug_control.block_index.load(Ordering::Relaxed);
146
147        while let Some(cmd) = self.mailbox.pop() {
148            if let CommandEnum::SetParameter(ref sp) = cmd {
149                let param_name = sp.parameter.as_str();
150                #[cfg(feature = "debug")]
151                let mut applied = false;
152                if !sp.anchor.is_empty() {
153                    if let Some(&node_idx) = self.anchor_map.get(&sp.anchor) {
154                        if let Some(&idx) = self.param_maps[node_idx].get(param_name) {
155                            self.pending.push(PendingParam {
156                                node_idx,
157                                param_idx: idx,
158                                value: sp.value.clone(),
159                                sample_pos: sp.sample_pos,
160                            });
161                            #[cfg(feature = "debug")]
162                            {
163                                applied = true;
164                            }
165                        }
166                    }
167                } else {
168                    for (node_idx, map) in self.param_maps.iter().enumerate() {
169                        if let Some(&idx) = map.get(param_name) {
170                            self.pending.push(PendingParam {
171                                node_idx,
172                                param_idx: idx,
173                                value: sp.value.clone(),
174                                sample_pos: sp.sample_pos,
175                            });
176                            #[cfg(feature = "debug")]
177                            {
178                                applied = true;
179                            }
180                            break;
181                        }
182                    }
183                }
184                #[cfg(feature = "debug")]
185                if applied {
186                    let _ = self.command_queue.push(CommandFrame {
187                        block_index: block_idx,
188                        command_kind: CmdStr::new("SetParameter"),
189                        node_name: CmdStr::new(&sp.anchor),
190                        param_name: CmdStr::new(&format!("{}", sp.parameter)),
191                        value_repr: CmdStr::new(&format!("{:?}", sp.value)),
192                    });
193                }
194            }
195        }
196    }
197
198    /// Apply pending parameter updates that are due by `chunk_end`.
199    /// Preserves sample-accurate scheduling: params with `sample_pos >= chunk_end`
200    /// are deferred to the next tick.
201    pub fn apply_due_params(&mut self, chunk_end: u64) {
202        if self.pending.is_empty() {
203            return;
204        }
205        self.pending.sort_by_key(|p| p.sample_pos.unwrap_or(0));
206        let split = self
207            .pending
208            .partition_point(|p| p.sample_pos.is_none_or(|sp| sp < chunk_end));
209        if split == 0 {
210            return;
211        }
212        for p in self.pending.drain(0..split) {
213            if p.node_idx < self.graph.nodes.len() {
214                self.graph.nodes[p.node_idx].set_param(p.param_idx, &p.value);
215            }
216        }
217    }
218
219    /// Main processing tick — applies pending params, copies inputs, runs nodes, copies outputs.
220    pub fn process_tick(
221        &mut self,
222        inputs: &[&[T]],
223        outputs: &mut [&mut [T]],
224        chunk_end: u64,
225    ) -> ProcessResult<()> {
226        #[cfg(feature = "debug")]
227        {
228            self.debug_control
229                .block_index
230                .fetch_add(1, Ordering::Relaxed);
231        }
232        self.drain_mailbox();
233
234        #[cfg(feature = "debug")]
235        {
236            while self.debug_control.global_pause.load(Ordering::Acquire)
237                && !self.debug_control.global_resume.load(Ordering::Acquire)
238            {
239                std::hint::spin_loop();
240            }
241            self.debug_control
242                .global_resume
243                .store(false, Ordering::Release);
244        }
245
246        self.apply_due_params(chunk_end);
247
248        for (i, input) in inputs.iter().enumerate() {
249            if i < self.graph.inputs && i < self.graph.buffers.len() {
250                let buf = self.graph.buffers[i].as_mut_slice();
251                let n = input.len().min(buf.len());
252                buf[..n].copy_from_slice(&input[..n]);
253            }
254        }
255
256        for node in &mut self.graph.nodes {
257            node.execute(&mut self.graph.buffers)?;
258        }
259
260        for (i, output) in outputs.iter_mut().enumerate() {
261            if i < self.graph.output_mapping.len() {
262                let src = self.graph.output_mapping[i];
263                if src < self.graph.buffers.len() {
264                    let buf = self.graph.buffers[src].as_slice();
265                    let n = output.len().min(buf.len());
266                    output[..n].copy_from_slice(&buf[..n]);
267                }
268            }
269        }
270
271        Ok(())
272    }
273
274    /// Reset all buffers and nodes to initial state.
275    pub fn reset(&mut self) {
276        for buf in &mut self.graph.buffers {
277            buf.fill(T::ZERO);
278        }
279        for node in &mut self.graph.nodes {
280            node.reset();
281        }
282    }
283}
284
285impl<T: Transcendental, const BUF_SIZE: usize> Algorithm<T> for CompiledGraphEngine<T, BUF_SIZE> {
286    fn process(&mut self, input: Option<&[T]>, output: &mut [T]) -> ProcessResult<()> {
287        let bufs: &[&[T]] = if let Some(inp) = input { &[inp] } else { &[] };
288        let out_bufs: &mut [&mut [T]] = &mut [output];
289        MultichannelAlgorithm::process(self, bufs, out_bufs)
290    }
291
292    fn reset(&mut self) {
293        Self::reset(self);
294    }
295}
296
297impl<T: Transcendental, const BUF_SIZE: usize> MultichannelAlgorithm<T>
298    for CompiledGraphEngine<T, BUF_SIZE>
299{
300    fn num_inputs(&self) -> usize {
301        self.graph.inputs
302    }
303
304    fn num_outputs(&self) -> usize {
305        self.graph.outputs
306    }
307
308    fn process(&mut self, inputs: &[&[T]], outputs: &mut [&mut [T]]) -> ProcessResult<()> {
309        self.process_tick(inputs, outputs, u64::MAX)
310    }
311
312    fn reset(&mut self) {
313        Self::reset(self);
314    }
315}
rill_lang/graph_engine.rs

rill_lang/
graph_engine.rs
