Files
effectcraft/crates/helper-host/src/client.rs
T
611a282bea Plug-ins: external effect helpers (safe, out-of-process native effects) + plug-in API 2 (#889)
* Plug-ins: API 2 (PluginHost, padding, layer_span)

Split out of the OpenFX work so it can be reviewed on its own. Additive; the WebAssembly ABI
stays at 1.

- point3 / layer / text / hidden parameters, integer sliders, parameter groups.
- EffectPlugin::padding (transparent pixels around the buffer, at most 4096 per side).
- EffectPlugin::render_v2 with a PluginHost: the input at other times, other layers, parameters
  at other times, frame rate, per-instance key, the layer's source range.
- EffectEnv::layer_span from the renderer; MAX_PLUGINS 1024.
- plugin_error: a plug-in effect's last render error, for the UI.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KxacwTpUwqyfpcLgKW8yjS

* Plug-ins: external effect helpers (crates/helper-host)

Native plug-in formats run in separate helper programs, never in EffectCraft's process
(docs/adr/0001-external-effect-helpers.md). New safe-Rust crate effectcraft-helper-host (L3,
native only, unsafe forbidden): it finds helper programs, starts each as a child process and
registers its effects through plug-in API 2, so the UI and command surface need nothing new.

- Protocol v1 (docs/external-helpers.md): JSON lines over stdin / stdout (helper replies are
  prefixed so plug-in chatter is ignored); pixels go through a frame exchange file (/dev/shm on
  Linux), never through the pipe. Render-time callbacks for the input at other times, other layers
  and parameters at other times, with nested renders through the same helper.
- A helper that exits, hangs past its timeout or breaks the protocol is stopped. The render fails
  with a message, the frame is left as it was, and the helper restarts for the next render. An
  effect that stops its helper 3 times is turned off (helper_blocklist.txt).
- Opt-in: nothing starts helpers unless the user asks (Effect > Scan for Effect Helpers), or opts
  in to starting them at launch with EFFECTCRAFT_HELPERS=1, an environment variable, so writing
  to the settings folder alone can't run code. effect.helpers.scan / load are refused for
  confined agent sessions (confine.rs), like file.executeFile.
- Discovery: helpers/ next to the app (Contents/Helpers on macOS), helpers/ in the settings
  folder, EFFECTCRAFT_HELPER_PATH.
- Commands effect.helpers.list / scan / load; Effect menu entries (translated); the last render
  error of an effect shows as a warning sign after its name in Effect Controls.
- serve(): the helper side for Rust helpers. tests/it/main.rs runs the whole protocol (render,
  padding, callbacks, nesting, threads, crash, restart, block list, hang) against a fake helper,
  with nothing to install.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KxacwTpUwqyfpcLgKW8yjS

* Effect helpers: address review (start failures, frame file mode, nextest listing)

- A helper that can't start (not installed any more, restarting too often, no frame exchange
  file) is reported but no longer charged to the effect being rendered: it is Failure::Refused,
  not Failure::Broken, so it never counts toward turning that effect off. New check in
  tests/it: once the helper is out of restarts, renders of an effect fail without blocking it
  (fails without the fix: the effect was turned off after 3 such renders).
- The frame exchange file is created owner-only (0o600) on Unix: it holds the user's frames and
  /dev/shm is shared. Asserted in the frames unit test.
- tests/it (harness = false) speaks the libtest listing protocol cargo-nextest uses:
  `--list --format terse` prints `helper_protocol: test` (nothing with `--ignored`), and
  `helper_protocol --exact` runs the checks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KxacwTpUwqyfpcLgKW8yjS

* Effect helpers: bound writes to a helper, drain stderr, keep callbacks in step

- Writes to a helper go through a writer thread with the request timeout: a
  helper that stops reading its stdin (hung with a large callback answer or
  render request pending) no longer blocks the render thread forever while
  holding the helper.
- stderr is drained as bytes (lossy, in 64 KiB pieces) to the end: output that
  isn't UTF-8 stopped the reader, closing the pipe under the helper (EPIPE /
  SIGPIPE on its next write); lines were unbounded.
- Callback images that can't be written are answered with `null` instead of
  failing the render while the helper waits (which left the protocol out of
  step); a frame that can't be written before the request is sent is not
  charged to the effect.
- Mid-render calls use the render's own process: no respawn or heartbeat ping
  while the helper is waiting for an answer.
- Waiting for a busy helper is bounded (load timeout) instead of forever;
  status() no longer blocks on a running render.
- Reap the child when start-up fails after spawning; cap the block list read.
- Scripts start helpers only with Allow Scripts to Write Files and Access
  Network on (they run native code), as for File.execute().
- helper-host declares the workspace's no-panic crate lints; the desktop
  main() keeps its doc comment.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Effect helpers: rustfmt the fake provider, build the test session in its initializer (clippy)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Plug-in Library: offer WebAssembly API 1 releases only; a helper already stopped mid-render isn't charged again

- WasmPlugin loads only modules at the WebAssembly ABI version, which stays 1 while
  PLUGIN_API_VERSION is 2: share it as effects::plugin::WASM_API_VERSION and have the
  library's compatibility check use it, so it never installs an API 2 release that
  can't load.
- with_running: a process that is already gone was stopped by what broke it (a nested
  render, charged there), so this render fails without charging its own effect.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* frames: plug-in effects count as reading other frames (plug-in API 2)

With main's narrowed frame invalidation (#2795), an edit inside a precomp under an
effect that reads its layer at other times must reach the layer's whole span.
Plug-in API 2 gives every plug-in effect the input at other times, so any effect
outside ec. counts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: FredApps <153173854+FredApps@users.noreply.github.com>
2026-11-05 11:11:57 +01:00

687 lines
28 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! EffectCraft's side of a helper: the child process, its pipes and frame exchange file, and
//! the [`EffectPlugin`] each remote effect is registered as.
//!
//! Nothing a helper does can take EffectCraft down: a helper that exits, hangs past its
//! timeout or answers nonsense is stopped, the render that hit it fails with a message (shown on
//! the effect, the frame is left as it was), and the next render starts the helper again. An
//! effect that took its helper down [`crate::blocklist::CRASH_LIMIT`] times is blocked.
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdin, Command, Stdio};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError, TryLockError};
use std::thread::ThreadId;
use std::time::{Duration, Instant};
use effectcraft_effects::plugin::{EffectPlugin, PluginFrame, PluginHost, PluginImage, PluginManifest, PluginParams};
use crate::blocklist;
use crate::frames::Exchange;
use crate::protocol::{HELPER_ARG, ImageRef, MAX_LINE, PROTOCOL_VERSION, RemoteEffect, RenderRequest, Reply, Request, parse_reply, request_line};
/// How deep renders through one helper may nest (an effect asking for its input, which another
/// effect of the same helper renders, ...).
const MAX_DEPTH: u32 = 16;
/// Restarts allowed per minute before a helper is left stopped until the next scan.
const MAX_RESTARTS_PER_MINUTE: usize = 5;
/// A helper idle for longer than this is pinged before it is used again.
const HEARTBEAT_AFTER: Duration = Duration::from_secs(30);
const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(10);
/// The longest piece of a helper's stderr logged as one line (longer lines are logged in parts).
const MAX_LOG_LINE: u64 = 64 << 10;
/// [`set_render_timeout`] (milliseconds; 0 = not set).
static RENDER_TIMEOUT_MS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
/// Override the render timeout (tests; `EFFECTCRAFT_HELPER_TIMEOUT` still wins).
pub fn set_render_timeout(t: Duration) {
RENDER_TIMEOUT_MS.store(u64::try_from(t.as_millis()).unwrap_or(u64::MAX), std::sync::atomic::Ordering::Relaxed);
}
/// Timeout for one render step, `EFFECTCRAFT_HELPER_TIMEOUT` seconds (default 120: heavy
/// plug-ins take a while on their first frame).
fn render_timeout() -> Duration {
if let Some(s) = env_secs("EFFECTCRAFT_HELPER_TIMEOUT") {
return Duration::from_secs(s);
}
match RENDER_TIMEOUT_MS.load(std::sync::atomic::Ordering::Relaxed) {
0 => Duration::from_secs(120),
ms => Duration::from_millis(ms),
}
}
/// Timeout for start-up, `describe` and `load`, `EFFECTCRAFT_HELPER_LOAD_TIMEOUT` seconds
/// (default 600: scanning large plug-in suites is slow).
fn load_timeout() -> Duration {
Duration::from_secs(env_secs("EFFECTCRAFT_HELPER_LOAD_TIMEOUT").unwrap_or(600))
}
fn env_secs(k: &str) -> Option<u64> {
std::env::var(k).ok()?.trim().parse().ok().filter(|s| *s > 0)
}
pub(crate) fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
m.lock().unwrap_or_else(PoisonError::into_inner)
}
/// What the reader thread passes on.
enum Line {
Reply(Result<Reply, String>),
Closed,
}
/// A running helper process.
struct Conn {
child: Child,
/// Request lines for the writer thread, which owns the helper's stdin: a helper that stops
/// reading can block that thread, never the render waiting for it ([`Conn::send`]).
input: Sender<Vec<u8>>,
/// The writer thread's result for each line, in order.
written: Receiver<Result<(), String>>,
rx: Receiver<Line>,
exchange: Exchange,
/// The first free byte of the exchange file (images are stacked, see [`Helper::render`]).
top: u64,
last_used: Instant,
generation: u64,
}
impl Conn {
/// Write one request, waiting at most `timeout` for the helper to take it.
fn send(&mut self, r: &Request, timeout: Duration) -> Result<(), Failure> {
let line = request_line(r).map_err(Failure::Broken)?;
let closed = || Failure::Broken("can't write to the helper: its input is closed".into());
self.input.send(line.into_bytes()).map_err(|_| closed())?;
match self.written.recv_timeout(timeout) {
Ok(Ok(())) => Ok(()),
Ok(Err(e)) => Err(Failure::Broken(format!("can't write to the helper: {e}"))),
Err(RecvTimeoutError::Timeout) => {
Err(Failure::Broken(format!("the helper did not answer within {} s (it stopped reading its input)", timeout.as_secs())))
}
Err(RecvTimeoutError::Disconnected) => Err(closed()),
}
}
fn recv(&mut self, timeout: Duration) -> Result<Reply, Failure> {
match self.rx.recv_timeout(timeout) {
Ok(Line::Reply(Ok(r))) => Ok(r),
Ok(Line::Reply(Err(e))) => Err(Failure::Broken(e)),
Ok(Line::Closed) | Err(RecvTimeoutError::Disconnected) => {
// Give the process a moment to finish exiting so its status can be reported.
std::thread::sleep(Duration::from_millis(50));
let status = match self.child.try_wait() {
Ok(Some(s)) => s.to_string(),
_ => "closed its output".into(),
};
Err(Failure::Broken(format!("the helper stopped ({status})")))
}
Err(RecvTimeoutError::Timeout) => Err(Failure::Broken(format!("the helper did not answer within {} s", timeout.as_secs()))),
}
}
fn call(&mut self, r: &Request, timeout: Duration) -> Result<Reply, Failure> {
self.send(r, timeout)?;
self.recv(timeout)
}
/// Reserve room for an image of `width × height` at the top of the exchange file.
fn alloc(&mut self, width: u32, height: u32, origin: [f64; 2], scale: f64) -> Result<ImageRef, String> {
let at = ImageRef { offset: self.top, width, height, origin, scale };
let bytes = at.bytes().ok_or_else(|| format!("image {width}×{height} is too large for a helper"))?;
// Keep images 64-byte aligned for helpers that map the file.
self.top = self.top.checked_add(bytes.div_ceil(64).saturating_mul(64)).ok_or("frame exchange file too large")?;
Ok(at)
}
fn stop(mut self) {
// Not waited for: the process is killed right after, which also ends a blocked write.
if let Ok(line) = request_line(&Request::Shutdown) {
let _ = self.input.send(line.into_bytes());
}
let _ = self.child.kill();
let _ = self.child.wait();
}
}
/// Why a request failed.
enum Failure {
/// The helper answered with an error and is fine.
Refused(String),
/// The helper crashed, hung or broke the protocol: it is stopped.
Broken(String),
}
/// Whose turn it is to talk to the helper: one thread at a time, re-entered by that thread for
/// nested renders.
#[derive(Default)]
struct Turn {
owner: Option<ThreadId>,
depth: u32,
}
struct TurnGuard<'a>(&'a Helper);
impl Drop for TurnGuard<'_> {
fn drop(&mut self) {
let mut t = lock(&self.0.turn);
t.depth = t.depth.saturating_sub(1);
if t.depth == 0 {
t.owner = None;
self.0.turn_cv.notify_all();
}
}
}
/// What [`Helper::status`] reports.
#[derive(Clone, Debug, Default)]
pub struct HelperStatus {
pub program: PathBuf,
pub name: String,
pub version: String,
pub loads: Vec<String>,
pub running: bool,
pub effects: usize,
pub restarts: usize,
pub last_error: Option<String>,
}
/// One helper program and, while it runs, its process.
pub struct Helper {
program: PathBuf,
info: Mutex<HelperStatus>,
conn: Mutex<Option<Conn>>,
turn: Mutex<Turn>,
turn_cv: Condvar,
/// Paths loaded with [`Helper::load`], loaded again after a restart.
loaded: Mutex<Vec<String>>,
restarts: Mutex<Vec<Instant>>,
generation: Mutex<u64>,
}
impl Helper {
/// Start a helper and list its effects.
pub fn start(program: &Path) -> Result<(Arc<Helper>, Vec<RemoteEffect>, Vec<String>), String> {
let h = Arc::new(Helper {
program: program.to_path_buf(),
info: Mutex::new(HelperStatus { program: program.to_path_buf(), ..Default::default() }),
conn: Mutex::new(None),
turn: Mutex::new(Turn::default()),
turn_cv: Condvar::new(),
loaded: Mutex::new(vec![]),
restarts: Mutex::new(vec![]),
generation: Mutex::new(0),
});
let (effects, errors) = {
let _turn = h.take_turn()?;
// A helper that crashes while listing its effects (one bad plug-in) is given a
// second and third chance: helpers remember what crashed them and skip it.
let mut last = String::new();
let mut out = None;
for _ in 0..3 {
match h.with_conn(|c| c.call(&Request::Describe, load_timeout())) {
Ok(Reply::Effects { effects, errors }) => {
out = Some((effects, errors));
break;
}
Ok(other) => last = format!("unexpected reply to describe: {other:?}"),
Err(Failure::Refused(e) | Failure::Broken(e)) => last = e,
}
}
out.ok_or_else(|| format!("{}: {last}", program.display()))?
};
lock(&h.info).effects = effects.len();
Ok((h, effects, errors))
}
pub fn program(&self) -> &Path {
&self.program
}
pub fn status(&self) -> HelperStatus {
let mut s = lock(&self.info).clone();
// A render holds the process for as long as the helper works on it: don't wait for it.
s.running = match self.conn.try_lock() {
Ok(c) => c.is_some(),
Err(TryLockError::Poisoned(c)) => c.into_inner().is_some(),
Err(TryLockError::WouldBlock) => true,
};
s.restarts = lock(&self.restarts).len();
s
}
pub fn loads(&self, ext: &str) -> bool {
lock(&self.info).loads.iter().any(|e| e.eq_ignore_ascii_case(ext))
}
pub(crate) fn add_effects(&self, n: usize) {
let mut i = lock(&self.info);
i.effects = i.effects.saturating_add(n);
}
/// Load effects from a path through the helper ([`Request::Load`]).
pub fn load(&self, path: &str) -> Result<(Vec<RemoteEffect>, Vec<String>), String> {
let _turn = self.take_turn()?;
let r = self.with_conn(|c| c.call(&Request::Load { path: path.to_string() }, load_timeout()));
match r {
Ok(Reply::Effects { effects, errors }) => {
let mut l = lock(&self.loaded);
if !l.iter().any(|p| p == path) {
l.push(path.to_string());
}
Ok((effects, errors))
}
Ok(Reply::Error { message }) => Err(message),
Ok(other) => Err(format!("unexpected reply to load: {other:?}")),
Err(Failure::Refused(e) | Failure::Broken(e)) => Err(e),
}
}
/// Stop the process (it starts again when next needed).
pub fn stop(&self) {
if let Some(c) = lock(&self.conn).take() {
c.stop();
}
}
fn take_turn(&self) -> Result<TurnGuard<'_>, String> {
let me = std::thread::current().id();
// Bounded, so a thread that can't get the helper (its holder waiting for work that a
// blocked thread pool would have to do) fails instead of hanging forever.
let limit = load_timeout();
let deadline = Instant::now().checked_add(limit);
let mut t = lock(&self.turn);
loop {
match t.owner {
None => {
t.owner = Some(me);
t.depth = 1;
break;
}
Some(o) if o == me => {
if t.depth >= MAX_DEPTH {
return Err("effects through this helper are nested too deeply".into());
}
t.depth += 1;
break;
}
Some(_) => {
let left = deadline.map_or(limit, |d| d.saturating_duration_since(Instant::now()));
if left.is_zero() {
return Err(format!("the effect helper was busy for more than {} s", limit.as_secs()));
}
t = self.turn_cv.wait_timeout(t, left).unwrap_or_else(PoisonError::into_inner).0;
}
}
}
Ok(TurnGuard(self))
}
/// Run `f` on the running process, starting it first if needed. A [`Failure::Broken`]
/// stops the process.
fn with_conn<T>(&self, f: impl FnOnce(&mut Conn) -> Result<T, Failure>) -> Result<T, Failure> {
let mut g = lock(&self.conn);
if g.is_none() {
*g = Some(self.spawn().map_err(|e| self.not_started(e))?);
}
let Some(c) = g.as_mut() else { return Err(Failure::Broken("the helper is not running".into())) };
if c.last_used.elapsed() > HEARTBEAT_AFTER && !matches!(c.call(&Request::Ping, HEARTBEAT_TIMEOUT), Ok(Reply::Pong)) {
if let Some(dead) = g.take() {
dead.stop();
}
self.note_error("the helper stopped answering; restarting it");
*g = Some(self.spawn().map_err(|e| self.not_started(e))?);
}
let Some(c) = g.as_mut() else { return Err(Failure::Broken("the helper is not running".into())) };
let r = f(c);
c.last_used = Instant::now();
if let Err(Failure::Broken(e)) = &r {
self.note_error(e);
if let Some(dead) = g.take() {
dead.stop();
}
}
r
}
/// Run `f` on the process a render started on (generation `started`), without starting or
/// pinging one: mid-render the helper is waiting for the answer to its question. A
/// [`Failure::Broken`] stops the process. A process already gone was stopped by what broke it
/// (a nested render, charged there): not charged again to this render's effect.
fn with_running<T>(&self, started: u64, f: impl FnOnce(&mut Conn) -> Result<T, Failure>) -> Result<T, Failure> {
let mut g = lock(&self.conn);
let Some(c) = g.as_mut().filter(|c| c.generation == started) else {
return Err(Failure::Refused("the helper stopped during the render".into()));
};
let r = f(c);
c.last_used = Instant::now();
if let Err(Failure::Broken(e)) = &r {
self.note_error(e);
if let Some(dead) = g.take() {
dead.stop();
}
}
r
}
/// A helper that could not be started (not installed any more, restarting too often, no
/// frame exchange file): reported, but not charged to the effect that asked for it, since
/// nothing it did stopped the helper ([`Failure::Refused`], not [`Failure::Broken`]).
fn not_started(&self, e: String) -> Failure {
self.note_error(&e);
Failure::Refused(e)
}
fn note_error(&self, e: &str) {
log::warn!("helper {}: {e}", self.program.display());
lock(&self.info).last_error = Some(e.to_string());
}
fn spawn(&self) -> Result<Conn, String> {
{
let mut r = lock(&self.restarts);
r.retain(|t| t.elapsed() < Duration::from_secs(60));
if r.len() >= MAX_RESTARTS_PER_MINUTE {
return Err("the helper keeps stopping; it stays off until the next scan (Effect ▸ Scan for Effect Helpers)".into());
}
r.push(Instant::now());
}
let exchange = Exchange::create().map_err(|e| format!("can't create the frame exchange file: {e}"))?;
let mut cmd = Command::new(&self.program);
cmd.arg(HELPER_ARG).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped());
if let Some(dir) = blocklist::state_dir() {
cmd.env("EFFECTCRAFT_HELPER_STATE_DIR", dir);
}
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
// CREATE_NO_WINDOW: no console window per helper.
cmd.creation_flags(0x0800_0000);
}
let mut child = cmd.spawn().map_err(|e| format!("can't start {}: {e}", self.program.display()))?;
let (Some(stdin), Some(stdout), stderr) = (child.stdin.take(), child.stdout.take(), child.stderr.take()) else {
let _ = child.kill();
let _ = child.wait();
return Err("the helper's pipes are unavailable".into());
};
let (tx, rx) = channel();
let label = self.program.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
let l2 = label.clone();
let spawned = std::thread::Builder::new().name(format!("helper-out {label}")).spawn(move || {
let mut r = BufReader::new(stdout);
loop {
let mut buf = vec![];
match (&mut r).take(MAX_LINE).read_until(b'\n', &mut buf) {
Ok(0) | Err(_) => {
let _ = tx.send(Line::Closed);
return;
}
Ok(_) => {
let line = String::from_utf8_lossy(&buf);
match parse_reply(&line) {
Some(reply) => {
if tx.send(Line::Reply(reply)).is_err() {
return;
}
}
None => log::info!("helper {l2}: {}", line.trim_end()),
}
}
}
}
});
if let Err(e) = spawned {
let _ = child.kill();
let _ = child.wait();
return Err(format!("can't start a reader thread: {e}"));
}
let (input, lines) = channel::<Vec<u8>>();
let (done, written) = channel();
let writer = std::thread::Builder::new().name(format!("helper-in {label}")).spawn(move || {
let mut stdin: ChildStdin = stdin;
for line in lines {
let r = stdin.write_all(&line).and_then(|_| stdin.flush()).map_err(|e| e.to_string());
let failed = r.is_err();
if done.send(r).is_err() || failed {
return;
}
}
});
if let Err(e) = writer {
let _ = child.kill();
let _ = child.wait();
return Err(format!("can't start a writer thread: {e}"));
}
if let Some(err) = stderr {
// Drained to the end whatever it holds (bytes that aren't UTF-8, endless lines): a
// helper blocked on a full stderr pipe would hang its render.
let _ = std::thread::Builder::new().name(format!("helper-err {label}")).spawn(move || {
let mut r = BufReader::new(err);
let mut buf = vec![];
loop {
buf.clear();
match (&mut r).take(MAX_LOG_LINE).read_until(b'\n', &mut buf) {
Ok(0) | Err(_) => return,
Ok(_) => log::info!("helper {label}: {}", String::from_utf8_lossy(&buf).trim_end()),
}
}
});
}
let generation = {
let mut g = lock(&self.generation);
*g = g.saturating_add(1);
*g
};
let mut c = Conn { child, input, written, rx, exchange, top: 0, last_used: Instant::now(), generation };
let frames = c.exchange.path().to_string_lossy().into_owned();
match c.call(&Request::Hello { version: PROTOCOL_VERSION, frames }, load_timeout()) {
Ok(Reply::Hello { version, name, helper_version, loads }) if version == PROTOCOL_VERSION => {
let mut i = lock(&self.info);
(i.name, i.version, i.loads) = (name, helper_version, loads);
}
Ok(Reply::Hello { version, .. }) => {
c.stop();
return Err(format!("the helper speaks protocol {version}; this EffectCraft speaks {PROTOCOL_VERSION}"));
}
Ok(Reply::Error { message }) => {
c.stop();
return Err(message);
}
Ok(other) => {
c.stop();
return Err(format!("unexpected reply to hello: {other:?}"));
}
Err(Failure::Refused(e) | Failure::Broken(e)) => {
c.stop();
return Err(e);
}
}
c.exchange.unlink();
// After a restart the new process knows no effects yet: list them again (the first
// start lists them in `Helper::start`), then load what the user loaded before.
if generation > 1 && !matches!(c.call(&Request::Describe, load_timeout()), Ok(Reply::Effects { .. })) {
c.stop();
return Err("the restarted helper could not list its effects".into());
}
for path in lock(&self.loaded).clone() {
if !matches!(c.call(&Request::Load { path: path.clone() }, load_timeout()), Ok(Reply::Effects { .. })) {
log::warn!("helper {}: reloading {path} failed", self.program.display());
}
}
Ok(c)
}
fn padding(&self, effect: &str, params: &PluginParams, time: f64, scale: f64, layer_size: [f64; 2], instance: u64) -> Result<u32, String> {
let _turn = self.take_turn()?;
let req = Request::Padding { effect: effect.to_string(), instance, time, scale, layer_size, flat: params.flat.to_vec(), texts: params.texts.to_vec() };
match self.with_conn(|c| c.call(&req, render_timeout())) {
Ok(Reply::Padding { pixels }) => Ok(pixels),
Ok(Reply::Error { message }) => Err(message),
Ok(other) => Err(format!("unexpected reply to padding: {other:?}")),
Err(Failure::Refused(e) | Failure::Broken(e)) => Err(e),
}
}
/// Render one frame through the helper. Images live on a stack in the exchange file: the
/// frame first, then whatever the helper asks for, freed again when the render ends (nested
/// renders stack above).
fn render(&self, effect: &str, frame: &mut PluginFrame, params: &PluginParams, time: f64, host: &dyn PluginHost) -> Result<(), Failure> {
let _turn = self.take_turn().map_err(Failure::Refused)?;
// The generation of the process this render runs on: taken inside `with_conn`, which
// may start the process (after a crash or `stop`), so it is never that of a dead one.
let (at, base, started) = self.with_conn(|c| {
let base = c.top;
let at = c.alloc(frame.width, frame.height, frame.origin, frame.scale).map_err(Failure::Refused)?;
// Nothing is sent yet, so the helper is fine (a full disk, say): not its fault.
if let Err(e) = c.exchange.write(&at, frame.pixels) {
c.top = base;
return Err(Failure::Refused(e));
}
Ok((at, base, c.generation))
})?;
let req = Request::Render(RenderRequest {
effect: effect.to_string(),
instance: host.instance_key(),
time,
frame: at,
layer_size: frame.layer_size,
flat: params.flat.to_vec(),
texts: params.texts.to_vec(),
frame_rate: host.frame_rate(),
frame_range: host.frame_range(),
});
let result = self.converse(req, frame, &at, host, started);
// Free this render's images (if the same process is still running).
if let Some(c) = lock(&self.conn).as_mut()
&& c.generation == started
{
c.top = base;
}
result
}
fn converse(&self, req: Request, frame: &mut PluginFrame, at: &ImageRef, host: &dyn PluginHost, started: u64) -> Result<(), Failure> {
let mut next = req;
loop {
let reply = self.with_running(started, |c| c.call(&next, render_timeout()))?;
next = match reply {
Reply::Done => return self.with_running(started, |c| c.exchange.read_into(at, frame.pixels).map_err(Failure::Broken)),
Reply::Error { message } => return Err(Failure::Refused(message)),
// Host callbacks run without holding the process, so they may render through
// this helper again (on this thread: the turn is re-entered).
Reply::InputAt { time } => self.send_image(host.input_at(time), started)?,
Reply::LayerInput { param, time } => self.send_image(host.layer_input(&param, time), started)?,
Reply::ParamsAt { time } => Request::Params { values: host.params_at(time) },
other => return Err(Failure::Broken(format!("unexpected reply during a render: {other:?}"))),
};
}
}
/// Put a callback's image in the exchange file and make the answer. The helper is waiting
/// for one, so an image that can't be passed on is answered with none (failing here would
/// leave it waiting, out of step with the next request).
fn send_image(&self, img: Option<PluginImage>, started: u64) -> Result<Request, Failure> {
let Some(img) = img.filter(|i| u64::from(i.width).saturating_mul(u64::from(i.height)) == i.pixels.len() as u64) else {
return Ok(Request::Image { image: None });
};
self.with_running(started, |c| {
let base = c.top;
let written = c.alloc(img.width, img.height, img.origin, img.scale).and_then(|at| c.exchange.write(&at, &img.pixels).map(|()| at));
Ok(match written {
Ok(at) => Request::Image { image: Some(at) },
Err(e) => {
log::warn!("helper {}: {e}", self.program.display());
c.top = base;
Request::Image { image: None }
}
})
})
}
}
impl Drop for Helper {
fn drop(&mut self) {
self.stop();
}
}
/// A helper's effect, registered like any plug-in effect.
pub struct HelperEffect {
helper: Arc<Helper>,
manifest: PluginManifest,
padding: bool,
}
impl HelperEffect {
pub fn new(helper: Arc<Helper>, remote: RemoteEffect) -> HelperEffect {
HelperEffect { helper, manifest: remote.manifest, padding: remote.padding }
}
}
/// A host for [`EffectPlugin::render`] without services.
struct NoHost;
impl PluginHost for NoHost {
fn input_at(&self, _: f64) -> Option<PluginImage> {
None
}
fn layer_input(&self, _: &str, _: f64) -> Option<PluginImage> {
None
}
fn params_at(&self, _: f64) -> Option<(Vec<f64>, Vec<String>)> {
None
}
fn frame_rate(&self) -> f64 {
30.0
}
fn time(&self) -> f64 {
0.0
}
}
impl EffectPlugin for HelperEffect {
fn manifest(&self) -> &PluginManifest {
&self.manifest
}
fn render(&self, frame: &mut PluginFrame, params: &PluginParams, time: f64) -> Result<(), String> {
self.render_v2(frame, params, time, &NoHost)
}
fn source(&self) -> String {
format!("helper: {}", self.helper.program.display())
}
fn padding(&self, params: &PluginParams, time: f64, scale: f64, layer_size: [f64; 2], instance_key: u64) -> u32 {
if !self.padding || blocklist::is_blocked(&self.manifest.id) {
return 0;
}
match self.helper.padding(&self.manifest.id, params, time, scale, layer_size, instance_key) {
Ok(p) => p,
Err(e) => {
log::warn!("helper effect `{}`: padding: {e}", self.manifest.id);
0
}
}
}
fn render_v2(&self, frame: &mut PluginFrame, params: &PluginParams, time: f64, host: &dyn PluginHost) -> Result<(), String> {
let id = &self.manifest.id;
if blocklist::is_blocked(id) {
return Err(format!(
"{} is turned off: it stopped its helper {} times (Effect ▸ Scan for Effect Helpers with retry turns it back on)",
self.manifest.name,
blocklist::CRASH_LIMIT
));
}
match self.helper.render(id, frame, params, time, host) {
Ok(()) => Ok(()),
Err(Failure::Refused(e)) => Err(e),
Err(Failure::Broken(e)) => {
blocklist::note_crash(id);
Err(format!("{}: the effect helper failed: {e}", self.manifest.name))
}
}
}
}