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

ferraus_plugins/
client.rs
