ferraus_plugins/
client.rs

1//! The ferraus side of the host protocol: spawns one host process per plugin instance,
2//! sends framed JSON requests, moves audio through a memory-mapped temp file, and enforces
3//! the watchdog. A crash or timeout becomes [`ErrorKind::Plugin`] (exit code 7) naming the
4//! plugin and its path; the calling process is never affected.
5
6use crate::protocol::*;
7use base64::Engine;
8use ferraus_core::{Error, ErrorKind, Result};
9use serde::{Deserialize, Serialize};
10use std::io::{BufRead, BufReader, Write};
11use std::path::{Path, PathBuf};
12use std::process::{Child, ChildStdin, Command, Stdio};
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
15use std::sync::{Arc, Mutex};
16use std::time::{Duration, Instant};
17
18/// Default watchdog: a host that makes no progress for this long is stopped.
19pub const DEFAULT_WATCHDOG_S: f64 = 60.0;
20
21/// The watchdog from `FERRAUS_PLUGIN_WATCHDOG_S`, else 60 s.
22pub fn default_watchdog() -> Duration {
23    std::env::var("FERRAUS_PLUGIN_WATCHDOG_S")
24        .ok()
25        .and_then(|v| v.trim().parse::<f64>().ok())
26        .filter(|v| *v > 0.0 && v.is_finite())
27        .map(Duration::from_secs_f64)
28        .unwrap_or(Duration::from_secs_f64(DEFAULT_WATCHDOG_S))
29}
30
31/// Plugin binary format.
32#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize, PartialOrd, Ord)]
33#[serde(rename_all = "lowercase")]
34pub enum PluginFormat {
35    Clap,
36    Vst3,
37}
38
39impl PluginFormat {
40    pub fn as_str(self) -> &'static str {
41        match self {
42            PluginFormat::Clap => "clap",
43            PluginFormat::Vst3 => "vst3",
44        }
45    }
46
47    pub fn parse(s: &str) -> Option<PluginFormat> {
48        match s.trim().to_ascii_lowercase().as_str() {
49            "clap" => Some(PluginFormat::Clap),
50            "vst3" | "vst" => Some(PluginFormat::Vst3),
51            _ => None,
52        }
53    }
54
55    /// Format from a bundle path's extension.
56    pub fn from_path(p: &Path) -> Option<PluginFormat> {
57        let ext = p.extension()?.to_string_lossy().to_ascii_lowercase();
58        PluginFormat::parse(&ext)
59    }
60
61    /// The host executable for this format.
62    pub fn host_program(self) -> &'static str {
63        match self {
64            PluginFormat::Clap => "ferraus-plughost",
65            PluginFormat::Vst3 => "ferraus-plughost-vst3",
66        }
67    }
68}
69
70impl std::fmt::Display for PluginFormat {
71    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72        f.write_str(self.as_str())
73    }
74}
75
76/// Locate the host executable: `FERRAUS_PLUGHOST` / `FERRAUS_PLUGHOST_VST3`, then next to
77/// the running executable (and its parent, which covers `target/<profile>/deps` test
78/// binaries), then `PATH`.
79pub fn host_binary(format: PluginFormat) -> Result<PathBuf> {
80    let var = match format {
81        PluginFormat::Clap => "FERRAUS_PLUGHOST",
82        PluginFormat::Vst3 => "FERRAUS_PLUGHOST_VST3",
83    };
84    if let Some(p) = std::env::var_os(var).filter(|v| !v.is_empty()) {
85        return Ok(PathBuf::from(p));
86    }
87    let file = format!("{}{}", format.host_program(), std::env::consts::EXE_SUFFIX);
88    if let Ok(exe) = std::env::current_exe() {
89        let mut dir = exe.parent().map(|p| p.to_path_buf());
90        for _ in 0..2 {
91            if let Some(d) = &dir {
92                let c = d.join(&file);
93                if c.is_file() {
94                    return Ok(c);
95                }
96                dir = d.parent().map(|p| p.to_path_buf());
97            }
98        }
99    }
100    if let Some(path) = std::env::var_os("PATH") {
101        for d in std::env::split_paths(&path) {
102            let c = d.join(&file);
103            if c.is_file() {
104                return Ok(c);
105            }
106        }
107    }
108    Err(Error::new(
109        ErrorKind::Plugin,
110        "plugin_host_missing",
111        format!("the {} plugin host '{}' was not found next to ferraus or on PATH", format, format.host_program()),
112    )
113    .with_hint(format!("reinstall ferraus, or set {var}=/path/to/{}", format.host_program())))
114}
115
116static SHM_COUNTER: AtomicU64 = AtomicU64::new(0);
117
118struct ShmFile {
119    path: PathBuf,
120    map: memmap2::MmapMut,
121    layout: ShmLayout,
122}
123
124impl ShmFile {
125    fn create(layout: ShmLayout) -> Result<ShmFile> {
126        let nanos =
127            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0);
128        let path = std::env::temp_dir().join(format!(
129            "ferraus-shm-{}-{}-{nanos}.bin",
130            std::process::id(),
131            SHM_COUNTER.fetch_add(1, Ordering::SeqCst)
132        ));
133        let file = std::fs::OpenOptions::new()
134            .read(true)
135            .write(true)
136            .create_new(true)
137            .open(&path)
138            .map_err(|e| Error::processing(format!("cannot create {}: {e}", path.display())))?;
139        file.set_len(layout.total_bytes() as u64)?;
140        // SAFETY: the file was just created by us and is only shared with our own host.
141        let mut map = unsafe { memmap2::MmapMut::map_mut(&file) }
142            .map_err(|e| Error::processing(format!("cannot map shared memory: {e}")))?;
143        layout.write_header(&mut map);
144        map.flush()?;
145        Ok(ShmFile { path, map, layout })
146    }
147}
148
149impl Drop for ShmFile {
150    fn drop(&mut self) {
151        let _ = std::fs::remove_file(&self.path);
152    }
153}
154
155/// What activation reported.
156#[derive(Clone, Copy, Debug)]
157pub struct Activated {
158    pub latency: u32,
159    pub tail: Option<u32>,
160    pub channels_in: u32,
161    pub channels_out: u32,
162}
163
164/// One running host process with one plugin.
165pub struct PluginHost {
166    child: Child,
167    stdin: Option<ChildStdin>,
168    rx: Receiver<std::io::Result<Vec<u8>>>,
169    stderr_tail: Arc<Mutex<String>>,
170    label: String,
171    path: PathBuf,
172    watchdog: Duration,
173    shm: Option<ShmFile>,
174    failed: Option<Error>,
175}
176
177impl PluginHost {
178    /// Start a host for `format`; `label` and `path` name the plugin in errors.
179    pub fn spawn(format: PluginFormat, label: &str, path: &Path, watchdog: Duration) -> Result<PluginHost> {
180        let bin = host_binary(format)?;
181        let mut child =
182            Command::new(&bin).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped()).spawn().map_err(
183                |e| Error::new(ErrorKind::Plugin, "plugin_host_failed", format!("cannot start {}: {e}", bin.display())),
184            )?;
185        let stdout = child.stdout.take().expect("piped stdout");
186        let stderr = child.stderr.take().expect("piped stderr");
187        let (tx, rx) = mpsc::channel();
188        std::thread::spawn(move || {
189            let mut r = BufReader::new(stdout);
190            loop {
191                match read_frame(&mut r) {
192                    Ok(Some(f)) => {
193                        if tx.send(Ok(f)).is_err() {
194                            break;
195                        }
196                    }
197                    Ok(None) => break,
198                    Err(e) => {
199                        let _ = tx.send(Err(e));
200                        break;
201                    }
202                }
203            }
204        });
205        let tail = Arc::new(Mutex::new(String::new()));
206        let tail2 = tail.clone();
207        let echo = std::env::var_os("FERRAUS_PLUGIN_LOG").is_some();
208        std::thread::spawn(move || {
209            let r = BufReader::new(stderr);
210            for line in r.lines() {
211                let Ok(line) = line else { break };
212                if echo {
213                    eprintln!("[plughost] {line}");
214                }
215                if let Ok(mut t) = tail2.lock() {
216                    t.push_str(&line);
217                    t.push('\n');
218                    if t.len() > 4000 {
219                        let cut = t.len() - 3000;
220                        let cut = (cut..t.len()).find(|i| t.is_char_boundary(*i)).unwrap_or(0);
221                        t.drain(..cut);
222                    }
223                }
224            }
225        });
226        let stdin = child.stdin.take();
227        let mut h = PluginHost {
228            child,
229            stdin,
230            rx,
231            stderr_tail: tail,
232            label: label.to_string(),
233            path: path.to_path_buf(),
234            watchdog,
235            shm: None,
236            failed: None,
237        };
238        match h.request(&Request::Hello)? {
239            Reply::Hello { protocol, .. } if protocol == PROTOCOL_VERSION => Ok(h),
240            Reply::Hello { protocol, host, .. } => Err(Error::new(
241                ErrorKind::Plugin,
242                "plugin_host_mismatch",
243                format!("{host} speaks protocol {protocol}, ferraus expects {PROTOCOL_VERSION}"),
244            )
245            .with_hint("reinstall ferraus so the plugin host matches")),
246            other => Err(h.unexpected(other)),
247        }
248    }
249
250    pub fn label(&self) -> &str {
251        &self.label
252    }
253
254    pub fn set_label(&mut self, label: &str) {
255        self.label = label.to_string();
256    }
257
258    fn who(&self) -> String {
259        format!("plugin '{}' ({})", self.label, self.path.display())
260    }
261
262    fn stderr_excerpt(&self) -> String {
263        let t = self.stderr_tail.lock().map(|t| t.clone()).unwrap_or_default();
264        let lines: Vec<&str> = t.lines().filter(|l| !l.trim().is_empty()).collect();
265        if lines.is_empty() {
266            String::new()
267        } else {
268            let start = lines.len().saturating_sub(5);
269            format!("; host said: {}", lines[start..].join(" | "))
270        }
271    }
272
273    fn unexpected(&self, r: Reply) -> Error {
274        Error::new(
275            ErrorKind::Plugin,
276            "plugin_protocol_error",
277            format!("{}: unexpected reply from the host: {r:?}", self.who()),
278        )
279    }
280
281    /// The host died: describe how.
282    fn crashed(&mut self) -> Error {
283        let deadline = Instant::now() + Duration::from_secs(3);
284        let status = loop {
285            match self.child.try_wait() {
286                Ok(Some(s)) => break Some(s),
287                Ok(None) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(10)),
288                _ => {
289                    let _ = self.child.kill();
290                    break self.child.wait().ok();
291                }
292            }
293        };
294        let how = match status {
295            Some(s) => describe_exit(s),
296            None => "the host process vanished".into(),
297        };
298        let e = Error::new(
299            ErrorKind::Plugin,
300            "plugin_crashed",
301            format!("{} crashed: {how}{}", self.who(), self.stderr_excerpt()),
302        )
303        .with_hint("the plugin failed inside its own process; ferraus is unaffected — try other parameters or report it to the plugin vendor");
304        self.failed = Some(e.clone());
305        e
306    }
307
308    fn timed_out(&mut self) -> Error {
309        let _ = self.child.kill();
310        let _ = self.child.wait();
311        let e = Error::new(
312            ErrorKind::Plugin,
313            "plugin_timeout",
314            format!(
315                "{} made no progress for {:.0} s and was stopped by the watchdog{}",
316                self.who(),
317                self.watchdog.as_secs_f64(),
318                self.stderr_excerpt()
319            ),
320        )
321        .with_hint("raise the limit with --watchdog 300s or FERRAUS_PLUGIN_WATCHDOG_S=300");
322        self.failed = Some(e.clone());
323        e
324    }
325
326    fn send(&mut self, req: &Request) -> Result<()> {
327        if let Some(e) = &self.failed {
328            return Err(e.clone());
329        }
330        let Some(stdin) = self.stdin.as_mut() else {
331            return Err(self.crashed());
332        };
333        if write_frame(stdin, req).is_err() {
334            return Err(self.crashed());
335        }
336        Ok(())
337    }
338
339    fn recv(&mut self) -> Result<Reply> {
340        if let Some(e) = &self.failed {
341            return Err(e.clone());
342        }
343        match self.rx.recv_timeout(self.watchdog) {
344            Ok(Ok(bytes)) => {
345                let r: Reply = serde_json::from_slice(&bytes).map_err(|e| {
346                    Error::new(
347                        ErrorKind::Plugin,
348                        "plugin_protocol_error",
349                        format!("{}: malformed reply: {e}", self.who()),
350                    )
351                })?;
352                if let Reply::Error { code, message } = r {
353                    return Err(self.reply_error(&code, &message));
354                }
355                Ok(r)
356            }
357            Ok(Err(_)) | Err(RecvTimeoutError::Disconnected) => Err(self.crashed()),
358            Err(RecvTimeoutError::Timeout) => Err(self.timed_out()),
359        }
360    }
361
362    fn reply_error(&self, code: &str, message: &str) -> Error {
363        let (kind, hint) = match code {
364            "unknown_param" | "bad_value" => {
365                (ErrorKind::Usage, Some(format!("ferraus plugin info \"{}\"", self.label)))
366            }
367            "unknown_preset" => (ErrorKind::Usage, Some(format!("ferraus plugin presets list \"{}\"", self.label))),
368            "bad_state" | "bad_request" => (ErrorKind::Usage, None),
369            _ => (ErrorKind::Plugin, None),
370        };
371        let e = Error::new(kind, code, format!("{}: {message}", self.who()));
372        match hint {
373            Some(h) => e.with_hint(h),
374            None => e,
375        }
376    }
377
378    /// Send one request and wait (under the watchdog) for its reply.
379    pub fn request(&mut self, req: &Request) -> Result<Reply> {
380        self.send(req)?;
381        self.recv()
382    }
383
384    pub fn probe(&mut self, path: &Path) -> Result<Vec<Descriptor>> {
385        match self.request(&Request::Probe { path: path.to_string_lossy().into_owned() })? {
386            Reply::Probe { plugins } => Ok(plugins),
387            other => Err(self.unexpected(other)),
388        }
389    }
390
391    pub fn load(&mut self, plugin_id: Option<&str>) -> Result<PluginInfo> {
392        match self.request(&Request::Load {
393            path: self.path.to_string_lossy().into_owned(),
394            plugin_id: plugin_id.map(|s| s.to_string()),
395        })? {
396            Reply::Loaded { info } => Ok(info),
397            other => Err(self.unexpected(other)),
398        }
399    }
400
401    fn params_reply(&mut self, req: Request) -> Result<Vec<ParamValue>> {
402        match self.request(&req)? {
403            Reply::Params { params } => Ok(params),
404            other => Err(self.unexpected(other)),
405        }
406    }
407
408    pub fn set_params(&mut self, params: &[(String, String)]) -> Result<Vec<ParamValue>> {
409        if params.is_empty() {
410            return Ok(Vec::new());
411        }
412        let params = params.iter().map(|(n, v)| ParamAssign { name: n.clone(), value: v.clone() }).collect();
413        self.params_reply(Request::SetParams { params })
414    }
415
416    pub fn get_params(&mut self) -> Result<Vec<ParamValue>> {
417        self.params_reply(Request::GetParams)
418    }
419
420    pub fn load_preset(&mut self, preset: &str) -> Result<Vec<ParamValue>> {
421        self.params_reply(Request::LoadPreset { preset: preset.to_string() })
422    }
423
424    pub fn list_presets(&mut self) -> Result<Vec<PresetInfo>> {
425        match self.request(&Request::ListPresets)? {
426            Reply::Presets { presets } => Ok(presets),
427            other => Err(self.unexpected(other)),
428        }
429    }
430
431    pub fn get_state(&mut self) -> Result<Vec<u8>> {
432        match self.request(&Request::GetState)? {
433            Reply::State { state_b64 } => base64::engine::general_purpose::STANDARD
434                .decode(state_b64)
435                .map_err(|e| Error::plugin(format!("{}: state is not base64: {e}", self.who()))),
436            other => Err(self.unexpected(other)),
437        }
438    }
439
440    pub fn set_state(&mut self, state: &[u8]) -> Result<()> {
441        let state_b64 = base64::engine::general_purpose::STANDARD.encode(state);
442        match self.request(&Request::SetState { state_b64 })? {
443            Reply::Ok => Ok(()),
444            other => Err(self.unexpected(other)),
445        }
446    }
447
448    /// Create the shared-memory ring and activate the plugin.
449    pub fn activate(
450        &mut self,
451        sample_rate: u32,
452        block_size: u32,
453        channels_in: u32,
454        channels_out: u32,
455        slots: u32,
456    ) -> Result<Activated> {
457        let layout = ShmLayout { channels_in, channels_out, block_size, slots: slots.max(1) };
458        let shm = ShmFile::create(layout)?;
459        let req = Request::Activate {
460            sample_rate: sample_rate as f64,
461            block_size,
462            shm_path: shm.path.to_string_lossy().into_owned(),
463        };
464        self.shm = Some(shm);
465        match self.request(&req)? {
466            Reply::Activated { latency, tail, channels_in, channels_out } => {
467                Ok(Activated { latency, tail, channels_in, channels_out })
468            }
469            other => Err(self.unexpected(other)),
470        }
471    }
472
473    /// Stream `total` frames through the plugin. `input` (planar, may be shorter than
474    /// `total` — the rest is silence) must match the shared-memory input channel count.
475    /// `events` are `(absolute frame, event)` pairs; their `time` fields are rewritten.
476    /// `progress` is called after each block with the fraction done and may abort.
477    pub fn run(
478        &mut self,
479        input: &[Vec<f32>],
480        total: usize,
481        events: &[(u64, Event)],
482        progress: &mut dyn FnMut(f64) -> Result<()>,
483    ) -> Result<Vec<Vec<f32>>> {
484        let layout = self.shm.as_ref().map(|s| s.layout).ok_or_else(|| Error::internal("run before activate"))?;
485        let bs = layout.block_size as usize;
486        let blocks = total.div_ceil(bs);
487        let mut out = vec![vec![0.0f32; total]; layout.channels_out as usize];
488        let mut events: Vec<(u64, Event)> = events.to_vec();
489        events.sort_by_key(|e| e.0);
490        let mut ev_i = 0usize;
491        let (mut sent, mut done) = (0usize, 0usize);
492        while done < blocks {
493            while sent < blocks && sent - done < layout.slots as usize {
494                let start = sent * bs;
495                let n = bs.min(total - start);
496                let slot = (sent % layout.slots as usize) as u32;
497                {
498                    let shm = self.shm.as_mut().unwrap();
499                    let base = shm.map.as_mut_ptr();
500                    for c in 0..layout.channels_in {
501                        // SAFETY: offsets from our own layout; the host is not using this
502                        // slot (at most `slots` blocks are in flight).
503                        let dst = unsafe { f32_slice_mut(base, layout.input_offset(slot, c), n) };
504                        dst.fill(0.0);
505                        if let Some(src) = input.get(c as usize) {
506                            if start < src.len() {
507                                let m = n.min(src.len() - start);
508                                dst[..m].copy_from_slice(&src[start..start + m]);
509                            }
510                        }
511                    }
512                }
513                let mut evs = Vec::new();
514                while ev_i < events.len() && (events[ev_i].0 as usize) < start + n {
515                    let (frame, e) = events[ev_i];
516                    let t = (frame as usize).saturating_sub(start) as u32;
517                    evs.push(match e {
518                        Event::NoteOn { channel, key, velocity, .. } => {
519                            Event::NoteOn { time: t, channel, key, velocity }
520                        }
521                        Event::NoteOff { channel, key, velocity, .. } => {
522                            Event::NoteOff { time: t, channel, key, velocity }
523                        }
524                        Event::Midi { data, .. } => Event::Midi { time: t, data },
525                    });
526                    ev_i += 1;
527                }
528                std::sync::atomic::fence(Ordering::SeqCst);
529                self.send(&Request::Process { slot, frames: n as u32, events: evs })?;
530                sent += 1;
531            }
532            match self.recv()? {
533                Reply::Processed { slot } => {
534                    let expect = (done % layout.slots as usize) as u32;
535                    if slot != expect {
536                        return Err(Error::plugin(format!(
537                            "{}: host answered slot {slot}, expected {expect}",
538                            self.who()
539                        )));
540                    }
541                    std::sync::atomic::fence(Ordering::SeqCst);
542                    let start = done * bs;
543                    let n = bs.min(total - start);
544                    let shm = self.shm.as_mut().unwrap();
545                    let base = shm.map.as_mut_ptr();
546                    for (c, ch) in out.iter_mut().enumerate() {
547                        let src = unsafe { f32_slice_mut(base, layout.output_offset(slot, c as u32), n) };
548                        ch[start..start + n].copy_from_slice(src);
549                    }
550                    done += 1;
551                    progress(done as f64 / blocks as f64)?;
552                }
553                other => return Err(self.unexpected(other)),
554            }
555        }
556        Ok(out)
557    }
558
559    /// Ask the host to exit and reap it.
560    pub fn shutdown(mut self) {
561        self.stop();
562    }
563
564    fn stop(&mut self) {
565        if self.failed.is_none() {
566            if let Some(mut s) = self.stdin.take() {
567                let _ = write_frame(&mut s, &Request::Quit);
568                let _ = s.flush();
569            }
570        }
571        self.stdin = None;
572        let deadline = Instant::now() + Duration::from_secs(2);
573        loop {
574            match self.child.try_wait() {
575                Ok(Some(_)) => break,
576                Ok(None) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(5)),
577                _ => {
578                    let _ = self.child.kill();
579                    let _ = self.child.wait();
580                    break;
581                }
582            }
583        }
584        self.shm = None;
585    }
586}
587
588impl Drop for PluginHost {
589    fn drop(&mut self) {
590        self.stop();
591    }
592}
593
594fn describe_exit(s: std::process::ExitStatus) -> String {
595    #[cfg(unix)]
596    {
597        use std::os::unix::process::ExitStatusExt;
598        if let Some(sig) = s.signal() {
599            let name = match sig {
600                4 => " (SIGILL)",
601                6 => " (SIGABRT)",
602                7 | 10 => " (SIGBUS)",
603                8 => " (SIGFPE)",
604                9 => " (SIGKILL)",
605                11 => " (SIGSEGV)",
606                _ => "",
607            };
608            return format!("the host process was killed by signal {sig}{name}");
609        }
610    }
611    match s.code() {
612        Some(c) => format!("the host process exited with code {c}"),
613        None => "the host process ended abnormally".into(),
614    }
615}
ferraus_plugins/client.rs

ferraus_plugins/
client.rs
