Mike-Solar
c9ba8695fc
- the oak-editor release build lacked the `windows_subsystem = "windows"` crate attribute, so launching the installed app opened a console window next to the editor (debug builds keep the console so `cargo run` shows logs); - oak-worker, the OFX host and the ffmpeg/ffprobe helpers are console-subsystem binaries spawned by the GUI app, so each spawn now carries CREATE_NO_WINDOW through the shared oak_core::miscutils::hide_console_window helper (a no-op off Windows).
832 lines
25 KiB
Rust
832 lines
25 KiB
Rust
// Oak Video Editor - Non-Linear Video Editor
|
|
// Copyright (C) 2026 Oak Team
|
|
//
|
|
// This program is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU General Public License
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
//! The single OpenFX host client (M3, design §3.2).
|
|
//!
|
|
//! One `oak-worker --ofx-host` child process hosts every OFX plugin
|
|
//! instance; the render eval thread submits [`JobSpec::Plugin`] jobs to it
|
|
//! over the NDJSON control plane and the shm frame-slot data plane:
|
|
//!
|
|
//! - Input textures are read back to CPU and written into the host's
|
|
//! **input pool** (the explicit CPU boundary of the plugin path);
|
|
//! the host renders and publishes its output frame into the **output
|
|
//! pool**, which this client copies out and re-uploads into the
|
|
//! evaluator's texture value.
|
|
//! - `plugin_progress` lines from the host are forwarded to the
|
|
//! process-wide progress callback (`procpool::set_plugin_progress_cb`);
|
|
//! `plugin_cancel` is sent on [`OfxHost::cancel`].
|
|
//! - A dying host (EOF on stdout) fails the in-flight submit, which
|
|
//! respawns the host and **re-posts the same job**; after
|
|
//! [`OfxHostConfig::max_failures`] consecutive crashes the client stays
|
|
//! permanently dead and the evaluator falls back to a purple frame.
|
|
//!
|
|
//! The client is single-slot by design (design §3.2): submissions are
|
|
//! synchronous, one job in flight, because the evaluator itself is
|
|
//! synchronous. The process-wide client is installed by the render
|
|
//! manager when the thread pipeline is selected ([`install_client`]);
|
|
//! with no client installed the evaluator keeps using the in-process
|
|
//! oakplugin executor.
|
|
|
|
use std::collections::HashMap;
|
|
use std::io::{BufRead, BufReader, Write};
|
|
use std::path::PathBuf;
|
|
use std::process::{Child, ChildStdin, Command, Stdio};
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::{Arc, Condvar, Mutex, MutexGuard, OnceLock};
|
|
use std::thread::JoinHandle;
|
|
|
|
use oak_core::frame::VideoParamsPod;
|
|
use oak_core::texture::{Frame, Texture};
|
|
use oak_core::PixelFormat;
|
|
|
|
use crate::error::{Error, Result};
|
|
use crate::eval::JobSpec;
|
|
use crate::ipc::{
|
|
plugin_cancel_json, write_message, HandshakeMsg, OfxInputRef, OfxJobMsg, OfxResultMsg,
|
|
PluginProgressMsg, WireEffectParam, WireNodeValue, TYPE_ERROR, TYPE_OFX_RESULT,
|
|
TYPE_PLUGIN_PROGRESS, TYPE_SHUTDOWN,
|
|
};
|
|
use crate::procpool::{plugin_progress_cb, ShmRegionView};
|
|
|
|
/// Slots per direction (input/output pools). Two slots per direction is
|
|
/// the floor; the client restarts the host with a bigger pool when a job
|
|
/// needs more (plugins with many clip inputs), so the idle cost stays
|
|
/// small.
|
|
pub const HOST_SLOTS: u32 = 2;
|
|
|
|
/// Initial per-slot capacity. Plugin frames are render-size F32; a 1080p
|
|
/// frame is ~8 MiB, and the pool grows (with a host restart) when a job
|
|
/// needs more. Small enough that an idle host costs little memory.
|
|
pub const HOST_SLOT_BYTES: usize = 8 * 1024 * 1024;
|
|
|
|
/// Consecutive crash budget before the client is permanently dead
|
|
/// (acceptance: three consecutive crashes fall back to purple frames).
|
|
pub const HOST_MAX_FAILURES: u32 = 3;
|
|
|
|
/// Configuration for [`OfxHost`].
|
|
#[derive(Clone, Debug)]
|
|
pub struct OfxHostConfig {
|
|
/// Host executable; `None` resolves `OAK_WORKER_BIN` / the
|
|
/// `oak-worker` binary next to the current executable.
|
|
pub host_bin: Option<PathBuf>,
|
|
/// Slots per direction.
|
|
pub slots: u32,
|
|
/// Initial per-slot capacity in bytes.
|
|
pub slot_bytes: usize,
|
|
/// Consecutive crash budget (a successful job resets it).
|
|
pub max_failures: u32,
|
|
/// Extra host argv (test crash hooks).
|
|
pub host_args: Vec<String>,
|
|
/// Extra host environment (test plugin discovery).
|
|
pub env: Vec<(String, String)>,
|
|
}
|
|
|
|
impl Default for OfxHostConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
host_bin: None,
|
|
slots: HOST_SLOTS,
|
|
slot_bytes: HOST_SLOT_BYTES,
|
|
max_failures: HOST_MAX_FAILURES,
|
|
host_args: Vec::new(),
|
|
env: Vec::new(),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
|
|
m.lock().unwrap_or_else(|e| e.into_inner())
|
|
}
|
|
|
|
/// Why a submit did not produce a frame: the host died (retryable) or it
|
|
/// reported a render failure (not a crash).
|
|
enum SubmitError {
|
|
/// EOF/pipe error: the host is gone and the job can be re-posted.
|
|
Died(String),
|
|
/// A real plugin/render failure (the evaluator falls back).
|
|
Failed(String),
|
|
}
|
|
|
|
struct HostInner {
|
|
child: Option<Child>,
|
|
stdin: Option<ChildStdin>,
|
|
input: Option<Arc<ShmRegionView>>,
|
|
output: Option<Arc<ShmRegionView>>,
|
|
/// Current per-slot capacity (grows on demand).
|
|
slot_bytes: usize,
|
|
/// Current slots per direction (grows on demand).
|
|
slots: u32,
|
|
generation: u64,
|
|
consecutive_failures: u32,
|
|
permanently_dead: bool,
|
|
reader: Option<JoinHandle<()>>,
|
|
/// Total host respawns (tests/observability).
|
|
restarts: u64,
|
|
}
|
|
|
|
struct WaitState {
|
|
results: HashMap<u64, std::result::Result<Frame, String>>,
|
|
dead: bool,
|
|
}
|
|
|
|
/// The single-host client.
|
|
pub struct OfxHost {
|
|
config: OfxHostConfig,
|
|
inner: Mutex<HostInner>,
|
|
wait: Arc<(Mutex<WaitState>, Condvar)>,
|
|
next_job: AtomicU64,
|
|
/// Serializes submissions: the one-job-in-flight contract is enforced
|
|
/// here, not left to the render thread being the only caller (a
|
|
/// concurrent second submit would interleave input-pool writes).
|
|
submit_lock: Mutex<()>,
|
|
}
|
|
|
|
/// One job's pre-read frames (read once, re-posted unchanged across
|
|
/// crashes).
|
|
struct HostJobFrames {
|
|
type_id: String,
|
|
time: f64,
|
|
effect_input_id: String,
|
|
inputs: Vec<(String, Frame)>,
|
|
/// The main source frame (sent separately from the named inputs; the
|
|
/// evaluator's `src` may not correspond to any named clip).
|
|
src: Option<Frame>,
|
|
values: Vec<WireEffectParam>,
|
|
}
|
|
|
|
impl HostJobFrames {
|
|
fn max_bytes(&self) -> usize {
|
|
self.inputs
|
|
.iter()
|
|
.map(|(_, f)| f.data.len())
|
|
.chain(self.src.iter().map(|f| f.data.len()))
|
|
.max()
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// Input-pool slots the job needs (named clips + the source).
|
|
fn input_slots(&self) -> u32 {
|
|
self.inputs.len() as u32 + u32::from(self.src.is_some())
|
|
}
|
|
}
|
|
|
|
impl OfxHost {
|
|
/// A client for `config`; the host process starts lazily on the first
|
|
/// submit (or via [`OfxHost::start`]).
|
|
pub fn new(config: OfxHostConfig) -> Result<Arc<Self>> {
|
|
Ok(Arc::new(Self {
|
|
config,
|
|
inner: Mutex::new(HostInner {
|
|
child: None,
|
|
stdin: None,
|
|
input: None,
|
|
output: None,
|
|
slot_bytes: 0,
|
|
slots: 0,
|
|
generation: 0,
|
|
consecutive_failures: 0,
|
|
permanently_dead: false,
|
|
reader: None,
|
|
restarts: 0,
|
|
}),
|
|
wait: Arc::new((
|
|
Mutex::new(WaitState {
|
|
results: HashMap::new(),
|
|
dead: false,
|
|
}),
|
|
Condvar::new(),
|
|
)),
|
|
next_job: AtomicU64::new(1),
|
|
submit_lock: Mutex::new(()),
|
|
}))
|
|
}
|
|
|
|
/// Start (or restart) the host now instead of on the first submit.
|
|
pub fn start(&self) -> Result<()> {
|
|
let mut inner = lock(&self.inner);
|
|
Self::ensure_started(&mut inner, &self.config, &self.wait)
|
|
}
|
|
|
|
/// True once the crash budget is exhausted.
|
|
pub fn is_permanently_dead(&self) -> bool {
|
|
lock(&self.inner).permanently_dead
|
|
}
|
|
|
|
/// Consecutive crashes since the last successful job.
|
|
pub fn failures(&self) -> u32 {
|
|
lock(&self.inner).consecutive_failures
|
|
}
|
|
|
|
/// Total host respawns (tests).
|
|
pub fn restarts(&self) -> u64 {
|
|
lock(&self.inner).restarts
|
|
}
|
|
|
|
/// Whether a host process is currently running.
|
|
pub fn is_running(&self) -> bool {
|
|
lock(&self.inner).child.is_some()
|
|
}
|
|
|
|
/// Submit one plugin job for `spec` (the evaluator's
|
|
/// [`JobSpec::Plugin`]) against `src`, returning the rendered frame.
|
|
/// Transparently re-posts the job across host crashes until the
|
|
/// crash budget is exhausted.
|
|
pub fn submit(&self, spec: &JobSpec, src: &Texture) -> Result<Frame> {
|
|
let _submit = lock(&self.submit_lock);
|
|
let frames = Self::read_job_frames(spec, src)?;
|
|
let job = self.next_job.fetch_add(1, Ordering::Relaxed);
|
|
loop {
|
|
match self.attempt(job, &frames) {
|
|
Ok(frame) => {
|
|
lock(&self.inner).consecutive_failures = 0;
|
|
return Ok(frame);
|
|
}
|
|
Err(SubmitError::Failed(err)) => return Err(Error::Failed(err)),
|
|
Err(SubmitError::Died(why)) => {
|
|
let mut inner = lock(&self.inner);
|
|
// Reap immediately: the crash budget may be exhausted
|
|
// here, and a dead child must not linger (is_running
|
|
// stays truthful) until shutdown.
|
|
Self::reap(&mut inner);
|
|
inner.consecutive_failures += 1;
|
|
if inner.consecutive_failures >= self.config.max_failures {
|
|
inner.permanently_dead = true;
|
|
return Err(Error::Failed(format!(
|
|
"OFX host crashed {} times in a row ({why}); giving up",
|
|
inner.consecutive_failures
|
|
)));
|
|
}
|
|
lock(&self.wait.0).dead = false;
|
|
// Loop: ensure_started spawns a fresh host and the job
|
|
// is sent again unchanged.
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Send `plugin_cancel` to the host (sticky flag; the host's progress
|
|
/// reporter then answers false at the plugin's next progressUpdate).
|
|
pub fn cancel(&self) {
|
|
let mut inner = lock(&self.inner);
|
|
if let Some(stdin) = inner.stdin.as_mut() {
|
|
let _ = write_message(stdin, &plugin_cancel_json());
|
|
let _ = stdin.flush();
|
|
}
|
|
}
|
|
|
|
/// Stop the host (idempotent). The client is permanently dead
|
|
/// afterwards, mirroring the evaluation fallback.
|
|
pub fn shutdown(&self) {
|
|
let mut inner = lock(&self.inner);
|
|
if let Some(stdin) = inner.stdin.as_mut() {
|
|
let _ = write_message(stdin, &serde_json::json!({ "type": TYPE_SHUTDOWN }));
|
|
let _ = stdin.flush();
|
|
}
|
|
Self::reap(&mut inner);
|
|
inner.permanently_dead = true;
|
|
}
|
|
|
|
/// One attempt: ensure a live host, write the inputs, send the job and
|
|
/// wait for its result (or the host's death).
|
|
fn attempt(&self, job: u64, frames: &HostJobFrames) -> std::result::Result<Frame, SubmitError> {
|
|
let mut inner = lock(&self.inner);
|
|
if inner.permanently_dead {
|
|
return Err(SubmitError::Failed(
|
|
"OFX host is permanently dead".to_string(),
|
|
));
|
|
}
|
|
// Size the pools before the first spawn (and restart the host when
|
|
// a job outgrows them; no in-flight job exists — submits are
|
|
// synchronous).
|
|
let needed_bytes = frames.max_bytes();
|
|
let needed_slots = frames.input_slots().max(1);
|
|
if needed_bytes > inner.slot_bytes || needed_slots > inner.slots {
|
|
if inner.child.is_some() {
|
|
Self::reap(&mut inner);
|
|
}
|
|
inner.slot_bytes = needed_bytes.next_power_of_two().max(self.config.slot_bytes);
|
|
inner.slots = needed_slots.next_power_of_two().max(self.config.slots);
|
|
}
|
|
if let Err(err) = Self::ensure_started(&mut inner, &self.config, &self.wait) {
|
|
return Err(SubmitError::Died(err.to_string()));
|
|
}
|
|
let input = inner
|
|
.input
|
|
.clone()
|
|
.ok_or_else(|| SubmitError::Died("OFX host has no input pool".to_string()))?;
|
|
let stdin = inner
|
|
.stdin
|
|
.as_mut()
|
|
.ok_or_else(|| SubmitError::Died("OFX host has no stdin".to_string()))?;
|
|
|
|
let mut inputs = Vec::with_capacity(frames.inputs.len());
|
|
for (name, frame) in &frames.inputs {
|
|
let slot = match write_input_frame(&input, frame) {
|
|
Ok(slot) => slot,
|
|
Err(err) => return Err(SubmitError::Failed(err)),
|
|
};
|
|
inputs.push(OfxInputRef {
|
|
name: name.clone(),
|
|
slot,
|
|
});
|
|
}
|
|
let src_slot = match &frames.src {
|
|
Some(frame) => match write_input_frame(&input, frame) {
|
|
Ok(slot) => Some(slot),
|
|
Err(err) => return Err(SubmitError::Failed(err)),
|
|
},
|
|
None => None,
|
|
};
|
|
let msg = OfxJobMsg {
|
|
job,
|
|
type_id: frames.type_id.clone(),
|
|
time: frames.time,
|
|
effect_input_id: frames.effect_input_id.clone(),
|
|
inputs,
|
|
values: frames.values.clone(),
|
|
src_slot,
|
|
};
|
|
if let Err(err) = write_message(stdin, &msg.to_json()).and_then(|_| stdin.flush()) {
|
|
return Err(SubmitError::Died(format!("OFX host stdin: {err}")));
|
|
}
|
|
drop(inner);
|
|
|
|
let (wait_lock, cv) = &*self.wait;
|
|
let mut wait = lock(wait_lock);
|
|
loop {
|
|
if let Some(result) = wait.results.remove(&job) {
|
|
return result.map_err(SubmitError::Failed);
|
|
}
|
|
if wait.dead {
|
|
return Err(SubmitError::Died("OFX host exited".to_string()));
|
|
}
|
|
wait = cv.wait(wait).unwrap_or_else(|e| e.into_inner());
|
|
}
|
|
}
|
|
|
|
/// Read the job's textures back to CPU frames once (the explicit OFX
|
|
/// boundary) and convert the scalar params to their wire form. The
|
|
/// `src` frame is kept separate: plugin jobs may name no clip for it
|
|
/// (montage) or the evaluator's `src` may be a clone the host cannot
|
|
/// re-identify by pointer.
|
|
fn read_job_frames(spec: &JobSpec, src: &Texture) -> Result<HostJobFrames> {
|
|
let JobSpec::Plugin {
|
|
type_id,
|
|
time,
|
|
effect_input_id,
|
|
inputs,
|
|
values,
|
|
..
|
|
} = spec
|
|
else {
|
|
return Err(Error::Invalid);
|
|
};
|
|
let frames = inputs
|
|
.iter()
|
|
.map(|(name, texture)| Ok((name.clone(), texture.to_frame()?)))
|
|
.collect::<Result<Vec<_>>>()?;
|
|
// A dummy source mirrors the in-process executor's rejection (the
|
|
// host receives no src and fails the job the same way).
|
|
let src = if src.is_dummy() {
|
|
None
|
|
} else {
|
|
Some(src.to_frame()?)
|
|
};
|
|
let values = values
|
|
.iter()
|
|
.filter_map(|(input, value)| {
|
|
WireNodeValue::from_node_value(value).map(|value| WireEffectParam {
|
|
input: input.clone(),
|
|
value,
|
|
})
|
|
})
|
|
.collect();
|
|
Ok(HostJobFrames {
|
|
type_id: type_id.clone(),
|
|
time: *time,
|
|
effect_input_id: effect_input_id.clone().unwrap_or_default(),
|
|
inputs: frames,
|
|
src,
|
|
values,
|
|
})
|
|
}
|
|
|
|
/// Create the pools and spawn the host. No-op when it is already up.
|
|
fn ensure_started(
|
|
inner: &mut HostInner,
|
|
config: &OfxHostConfig,
|
|
wait: &Arc<(Mutex<WaitState>, Condvar)>,
|
|
) -> Result<()> {
|
|
if inner.permanently_dead {
|
|
return Err(Error::Failed("OFX host is permanently dead".into()));
|
|
}
|
|
if inner.child.is_some() {
|
|
return Ok(());
|
|
}
|
|
if inner.slot_bytes == 0 {
|
|
inner.slot_bytes = config.slot_bytes;
|
|
}
|
|
if inner.slots == 0 {
|
|
inner.slots = config.slots.max(1);
|
|
}
|
|
inner.generation += 1;
|
|
let generation = inner.generation;
|
|
let slots = inner.slots.max(1);
|
|
let input = ShmRegionView::create(
|
|
&format!("oak-ofx-in-{}-{generation}", std::process::id()),
|
|
slots,
|
|
inner.slot_bytes,
|
|
)?;
|
|
let output = ShmRegionView::create(
|
|
&format!("oak-ofx-out-{}-{generation}", std::process::id()),
|
|
slots,
|
|
inner.slot_bytes,
|
|
)?;
|
|
let bin = resolve_host_bin(config)?;
|
|
let mut command = Command::new(&bin);
|
|
command
|
|
.arg("--ofx-host")
|
|
.args(&config.host_args)
|
|
.stdin(Stdio::piped())
|
|
.stdout(Stdio::piped())
|
|
.stderr(Stdio::inherit());
|
|
for (key, value) in &config.env {
|
|
command.env(key, value);
|
|
}
|
|
// A GUI Oak spawning a console child would flash a console window.
|
|
oak_core::miscutils::hide_console_window(&mut command);
|
|
let mut child = command
|
|
.spawn()
|
|
.map_err(|e| Error::Failed(format!("spawn OFX host {}: {e}", bin.display())))?;
|
|
let mut stdin = child
|
|
.stdin
|
|
.take()
|
|
.ok_or_else(|| Error::Failed("OFX host stdin not piped".into()))?;
|
|
let stdout = child
|
|
.stdout
|
|
.take()
|
|
.ok_or_else(|| Error::Failed("OFX host stdout not piped".into()))?;
|
|
let handshake = HandshakeMsg {
|
|
protocol_version: 1,
|
|
shm_key: output.key().to_string(),
|
|
input_shm_key: input.key().to_string(),
|
|
input_slots: slots as i32,
|
|
output_slots: slots as i32,
|
|
slot_data_bytes: inner.slot_bytes as i64,
|
|
input_slot_data_bytes: inner.slot_bytes as i64,
|
|
};
|
|
write_message(&mut stdin, &handshake.to_json())
|
|
.and_then(|_| stdin.flush())
|
|
.map_err(|e| Error::Failed(format!("OFX host handshake: {e}")))?;
|
|
let reader = spawn_reader(stdout, output.clone(), wait.clone());
|
|
inner.child = Some(child);
|
|
inner.stdin = Some(stdin);
|
|
inner.input = Some(input);
|
|
inner.output = Some(output);
|
|
inner.reader = Some(reader);
|
|
Ok(())
|
|
}
|
|
|
|
/// Kill/join the current generation and drop its pools.
|
|
fn reap(inner: &mut HostInner) {
|
|
if let Some(mut child) = inner.child.take() {
|
|
let _ = child.kill();
|
|
let _ = child.wait();
|
|
}
|
|
inner.stdin = None;
|
|
if let Some(reader) = inner.reader.take() {
|
|
let _ = reader.join();
|
|
}
|
|
inner.input = None;
|
|
inner.output = None;
|
|
inner.restarts += 1;
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Process-wide client slot
|
|
// ---------------------------------------------------------------------------
|
|
|
|
static CLIENT: OnceLock<Mutex<Option<Arc<OfxHost>>>> = OnceLock::new();
|
|
|
|
fn client_slot() -> &'static Mutex<Option<Arc<OfxHost>>> {
|
|
CLIENT.get_or_init(|| Mutex::new(None))
|
|
}
|
|
|
|
/// Install (or clear) the process-wide host client. Installed by the
|
|
/// render manager when the thread pipeline is selected; the evaluator
|
|
/// prefers the client over the in-process plugin executor while it is
|
|
/// installed.
|
|
pub fn install_client(client: Option<Arc<OfxHost>>) {
|
|
*lock(client_slot()) = client;
|
|
}
|
|
|
|
/// The installed client, if any.
|
|
pub fn client() -> Option<Arc<OfxHost>> {
|
|
lock(client_slot()).clone()
|
|
}
|
|
|
|
/// Broadcast `plugin_cancel` to the host (the app's cancel button; the
|
|
/// worker pool gets its own broadcast in `procpool`).
|
|
pub fn request_cancel_all() {
|
|
if let Some(host) = client() {
|
|
host.cancel();
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Resolve the host executable: the configured path, `OAK_WORKER_BIN`, or
|
|
/// an `oak-worker` binary next to the current executable (the host is a
|
|
/// mode of the worker binary, not a separate target).
|
|
fn resolve_host_bin(config: &OfxHostConfig) -> Result<PathBuf> {
|
|
if let Some(path) = &config.host_bin {
|
|
return Ok(path.clone());
|
|
}
|
|
if let Ok(path) = std::env::var("OAK_WORKER_BIN") {
|
|
return Ok(PathBuf::from(path));
|
|
}
|
|
let exe = std::env::current_exe()
|
|
.map_err(|e| Error::Failed(format!("resolve oak-worker: current exe: {e}")))?;
|
|
let candidate = exe
|
|
.parent()
|
|
.ok_or_else(|| Error::Failed("resolve oak-worker: no exe parent".into()))?
|
|
.join(format!("oak-worker{}", std::env::consts::EXE_SUFFIX));
|
|
if candidate.exists() {
|
|
return Ok(candidate);
|
|
}
|
|
Err(Error::Failed(format!(
|
|
"oak-worker binary not found at {}; set OfxHostConfig::host_bin or OAK_WORKER_BIN",
|
|
candidate.display()
|
|
)))
|
|
}
|
|
|
|
/// Write one input frame into the host's input pool, returning its slot.
|
|
fn write_input_frame(region: &ShmRegionView, frame: &Frame) -> std::result::Result<u32, String> {
|
|
let pool = region.pool();
|
|
let mut slot = 0u32;
|
|
if !unsafe { pool.acquire(&mut slot) } {
|
|
return Err("OFX host input pool is full".to_string());
|
|
}
|
|
let bytes = frame.data.len();
|
|
if bytes > region.slot_data_bytes() {
|
|
unsafe {
|
|
pool.release(slot);
|
|
}
|
|
return Err(format!(
|
|
"input frame is {bytes} bytes, larger than the OFX host slot ({})",
|
|
region.slot_data_bytes()
|
|
));
|
|
}
|
|
// SAFETY: `slot` was just acquired from this live pool; the copy and
|
|
// meta fill are the producer side of the SPSC protocol.
|
|
unsafe {
|
|
std::ptr::copy_nonoverlapping(frame.data.as_ptr(), pool.slot_data(slot), bytes);
|
|
let meta = &mut *pool.meta(slot);
|
|
*meta = Default::default();
|
|
meta.width = frame.width;
|
|
meta.height = frame.height;
|
|
meta.format = PixelFormat::F32 as i32;
|
|
meta.channel_count = 4;
|
|
meta.linesize = frame.linesize_bytes() as i32;
|
|
meta.data_size = bytes as i32;
|
|
meta.time_num = frame.timestamp.numerator();
|
|
meta.time_den = frame.timestamp.denominator().max(1);
|
|
if !pool.publish(slot) {
|
|
pool.release(slot);
|
|
return Err("OFX host input publish failed".to_string());
|
|
}
|
|
}
|
|
Ok(slot)
|
|
}
|
|
|
|
/// Copy one output frame out of the host's output pool and release the
|
|
/// slot (the consumer side of the SPSC protocol).
|
|
fn read_output_frame(
|
|
region: &ShmRegionView,
|
|
slot: u32,
|
|
) -> std::result::Result<Frame, String> {
|
|
let pool = region.pool();
|
|
let mut consumed = 0u32;
|
|
if !unsafe { pool.consume(&mut consumed) } {
|
|
return Err("OFX host output slot missing".to_string());
|
|
}
|
|
if consumed != slot {
|
|
// Keep the pool consistent: recycle the unexpected entry.
|
|
unsafe {
|
|
pool.release(consumed);
|
|
}
|
|
return Err(format!(
|
|
"OFX host output slot mismatch: expected {slot}, got {consumed}"
|
|
));
|
|
}
|
|
let meta = region.meta_copy(slot);
|
|
if meta.width <= 0 || meta.height <= 0 || meta.data_size < 0 {
|
|
unsafe {
|
|
pool.release(slot);
|
|
}
|
|
return Err("OFX host returned invalid frame metadata".to_string());
|
|
}
|
|
let len = (meta.data_size as usize).min(region.slot_data_bytes());
|
|
let data = region.slot_bytes(slot)[..len].to_vec();
|
|
unsafe {
|
|
pool.release(slot);
|
|
}
|
|
let mut frame = Frame::new();
|
|
let mut pod = VideoParamsPod::default();
|
|
pod.width = meta.width;
|
|
pod.height = meta.height;
|
|
pod.format = PixelFormat::F32 as i32;
|
|
frame.set_video_params(pod);
|
|
frame.data = data;
|
|
Ok(frame)
|
|
}
|
|
|
|
/// The host's stdout reader: forwards progress, delivers results and
|
|
/// flags death for the blocked submitter.
|
|
fn spawn_reader(
|
|
stdout: std::process::ChildStdout,
|
|
output: Arc<ShmRegionView>,
|
|
wait: Arc<(Mutex<WaitState>, Condvar)>,
|
|
) -> JoinHandle<()> {
|
|
std::thread::Builder::new()
|
|
.name("oak-ofx-host-reader".into())
|
|
.spawn(move || {
|
|
let mut reader = BufReader::new(stdout);
|
|
let mut line = String::new();
|
|
loop {
|
|
line.clear();
|
|
match reader.read_line(&mut line) {
|
|
Ok(0) | Err(_) => break,
|
|
Ok(_) => {}
|
|
}
|
|
let Ok(value) = serde_json::from_str::<serde_json::Value>(&line) else {
|
|
continue;
|
|
};
|
|
match value.get("type").and_then(|t| t.as_str()) {
|
|
Some(TYPE_OFX_RESULT) => {
|
|
let msg: OfxResultMsg = serde_json::from_value(value).unwrap_or_default();
|
|
let result = if msg.slot >= 0 {
|
|
read_output_frame(&output, msg.slot as u32)
|
|
} else {
|
|
Err(msg.error)
|
|
};
|
|
let (state, cv) = &*wait;
|
|
lock(state).results.insert(msg.job, result);
|
|
cv.notify_all();
|
|
}
|
|
Some(TYPE_PLUGIN_PROGRESS) => {
|
|
let msg: PluginProgressMsg = serde_json::from_value(value).unwrap_or_default();
|
|
if let Some(cb) = plugin_progress_cb() {
|
|
cb(msg.label, msg.message, msg.fraction);
|
|
}
|
|
}
|
|
Some(TYPE_ERROR) => {
|
|
let message = value
|
|
.get("message")
|
|
.and_then(|m| m.as_str())
|
|
.unwrap_or("unknown OFX host error");
|
|
eprintln!("OFX host error: {message}");
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
let (state, cv) = &*wait;
|
|
lock(state).dead = true;
|
|
cv.notify_all();
|
|
})
|
|
.expect("spawn OFX host reader")
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::eval::generate_frame;
|
|
use oak_core::Rational;
|
|
|
|
fn plugin_spec(type_id: &str) -> JobSpec {
|
|
JobSpec::Plugin {
|
|
instance: 0,
|
|
type_id: type_id.to_string(),
|
|
time: 0.0,
|
|
effect_input_id: None,
|
|
inputs: Vec::new(),
|
|
values: Vec::new(),
|
|
}
|
|
}
|
|
|
|
fn small_texture() -> Texture {
|
|
Texture::wrap_frame(
|
|
generate_frame(Rational::new(0, 1), (4, 4), PixelFormat::F32).unwrap(),
|
|
)
|
|
}
|
|
|
|
#[test]
|
|
#[cfg(unix)]
|
|
fn submit_exhausts_the_crash_budget() {
|
|
let host = OfxHost::new(OfxHostConfig {
|
|
host_bin: Some(PathBuf::from("/bin/false")),
|
|
max_failures: 3,
|
|
..Default::default()
|
|
})
|
|
.unwrap();
|
|
let spec = plugin_spec("org.oak.missing");
|
|
let src = small_texture();
|
|
assert!(
|
|
host.submit(&spec, &src).is_err(),
|
|
"a host that never comes up must fail the submit"
|
|
);
|
|
assert_eq!(host.failures(), 3, "three consecutive crashes");
|
|
assert!(host.is_permanently_dead());
|
|
// Further submissions fail fast without spawning again.
|
|
let restarts = host.restarts();
|
|
assert!(host.submit(&spec, &src).is_err());
|
|
assert_eq!(host.restarts(), restarts, "no more spawns after the budget");
|
|
}
|
|
|
|
#[test]
|
|
#[cfg(unix)]
|
|
fn shutdown_is_idempotent_and_marks_dead() {
|
|
let host = OfxHost::new(OfxHostConfig {
|
|
host_bin: Some(PathBuf::from("/bin/false")),
|
|
..Default::default()
|
|
})
|
|
.unwrap();
|
|
host.shutdown();
|
|
host.shutdown();
|
|
assert!(host.is_permanently_dead());
|
|
assert!(!host.is_running());
|
|
}
|
|
|
|
#[test]
|
|
fn read_job_frames_keeps_the_source_mapping() {
|
|
let mut source = generate_frame(Rational::new(0, 1), (2, 2), PixelFormat::F32).unwrap();
|
|
source.data[0] = 0x7F;
|
|
let source = Texture::wrap_frame(source);
|
|
// The same texture value appears under the effect input name.
|
|
let spec = JobSpec::Plugin {
|
|
instance: 0,
|
|
type_id: "org.oak.test-plugin".into(),
|
|
time: 1.0,
|
|
effect_input_id: Some("Source".into()),
|
|
inputs: vec![("Source".to_string(), source.clone())],
|
|
values: vec![(
|
|
"gain".to_string(),
|
|
oak_node::value::NodeValue::Float(0.5),
|
|
)],
|
|
};
|
|
let frames = OfxHost::read_job_frames(&spec, &source).unwrap();
|
|
assert!(frames.src.is_some(), "a real source is sent separately");
|
|
assert_eq!(frames.src.as_ref().unwrap().data[0], 0x7F);
|
|
assert_eq!(frames.effect_input_id, "Source");
|
|
assert_eq!(frames.inputs.len(), 1);
|
|
assert_eq!(frames.inputs[0].1.data[0], 0x7F);
|
|
assert_eq!(frames.input_slots(), 2, "named clip + source");
|
|
assert_eq!(
|
|
frames.values,
|
|
vec![WireEffectParam {
|
|
input: "gain".into(),
|
|
value: WireNodeValue::Float(0.5),
|
|
}]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn wait_state_wakes_on_result() {
|
|
// The wait protocol itself (independent of a real host).
|
|
let host = OfxHost::new(OfxHostConfig::default()).unwrap();
|
|
let (state, cv) = &*host.wait;
|
|
let mut wait = lock(state);
|
|
// No result yet and not dead: a timed wait returns empty-handed.
|
|
let (guard, timeout) = cv
|
|
.wait_timeout(wait, std::time::Duration::from_millis(10))
|
|
.unwrap();
|
|
wait = guard;
|
|
assert!(timeout.timed_out());
|
|
assert!(!wait.dead);
|
|
// A result wakes a waiter.
|
|
let frame = Frame::dummy();
|
|
wait.results.insert(7, Ok(frame));
|
|
let (guard, _) = cv
|
|
.wait_timeout(wait, std::time::Duration::from_millis(10))
|
|
.unwrap();
|
|
wait = guard;
|
|
assert!(wait.results.remove(&7).is_some());
|
|
}
|
|
}
|