oak-team/oak-editor
3
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(cli,worker): cut liboakengine, link module rlibs directly (M14 R2)

Browse Source 
- oak-cli: new engine.rs assembly layer maps every facade call to
  module Rust APIs (oaknode graph/serializer, oaktimeline commands,
  oakrender ticket arena, oaktask ExportTask, oakcommon config); the
  ffi/optional/host layers and build.rs link config are gone
- oak-worker: the worker session + POSIX shm transport moved into the
  crate (oakrender backend + serde_json control plane); no dylib
- both binaries carry zero liboakengine references (otool verified);
  tests green (30 cli / 41 worker)
This commit is contained in:
Mike-Solar
2026-08-16 21:16:27 +08:00
parent 8a2e45225f
commit a45a7af2ac
24 changed files with 3713 additions and 3196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/oak-worker/src/engine_ipc.rs
-652
View File
@@ -1,652 +0,0 @@
// 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 `liboakengine` C-ABI surface, as consumed by the worker binary.
//!
//! The engine facade (crates/oakengine) is the cdylib and the ONLY
//! implementation owner of the worker/IPC runtime; this crate is a pure
//! C-ABI consumer. Everything here goes through the `extern "C"`
//! declarations of the `oakengine_*` symbols below — never a direct Rust
//! call into an oak* module crate.
//!
//! The module provides three layers:
//!
//! - **The externs.** `#[link(name = "oakengine", kind = "dylib")]`
//! declarations of the frozen `oakengine_worker_*` and
//! `oakengine_ipc_*` exports (`engine/include/oakengine/worker.h` and
//! `ipc.h`), plus the opaque `Oak*` handle mirrors and the POD
//! [`FrameSlotMeta`] / [`ShmMode`] mirrors the signatures reference.
//! build.rs points the linker at the built dylib and embeds its rpath.
//! - **In-process wrapper types.** [`SharedMemoryRegion`] and
//! [`FrameSlotPool`] wrap the opaque handles with `Drop` and a safe
//! surface, so [`crate::transport`]/[`crate::session`] (and their
//! tests) can attach real shared-memory pools through the engine.
//! - **The wire protocol.** [`HandshakeMsg`]/[`RenderFrameMsg`]/
//! [`LoadGraphMsg`] + [`error_message`] + the `TYPE_*` constants —
//! serde-only structs matching the engine's NDJSON control plane; the
//! worker-side session mirror validates and builds these lines
//! locally.
#![allow(dead_code)]
use std::ffi::{c_char, c_int, c_void};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
// ---------------------------------------------------------------------------
// Opaque handle mirrors (engine/include/oakengine/{worker,ipc}.h)
// ---------------------------------------------------------------------------
/// Opaque `OakWorkerSession` handle (worker.h). The engine owns the box;
/// this crate only ever sees the pointer.
#[repr(C)]
pub struct OakWorkerSession {
_opaque: [u8; 0],
}
/// Opaque `OakSharedMemoryRegion` handle (ipc.h).
#[repr(C)]
pub struct OakSharedMemoryRegion {
_opaque: [u8; 0],
}
/// Opaque `OakFrameSlotPool` handle (ipc.h).
#[repr(C)]
pub struct OakFrameSlotPool {
_opaque: [u8; 0],
}
/// `OAK_IPC_SHM_KEY_CAP` — capacity of shm key strings (ipc.h), incl. NUL.
pub const OAK_IPC_SHM_KEY_CAP: usize = 128;
/// `OAK_IPC_COLORSPACE_CAP` — capacity of `oak_frame_slot_meta::colorspace`.
pub const OAK_IPC_COLORSPACE_CAP: usize = 128;
/// Per-slot metadata describing the frame currently occupying a slot —
/// field-for-field `oak_frame_slot_meta` from `engine/include/oakengine/ipc.h`
/// (the POD lives in shared memory; the layout is the version-1 wire
/// protocol).
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FrameSlotMeta {
/// Caller-defined tag (ticket id, or footage stream hash).
pub id: i64,
/// Frame timestamp numerator.
pub time_num: i64,
/// Frame timestamp denominator.
pub time_den: i64,
/// Frame width.
pub width: i32,
/// Frame height.
pub height: i32,
/// `PixelFormat::Format` value.
pub format: i32,
/// Channel count.
pub channel_count: i32,
/// Bytes per scanline (stride).
pub linesize: i32,
/// Valid bytes written into the slot's data block.
pub data_size: i32,
/// Input colorspace name.
pub colorspace: [c_char; OAK_IPC_COLORSPACE_CAP],
}
/// `OAK_IPC_SHM_MODE_CREATE` / `OAK_IPC_SHM_MODE_ATTACH` (ipc.h).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ShmMode {
/// Create (and own) the segment. Fails if it already exists; the owner
/// unlinks it on close.
Create,
/// Attach to a segment created by the peer. Does not unlink on close.
Attach,
}
// ---------------------------------------------------------------------------
// C ABI exports (engine/include/oakengine/{worker,ipc}.h)
// ---------------------------------------------------------------------------
// The built `liboakengine` dylib (crates/oakengine, cdylib). build.rs
// emits the link-search path and rpath for the target profile dir.
#[link(name = "oakengine", kind = "dylib")]
extern "C" {
// ---- engine/include/oakengine/worker.h ----
/// Full render-worker main (argv-based `--backend` scanning, startup
/// handshake, NDJSON control loop). Returns the process exit code.
pub fn oakengine_worker_main(argc: c_int, argv: *mut *mut c_char) -> c_int;
/// Create a session for the given render backend (NULL = none).
pub fn oakengine_worker_session_create(backend: *const c_char) -> *mut OakWorkerSession;
/// Free a session. NULL no-op.
pub fn oakengine_worker_session_free(self_: *mut OakWorkerSession);
/// 1 when the session holds an initialized render backend.
pub fn oakengine_worker_session_has_renderer(self_: *const OakWorkerSession) -> c_int;
/// Load the runtime services; 1 on success, 0 for a NULL session.
pub fn oakengine_worker_session_initialize_runtime(self_: *mut OakWorkerSession) -> c_int;
/// Build the startup handshake (buf/size convention; returns the
/// required size, -1 on failure).
pub fn oakengine_worker_session_startup_handshake(
self_: *const OakWorkerSession,
buf: *mut c_char,
buf_size: c_int,
) -> c_int;
/// Handle one NDJSON control line; 0 for "no response", the response
/// length for a response, -1 on a fatal handler failure.
pub fn oakengine_worker_session_handle_json(
self_: *mut OakWorkerSession,
line: *const c_char,
response_buf: *mut c_char,
response_buf_size: c_int,
) -> c_int;
/// 1 once a shutdown control message has been received.
pub fn oakengine_worker_session_shutdown_requested(self_: *const OakWorkerSession) -> c_int;
// ---- engine/include/oakengine/ipc.h ----
/// Allocate an empty (invalid) region object.
pub fn oakengine_ipc_shm_create() -> *mut OakSharedMemoryRegion;
/// Free a region object. NULL no-op.
pub fn oakengine_ipc_shm_free(self_: *mut OakSharedMemoryRegion);
/// Open the segment; 1 on success, 0 on failure (shm_error carries the
/// reason). Mode: 0 = create, 1 = attach.
pub fn oakengine_ipc_shm_open(
self_: *mut OakSharedMemoryRegion,
key: *const c_char,
size: usize,
mode: c_int,
) -> c_int;
/// Unmap and (if owner) unlink.
pub fn oakengine_ipc_shm_close(self_: *mut OakSharedMemoryRegion);
/// 1 when the region holds a live mapping.
pub fn oakengine_ipc_shm_is_valid(self_: *const OakSharedMemoryRegion) -> c_int;
/// The mapped data pointer (NULL when invalid).
pub fn oakengine_ipc_shm_data(self_: *mut OakSharedMemoryRegion) -> *mut c_void;
/// Mapping size in bytes.
pub fn oakengine_ipc_shm_size(self_: *const OakSharedMemoryRegion) -> usize;
/// The key the region was opened with (buf/size convention).
pub fn oakengine_ipc_shm_key(
self_: *const OakSharedMemoryRegion,
buf: *mut c_char,
buf_size: c_int,
) -> c_int;
/// Reason of the last failed open (buf/size convention).
pub fn oakengine_ipc_shm_error(
self_: *const OakSharedMemoryRegion,
buf: *mut c_char,
buf_size: c_int,
) -> c_int;
/// Build a unique segment key ("olive-rw-<pid>-<index>", buf/size
/// convention).
pub fn oakengine_ipc_shm_make_key(
owner_pid: i64,
worker_index: c_int,
buf: *mut c_char,
buf_size: c_int,
) -> c_int;
/// Total bytes a region must provide to back a pool of
/// `slot_count` x `slot_data_bytes`.
pub fn oakengine_ipc_framepool_bytes_needed(slot_count: u32, slot_data_bytes: usize) -> usize;
/// Lay out and initialize a brand-new pool over `mem` (owner side,
/// once). The handle does not own `mem`.
pub fn oakengine_ipc_framepool_create(
mem: *mut c_void,
slot_count: u32,
slot_data_bytes: usize,
) -> *mut OakFrameSlotPool;
/// Map an existing pool (peer side). NULL when the segment holds no
/// pool.
pub fn oakengine_ipc_framepool_attach(mem: *mut c_void) -> *mut OakFrameSlotPool;
/// Copy the view (same shared memory, independent handle).
pub fn oakengine_ipc_framepool_copy(self_: *const OakFrameSlotPool) -> *mut OakFrameSlotPool;
/// Free a pool view. NULL no-op.
pub fn oakengine_ipc_framepool_free(self_: *mut OakFrameSlotPool);
/// 1 when the pool was attached to a valid pool header.
pub fn oakengine_ipc_framepool_is_valid(self_: *const OakFrameSlotPool) -> c_int;
/// Number of slots in the pool.
pub fn oakengine_ipc_framepool_slot_count(self_: *const OakFrameSlotPool) -> u32;
/// Bytes available in every slot's pixel-data block.
pub fn oakengine_ipc_framepool_slot_data_bytes(self_: *const OakFrameSlotPool) -> usize;
/// Take a free slot; 1 on success (`*index` set).
pub fn oakengine_ipc_framepool_acquire(self_: *mut OakFrameSlotPool, index: *mut u32) -> c_int;
/// Pointer to a slot's pixel data block.
pub fn oakengine_ipc_framepool_slot_data(
self_: *mut OakFrameSlotPool,
index: u32,
) -> *mut c_void;
/// Immutable pixel data for a slot.
pub fn oakengine_ipc_framepool_slot_data_const(
self_: *const OakFrameSlotPool,
index: u32,
) -> *const c_void;
/// Mutable per-slot metadata (borrowed).
pub fn oakengine_ipc_framepool_meta(
self_: *mut OakFrameSlotPool,
index: u32,
) -> *mut FrameSlotMeta;
/// Immutable per-slot metadata.
pub fn oakengine_ipc_framepool_meta_const(
self_: *const OakFrameSlotPool,
index: u32,
) -> *const FrameSlotMeta;
/// Publish a filled slot; 1 on success.
pub fn oakengine_ipc_framepool_publish(self_: *mut OakFrameSlotPool, index: u32) -> c_int;
/// Take the next published slot; 1 on success (`*index` set).
pub fn oakengine_ipc_framepool_consume(self_: *mut OakFrameSlotPool, index: *mut u32) -> c_int;
/// Return a consumed slot to the free pool; 1 on success.
pub fn oakengine_ipc_framepool_release(self_: *mut OakFrameSlotPool, index: u32) -> c_int;
}
// ---------------------------------------------------------------------------
// Wire protocol (the engine's NDJSON control plane, mirrored locally)
// ---------------------------------------------------------------------------
/// `"handshake"`.
pub const TYPE_HANDSHAKE: &str = "handshake";
/// `"load_graph"`.
pub const TYPE_LOAD_GRAPH: &str = "load_graph";
/// `"render_frame"`.
pub const TYPE_RENDER_FRAME: &str = "render_frame";
/// `"frame_ready"`.
pub const TYPE_FRAME_READY: &str = "frame_ready";
/// `"cancel"`.
pub const TYPE_CANCEL: &str = "cancel";
/// `"graph_update"`.
pub const TYPE_GRAPH_UPDATE: &str = "graph_update";
/// `"shutdown"`.
pub const TYPE_SHUTDOWN: &str = "shutdown";
/// `"error"`.
pub const TYPE_ERROR: &str = "error";
/// `handshake` — field-for-field equivalent of `oak_ipc_handshake`
/// (ipc.h). Wire field names match the C++ serializer.
#[derive(Serialize, Deserialize, Default, Debug, Clone)]
#[serde(default)]
pub struct HandshakeMsg {
/// Protocol version.
pub protocol_version: i32,
/// Worker->main output shared-memory segment key.
pub shm_key: String,
/// Main->worker input shared-memory segment key (optional).
pub input_shm_key: String,
/// Number of main->worker input frame slots.
pub input_slots: i32,
/// Number of worker->main output frame slots.
pub output_slots: i32,
/// Per-output-slot pixel block size.
pub slot_data_bytes: i64,
/// Per-input-slot pixel block size.
pub input_slot_data_bytes: i64,
}
impl HandshakeMsg {
/// The worker's startup handshake (`worker.cpp startup_handshake()`).
pub fn to_json(&self) -> Value {
json!({
"type": TYPE_HANDSHAKE,
"protocol_version": self.protocol_version,
"shm_key": self.shm_key,
"input_shm_key": self.input_shm_key,
"input_slots": self.input_slots,
"output_slots": self.output_slots,
"slot_data_bytes": self.slot_data_bytes,
"input_slot_data_bytes": self.input_slot_data_bytes,
})
}
}
/// `render_frame` — request a frame render. Wire names per ipcmessage.cpp:
/// `ticket`, `node`, `channels` (not the ipc.h POD names).
#[derive(Serialize, Deserialize, Default, Debug, Clone)]
#[serde(default)]
pub struct RenderFrameMsg {
/// Correlates with the eventual frame_ready.
pub ticket: i64,
/// Viewer node stable uuid in the loaded graph.
pub node: String,
pub time_num: i64,
pub time_den: i64,
/// Forced output size (0 = graph default).
pub width: i32,
pub height: i32,
/// Forced PixelFormat (-1 = default).
pub format: i32,
/// Channel count (0 = default).
pub channels: i32,
/// RenderMode.
pub mode: i32,
/// Optional decoded input slot (-1 = none).
pub input_slot: i32,
/// Ordered decoded input slots.
pub input_slots: Vec<i32>,
/// Output color transform present?
pub has_color_transform: bool,
pub color_is_display: bool,
pub color_output: String,
pub color_view: String,
pub color_look: String,
}
/// `load_graph` — path to a temporary file holding the serialized graph.
#[derive(Serialize, Deserialize, Default, Debug, Clone)]
#[serde(default)]
pub struct LoadGraphMsg {
pub path: String,
}
/// Build a worker-side error report, mirroring `error_message()` in
/// worker.cpp: `{"type":"error","message":...}` plus `"ticket"` when
/// non-zero.
pub fn error_message(message: &str, ticket: Option<i64>) -> Value {
match ticket.filter(|t| *t != 0) {
Some(t) => json!({ "type": TYPE_ERROR, "message": message, "ticket": t }),
None => json!({ "type": TYPE_ERROR, "message": message }),
}
}
// ---------------------------------------------------------------------------
// In-process wrapper types over the opaque C-ABI handles
// ---------------------------------------------------------------------------
/// Two-stage read of a buf/size-convention string getter (the engine's
/// `write_string` reports the length excluding the NUL for a NULL buffer).
unsafe fn region_string(
region: *const OakSharedMemoryRegion,
getter: unsafe extern "C" fn(*const OakSharedMemoryRegion, *mut c_char, c_int) -> c_int,
) -> String {
// SAFETY: the engine's buf/size convention: a NULL/0 buffer only
// queries the required length.
let len = unsafe { getter(region, std::ptr::null_mut(), 0) };
if len <= 0 {
return String::new();
}
let mut buf = vec![0 as c_char; len as usize + 1];
// SAFETY: `buf` provides len + 1 writable bytes (length + NUL).
unsafe { getter(region, buf.as_mut_ptr(), buf.len() as c_int) };
// SAFETY: the engine NUL-terminates what it writes into `buf`.
unsafe { std::ffi::CStr::from_ptr(buf.as_ptr()) }
.to_string_lossy()
.into_owned()
}
/// A named, fixed-size POSIX shared-memory segment mapped inside the
/// engine process (C-ABI wrapper around `OakSharedMemoryRegion`). One side
/// opens the segment in [`ShmMode::Create`] (owner, unlinks on close); the
/// peer attaches by key. The wrapper owns the engine-side handle and
/// closes/frees it on drop.
pub struct SharedMemoryRegion {
handle: *mut OakSharedMemoryRegion,
}
impl SharedMemoryRegion {
/// An empty (invalid) region.
pub fn new() -> SharedMemoryRegion {
// SAFETY: shm_create is the factory for owned handles (never
// observes caller memory).
SharedMemoryRegion {
handle: unsafe { oakengine_ipc_shm_create() },
}
}
/// Build a unique segment key for a worker, e.g.
/// "olive-rw-<pid>-<index>" (the engine's `SharedMemoryRegion::make_key`
/// behind the C ABI). Centralized so the owner and the spawned worker
/// agree on the same name.
pub fn make_key(owner_pid: i64, worker_index: i32) -> String {
let mut buf = [0 as c_char; OAK_IPC_SHM_KEY_CAP];
// SAFETY: `buf` is writable and sized by the engine's key capacity.
let n = unsafe {
oakengine_ipc_shm_make_key(owner_pid, worker_index, buf.as_mut_ptr(), buf.len() as c_int)
};
if n <= 0 {
return String::new();
}
// SAFETY: the engine NUL-terminates the key within `buf`.
unsafe { std::ffi::CStr::from_ptr(buf.as_ptr()) }
.to_string_lossy()
.into_owned()
}
/// Open the segment identified by `key` with the given `size` in bytes
/// (through the engine). Returns true on success; on failure
/// [`Self::error`] carries a human-readable reason.
pub fn open(&mut self, key: &str, size: usize, mode: ShmMode) -> bool {
let Ok(key_c) = std::ffi::CString::new(key) else {
return false;
};
// SAFETY: `key_c` is a valid C string; `self.handle` is the live
// handle this wrapper owns.
unsafe { oakengine_ipc_shm_open(self.handle, key_c.as_ptr(), size, mode as c_int) == 1 }
}
/// Unmap and (if owner) unlink the segment.
pub fn close(&mut self) {
// SAFETY: the engine's shm_close is a NULL no-op.
unsafe { oakengine_ipc_shm_close(self.handle) };
}
/// True when the region holds a live mapping.
pub fn is_valid(&self) -> bool {
// SAFETY: the engine's shm_is_valid returns 0 for NULL handles.
unsafe { oakengine_ipc_shm_is_valid(self.handle) == 1 }
}
/// The mapped data pointer (null when invalid).
pub fn data(&self) -> *mut u8 {
// SAFETY: the engine returns NULL for invalid regions.
unsafe { oakengine_ipc_shm_data(self.handle) as *mut u8 }
}
/// The mapping size in bytes.
pub fn size(&self) -> usize {
// SAFETY: the engine returns 0 for NULL handles.
unsafe { oakengine_ipc_shm_size(self.handle) }
}
/// The key the region was opened with.
pub fn key(&self) -> String {
// SAFETY: region_string follows the engine's buf/size convention.
unsafe { region_string(self.handle, oakengine_ipc_shm_key) }
}
/// Human-readable reason of the last failed open.
pub fn error(&self) -> String {
// SAFETY: region_string follows the engine's buf/size convention.
unsafe { region_string(self.handle, oakengine_ipc_shm_error) }
}
}
impl Drop for SharedMemoryRegion {
fn drop(&mut self) {
// Close first so the owner unlinks the segment, then free the
// engine-side box. Both are NULL no-ops in the engine.
// SAFETY: `self.handle` is the handle this wrapper owns and is not
// used after this.
unsafe {
oakengine_ipc_shm_close(self.handle);
oakengine_ipc_shm_free(self.handle);
}
}
}
/// A fixed-size pool of equal-sized frame slots in shared memory with
/// lock-free hand-off (C-ABI wrapper around `OakFrameSlotPool`). The pool
/// does NOT own the memory — it is a view the engine attaches over a
/// mapped [`SharedMemoryRegion`]. Lifecycle: the filler `acquire`s a free
/// slot, writes meta + pixels, then `publish`es it; the drainer `consume`s
/// the next ready slot, reads it, and `release`s it back to the free ring.
///
/// `Clone` mirrors the engine's pool-view copy (same shared memory,
/// independent handle).
pub struct FrameSlotPool {
handle: *mut OakFrameSlotPool,
}
impl FrameSlotPool {
/// Total bytes a region must provide to back a pool of
/// `slot_count` x `slot_data_bytes` (the engine's
/// `FrameSlotPool::bytes_needed` behind the C ABI).
pub fn bytes_needed(slot_count: u32, slot_data_bytes: usize) -> usize {
// SAFETY: pure computation, no handles involved.
unsafe { oakengine_ipc_framepool_bytes_needed(slot_count, slot_data_bytes) }
}
/// Lay out and initialize a brand-new pool over `mem` (owner side,
/// once) through the engine.
///
/// # Safety
/// `mem` must be a valid, writable, aligned buffer of at least
/// [`Self::bytes_needed`] bytes, not concurrently written during this
/// call.
pub unsafe fn create(mem: *mut u8, slot_count: u32, slot_data_bytes: usize) -> FrameSlotPool {
// SAFETY: forwarded to the engine's create contract.
FrameSlotPool {
handle: unsafe {
oakengine_ipc_framepool_create(mem as *mut c_void, slot_count, slot_data_bytes)
},
}
}
/// Map an existing, already-initialized pool (peer side) through the
/// engine. The returned pool reports `is_valid() == false` when the
/// segment does not contain a pool.
///
/// # Safety
/// `mem` must point to a mapped segment that either contains a pool or
/// is an arbitrary buffer whose first bytes are readable.
pub unsafe fn attach(mem: *mut u8) -> FrameSlotPool {
// SAFETY: forwarded to the engine's attach contract.
FrameSlotPool {
handle: unsafe { oakengine_ipc_framepool_attach(mem as *mut c_void) },
}
}
/// True when the pool was attached to a segment containing a valid pool
/// header.
pub fn is_valid(&self) -> bool {
// SAFETY: the engine returns 0 for NULL handles.
unsafe { oakengine_ipc_framepool_is_valid(self.handle) == 1 }
}
/// Number of slots in the pool (0 for an invalid pool).
pub fn slot_count(&self) -> u32 {
// SAFETY: the engine returns 0 for NULL handles.
unsafe { oakengine_ipc_framepool_slot_count(self.handle) }
}
/// Bytes available in every slot's pixel-data block (0 for invalid).
pub fn slot_data_bytes(&self) -> usize {
// SAFETY: the engine returns 0 for NULL handles.
unsafe { oakengine_ipc_framepool_slot_data_bytes(self.handle) }
}
/// Take ownership of a free slot. Returns false (leaving `index`
/// untouched) if none is free.
///
/// # Safety
/// The pool must be a valid view of a live segment.
pub unsafe fn acquire(&self, index: &mut u32) -> bool {
// SAFETY: forwarded to the engine's acquire contract.
unsafe { oakengine_ipc_framepool_acquire(self.handle, index) == 1 }
}
/// Pointer to a slot's pixel data block (`slot_data_bytes` available).
///
/// # Safety
/// `index` must be in `0..slot_count`; the pool must be a valid view of
/// a live segment.
pub unsafe fn slot_data(&self, index: u32) -> *mut u8 {
// SAFETY: forwarded to the engine's slot_data contract.
unsafe { oakengine_ipc_framepool_slot_data(self.handle, index) as *mut u8 }
}
/// Immutable pixel data for a slot.
///
/// # Safety
/// `index` must be in `0..slot_count`; the pool must be a valid view of
/// a live segment.
pub unsafe fn slot_data_const(&self, index: u32) -> *const u8 {
// SAFETY: forwarded to the engine's slot_data_const contract.
unsafe { oakengine_ipc_framepool_slot_data_const(self.handle, index) as *const u8 }
}
/// Mutable metadata for a slot. The filler writes this before
/// [`Self::publish`]. The returned pointer addresses shared memory; it
/// is borrowed, not owned.
///
/// # Safety
/// `index` must be in `0..slot_count`; the pool must be a valid view of
/// a live segment.
pub unsafe fn meta(&self, index: u32) -> *mut FrameSlotMeta {
// SAFETY: forwarded to the engine's meta contract.
unsafe { oakengine_ipc_framepool_meta(self.handle, index) }
}
/// Immutable metadata for a slot (drainer side).
///
/// # Safety
/// `index` must be in `0..slot_count`; the pool must be a valid view of
/// a live segment.
pub unsafe fn meta_const(&self, index: u32) -> *const FrameSlotMeta {
// SAFETY: forwarded to the engine's meta_const contract.
unsafe { oakengine_ipc_framepool_meta_const(self.handle, index) }
}
/// Publish a filled slot to the drainer. Must follow a successful
/// [`Self::acquire`] of `index`. Returns false if the ready ring is
/// full.
///
/// # Safety
/// `index` must be a slot previously acquired and not yet released.
pub unsafe fn publish(&self, index: u32) -> bool {
// SAFETY: forwarded to the engine's publish contract.
unsafe { oakengine_ipc_framepool_publish(self.handle, index) == 1 }
}
/// Take the next published slot. Returns false if nothing is ready.
///
/// # Safety
/// The pool must be a valid view of a live segment.
pub unsafe fn consume(&self, index: &mut u32) -> bool {
// SAFETY: forwarded to the engine's consume contract.
unsafe { oakengine_ipc_framepool_consume(self.handle, index) == 1 }
}
/// Return a consumed slot to the free pool for reuse. Must follow a
/// successful [`Self::consume`] of `index`. Returns false if the free
/// ring is full.
///
/// # Safety
/// `index` must be a slot previously consumed and not yet re-acquired.
pub unsafe fn release(&self, index: u32) -> bool {
// SAFETY: forwarded to the engine's release contract.
unsafe { oakengine_ipc_framepool_release(self.handle, index) == 1 }
}
}
impl Clone for FrameSlotPool {
fn clone(&self) -> FrameSlotPool {
// SAFETY: the engine's framepool_copy yields NULL for a NULL
// handle.
FrameSlotPool {
handle: unsafe { oakengine_ipc_framepool_copy(self.handle) },
}
}
}
impl Drop for FrameSlotPool {
fn drop(&mut self) {
// SAFETY: the engine's framepool_free is a NULL no-op; the handle
// is not used after this.
unsafe { oakengine_ipc_framepool_free(self.handle) };
}
}
crates/oak-worker/src/ipc.rs
+1477
View File
File diff suppressed because it is too large Load Diff
crates/oak-worker/src/main.rs
+70 -34
View File
@@ -16,31 +16,27 @@
//! oak-worker: headless render worker process (Rust).
//!
//! A thin C-ABI shell over the `liboakengine` facade dylib, mirroring
//! `worker/workermain.cpp`: all runtime logic — render backend selection
//! (dynamic -> OpenGL fallback through the oakrender module), the startup
//! handshake and the NDJSON control loop — lives in the engine
//! (`oakengine_worker_main`, the port of `engine/src/capi/worker.cpp`,
//! contract in `engine/include/oakengine/worker.h`). This crate keeps only
//! the argv forwarding, the `#[link]` declarations ([`engine_ipc`]) and
//! the in-process session mirror ([`session`], [`transport`]) that its
//! unit tests exercise against the engine's real shared-memory transport.
//! The whole runtime lives in this binary (M14 R2): render backend
//! selection (dynamic -> OpenGL fallback through the oakrender crate's
//! direct Rust API), the startup handshake and the NDJSON control loop
//! ([`worker`], the port of `engine/src/capi/worker.cpp`), plus the
//! shared-memory frame-slot transport ([`ipc`], the port of
//! `engine/render/ipc/`). No `liboakengine` dylib is linked; the facade
//! keeps its own copies of both modules for the frozen
//! `oakengine_worker_*` / `oakengine_ipc_*` C ABI (external consumers).
//!
//! See README.md for the full status.
#![deny(unsafe_op_in_unsafe_fn)]
#![warn(missing_docs)]
mod engine_ipc;
mod session;
mod transport;
mod ipc;
mod worker;
use std::ffi::{c_char, c_int, CString};
use std::process::exit;
/// Protocol version announced in the startup handshake
/// (`k_protocol_version` in worker.cpp). Mirrors the engine worker
/// module's `PROTOCOL_VERSION`.
/// (`k_protocol_version` in worker.cpp).
pub const PROTOCOL_VERSION: i32 = 1;
/// Log a worker-side message to stderr, mirroring worker.cpp `log_error()`
@@ -49,26 +45,26 @@ pub fn log_error(message: &str) {
eprintln!("worker: {message}");
}
/// Scan argv for `--backend <name>` (worker.cpp `oakengine_worker_main`).
/// The default is `"opengl"`; the value is lowercased; the last flag wins.
fn parse_backend(args: &[String]) -> String {
let mut backend = "opengl".to_string();
let mut i = 1usize;
while i < args.len() {
if args[i] == "--backend" && i + 1 < args.len() {
backend = args[i + 1].to_ascii_lowercase();
i += 2;
} else {
i += 1;
}
}
backend
}
fn main() {
// Forward argv verbatim: the engine's oakengine_worker_main() scans
// for `--backend` itself (workermain.cpp does the same).
let args: Vec<String> = std::env::args_os()
.map(|a| a.to_string_lossy().into_owned())
.collect();
let cstrings: Vec<CString> = args
.iter()
.map(|a| CString::new(a.as_str()).unwrap_or_else(|_| CString::new("").unwrap()))
.collect();
let mut argv: Vec<*mut c_char> = cstrings
.iter()
.map(|c| c.as_ptr() as *mut c_char)
.collect();
// SAFETY: `argv` is an array of `argc` NUL-terminated C strings, kept
// alive for the whole call; the engine only reads them.
let code = unsafe {
engine_ipc::oakengine_worker_main(argv.len() as c_int, argv.as_mut_ptr())
};
exit(code);
let args: Vec<String> = std::env::args().collect();
let backend = parse_backend(&args);
exit(worker::worker_main(&backend));
}
#[cfg(test)]
@@ -79,4 +75,44 @@ mod tests {
fn protocol_version_is_one() {
assert_eq!(PROTOCOL_VERSION, 1);
}
#[test]
fn backend_parsing_matches_engine_main() {
// Default is opengl.
assert_eq!(parse_backend(&["oak-worker".to_string()]), "opengl");
// Value lowercased.
assert_eq!(
parse_backend(&[
"oak-worker".to_string(),
"--backend".to_string(),
"Vulkan".to_string()
]),
"vulkan"
);
// "none" skips renderer creation.
assert_eq!(
parse_backend(&[
"oak-worker".to_string(),
"--backend".to_string(),
"none".to_string()
]),
"none"
);
// Last flag wins.
assert_eq!(
parse_backend(&[
"oak-worker".to_string(),
"--backend".to_string(),
"vulkan".to_string(),
"--backend".to_string(),
"opengl".to_string()
]),
"opengl"
);
// Missing value leaves the default.
assert_eq!(
parse_backend(&["oak-worker".to_string(), "--backend".to_string()]),
"opengl"
);
}
}
crates/oak-worker/src/session.rs
-527
View File
@@ -1,527 +0,0 @@
// 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 worker-side session state machine — the in-process mirror of
//! `OakWorkerSession` in `engine/src/capi/worker.cpp` (whose production
//! Rust port lives in the engine behind the `oakengine_worker_session_*`
//! C ABI).
//!
//! The session holds the attached shared-memory frame-slot pools
//! ([`crate::transport::AttachedPools`]) and the shutdown flag, and
//! answers one NDJSON control message at a time. Renderer creation and the
//! main loop are the engine's job (see `crate::main`); this module keeps
//! the message handling testable in-process against the engine's real
//! shared-memory transport ([`crate::engine_ipc`], consumed through the
//! `oakengine_ipc_*` C ABI). Where the C++ session has machinery the Rust
//! mirror lacks, the handler reproduces the *validation* faithfully and
//! then reports the documented stub ([`crate::transport`]) — it never
//! fakes a result.
//!
//! The production session is the engine's `OakWorkerSession`; this mirror
//! exists to keep the message handling testable in-process without a GPU
//! backend, so its public surface is exercised by the unit tests only.
#![allow(dead_code)]
use serde_json::Value;
use crate::engine_ipc::{self as ipc, HandshakeMsg, LoadGraphMsg, RenderFrameMsg};
use crate::transport::{self, AttachedPools};
/// Worker-side session: attached frame-slot pools + message-handling state.
pub struct WorkerSession {
/// Attached shared-memory frame-slot pools (output + optional input).
pools: Option<AttachedPools>,
shutdown_requested: bool,
}
impl WorkerSession {
/// A fresh session with no attached pools.
pub fn new() -> WorkerSession {
WorkerSession {
pools: None,
shutdown_requested: false,
}
}
/// 1 once the handshake has attached the shared-memory frame-slot
/// pools.
pub fn has_pools(&self) -> bool {
self.pools.is_some()
}
/// 1 once a shutdown control message has been received.
pub fn shutdown_requested(&self) -> bool {
self.shutdown_requested
}
/// The attached output pool (the worker->main frame-slot pool).
pub fn output_pool(&self) -> Option<&crate::engine_ipc::FrameSlotPool> {
self.pools.as_ref().map(|p| &p.output_pool)
}
/// The startup handshake the worker sends to its parent
/// (`worker.cpp startup_handshake()`): protocol version 1 and empty
/// shared-memory geometry — the parent creates the segments and
/// announces their geometry in its handshake reply.
pub fn startup_handshake(&self) -> Value {
HandshakeMsg {
protocol_version: crate::PROTOCOL_VERSION,
shm_key: String::new(),
input_shm_key: String::new(),
input_slots: 0,
output_slots: 0,
slot_data_bytes: 0,
input_slot_data_bytes: 0,
}
.to_json()
}
/// Handle one complete NDJSON control line and produce the response, if
/// any — the port of worker.cpp `handle()`. A malformed line yields an
/// error response (the loop continues), never a failure.
pub fn handle_line(&mut self, line: &str) -> Option<Value> {
let msg: Value = match serde_json::from_str::<Value>(line) {
Ok(v) if v.is_object() => v,
_ => return Some(ipc::error_message("malformed control message", None)),
};
let typ = msg.get("type").and_then(Value::as_str).unwrap_or("");
match typ {
ipc::TYPE_HANDSHAKE => self.handle_handshake(&msg),
ipc::TYPE_LOAD_GRAPH => self.handle_load_graph(&msg),
ipc::TYPE_RENDER_FRAME => self.handle_render_frame(&msg),
// cancel: the C++ worker does synchronous single-frame work
// (nothing in flight), so a cancel produces no response.
ipc::TYPE_CANCEL => None,
ipc::TYPE_SHUTDOWN => {
self.shutdown_requested = true;
None
}
other => Some(ipc::error_message(
&format!("unknown message type: {other}"),
None,
)),
}
}
/// `handshake`: validate and attach the shared-memory frame-slot pools.
/// Validation mirrors worker.cpp `attach_output_pool()`; the attachment
/// itself goes through the real [`crate::transport`].
fn handle_handshake(&mut self, msg: &Value) -> Option<Value> {
let hs: HandshakeMsg = match serde_json::from_value(msg.clone()) {
Ok(hs) => hs,
Err(_) => return Some(ipc::error_message("invalid handshake message", None)),
};
if hs.protocol_version != crate::PROTOCOL_VERSION {
return Some(ipc::error_message(
&format!("unsupported protocol version {}", hs.protocol_version),
None,
));
}
if hs.shm_key.is_empty() || hs.output_slots <= 0 || hs.slot_data_bytes <= 0 {
return Some(ipc::error_message(
"handshake missing output shared-memory geometry",
None,
));
}
if hs.input_slots > 0 && (hs.input_shm_key.is_empty() || hs.input_slot_data_bytes <= 0) {
return Some(ipc::error_message(
"handshake missing input shared-memory geometry",
None,
));
}
match transport::attach_pools(&hs) {
Ok(pools) => {
self.pools = Some(pools);
None
}
Err(msg) => Some(ipc::error_message(&msg, None)),
}
}
/// `load_graph`: the file checks are real (mirror worker.cpp
/// `load_graph()`); the deserialization is the documented stub.
fn handle_load_graph(&mut self, msg: &Value) -> Option<Value> {
let load: LoadGraphMsg = match serde_json::from_value(msg.clone()) {
Ok(l) => l,
Err(_) => return Some(ipc::error_message("invalid load_graph message", None)),
};
match std::fs::metadata(&load.path) {
Err(_) => Some(ipc::error_message(
&format!("graph file does not exist: {}", load.path),
None,
)),
Ok(md) if md.len() == 0 => Some(ipc::error_message(
&format!("graph file is empty: {}", load.path),
None,
)),
Ok(md) => {
crate::log_error(&format!(
"LoadGraph: loading {} ({} bytes)",
load.path,
md.len()
));
Some(ipc::error_message(transport::GRAPH_STUB, None))
}
}
}
/// `render_frame`: the graph/render pipeline has no Rust backing, so a
/// render request is answered with a clear error carrying the ticket —
/// the same `error_message()` shape the C++ worker uses for its own
/// failures.
fn handle_render_frame(&mut self, msg: &Value) -> Option<Value> {
let render: RenderFrameMsg = match serde_json::from_value(msg.clone()) {
Ok(r) => r,
Err(_) => return Some(ipc::error_message("invalid render_frame message", None)),
};
Some(ipc::error_message(
transport::RENDER_STUB,
Some(render.ticket),
))
}
}
impl Default for WorkerSession {
fn default() -> Self {
WorkerSession::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::engine_ipc::{FrameSlotPool, SharedMemoryRegion, ShmMode};
use serde_json::json;
/// A unique, temporary POSIX segment key for a test.
fn test_key(name: &str) -> String {
static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
SharedMemoryRegion::make_key(i64::from(std::process::id()), (n & 0x7FFF) as i32)
+ &format!("-t-{name}")
}
/// The "parent" side of a handshake: create an output segment holding a
/// pool, optionally an input segment, and return the handshake message
/// plus the owner regions (kept alive by the caller).
fn parent_side(
slots: i32,
slot_bytes: i64,
input: bool,
) -> (Value, SharedMemoryRegion, Option<SharedMemoryRegion>) {
let out_key = test_key("out");
let out_bytes = FrameSlotPool::bytes_needed(slots as u32, slot_bytes as usize);
let mut out_region = SharedMemoryRegion::new();
assert!(
out_region.open(&out_key, out_bytes, ShmMode::Create),
"{}",
out_region.error()
);
// SAFETY: live mapping sized by bytes_needed.
let _ =
unsafe { FrameSlotPool::create(out_region.data(), slots as u32, slot_bytes as usize) };
let (in_key, in_bytes, in_region) = if input {
let in_key = test_key("in");
let in_bytes = FrameSlotPool::bytes_needed(slots as u32, slot_bytes as usize);
let mut in_region = SharedMemoryRegion::new();
assert!(in_region.open(&in_key, in_bytes, ShmMode::Create));
// SAFETY: live mapping.
let _ = unsafe {
FrameSlotPool::create(in_region.data(), slots as u32, slot_bytes as usize)
};
(Some(in_key), Some(in_bytes), Some(in_region))
} else {
(None, None, None)
};
let hs = json!({
"type": "handshake",
"protocol_version": crate::PROTOCOL_VERSION,
"shm_key": out_key,
"input_shm_key": in_key.unwrap_or_default(),
"input_slots": if input { slots } else { 0 },
"output_slots": slots,
"slot_data_bytes": slot_bytes,
"input_slot_data_bytes": in_bytes.unwrap_or(0),
});
(hs, out_region, in_region)
}
#[test]
fn session_starts_without_pools() {
let s = WorkerSession::new();
assert!(!s.has_pools());
assert!(!s.shutdown_requested());
}
#[test]
fn startup_handshake_is_protocol_version_1_with_empty_geometry() {
let s = WorkerSession::new();
let hs = s.startup_handshake();
assert_eq!(
hs,
json!({
"type": "handshake",
"protocol_version": 1,
"shm_key": "",
"input_shm_key": "",
"input_slots": 0,
"output_slots": 0,
"slot_data_bytes": 0,
"input_slot_data_bytes": 0,
})
);
}
#[test]
fn malformed_line_yields_error_response() {
let mut s = WorkerSession::new();
let resp = s.handle_line("this is not json").unwrap();
assert_eq!(resp["type"], "error");
assert_eq!(resp["message"], "malformed control message");
}
#[test]
fn non_object_json_yields_error_response() {
let mut s = WorkerSession::new();
let resp = s.handle_line("[1,2,3]").unwrap();
assert_eq!(resp["message"], "malformed control message");
}
#[test]
fn unknown_message_type_yields_error_response() {
let mut s = WorkerSession::new();
let resp = s.handle_line(r#"{"type":"frobnicate"}"#).unwrap();
assert_eq!(resp["type"], "error");
assert_eq!(resp["message"], "unknown message type: frobnicate");
}
#[test]
fn missing_type_field_yields_unknown_error() {
let mut s = WorkerSession::new();
let resp = s.handle_line(r#"{"hello":1}"#).unwrap();
assert_eq!(resp["message"], "unknown message type: ");
}
#[test]
fn cancel_produces_no_response() {
let mut s = WorkerSession::new();
assert!(s.handle_line(r#"{"type":"cancel","ticket":5}"#).is_none());
assert!(!s.shutdown_requested());
}
#[test]
fn shutdown_sets_flag_and_has_no_response() {
let mut s = WorkerSession::new();
assert!(s.handle_line(r#"{"type":"shutdown"}"#).is_none());
assert!(s.shutdown_requested());
}
#[test]
fn handshake_wrong_protocol_version() {
let mut s = WorkerSession::new();
let resp = s
.handle_line(r#"{"type":"handshake","protocol_version":99,"shm_key":"k","output_slots":1,"slot_data_bytes":16}"#)
.unwrap();
assert_eq!(resp["message"], "unsupported protocol version 99");
}
#[test]
fn handshake_missing_geometry() {
let mut s = WorkerSession::new();
let resp = s
.handle_line(r#"{"type":"handshake","protocol_version":1}"#)
.unwrap();
assert_eq!(
resp["message"],
"handshake missing output shared-memory geometry"
);
}
#[test]
fn handshake_missing_input_geometry_is_an_error() {
let mut s = WorkerSession::new();
let (mut hs, _out, _in) = parent_side(2, 256, false);
// Ask for input slots without announcing their geometry.
hs["input_slots"] = json!(2);
let resp = s.handle_line(&hs.to_string()).unwrap();
assert_eq!(
resp["message"],
"handshake missing input shared-memory geometry"
);
}
#[test]
fn handshake_attaches_real_output_pool() {
let mut s = WorkerSession::new();
let (hs, out_region, _in) = parent_side(4, 4096, false);
let resp = s.handle_line(&hs.to_string());
assert!(resp.is_none(), "unexpected error: {resp:?}");
assert!(s.has_pools());
let out_pool = s.output_pool().unwrap();
assert_eq!(out_pool.slot_count(), 4);
assert_eq!(out_pool.slot_data_bytes(), 4096);
// The pool is real shared state: the parent's publish lands in the
// worker's ready ring.
// SAFETY: `out_region` is a live mapping of the pool the session
// attached to.
let parent_pool = unsafe { FrameSlotPool::attach(out_region.data()) };
let mut slot = 0;
assert!(unsafe { parent_pool.acquire(&mut slot) });
assert_eq!(slot, 0);
// SAFETY: acquired slot.
let meta = unsafe { &mut *parent_pool.meta(slot) };
meta.id = 7;
assert!(unsafe { parent_pool.publish(slot) });
let mut consumed = 0;
assert!(unsafe { out_pool.consume(&mut consumed) });
assert_eq!(consumed, 0);
// SAFETY: consumed slot.
assert_eq!(unsafe { (*out_pool.meta_const(consumed)).id }, 7);
unsafe { out_pool.release(consumed) };
}
#[test]
fn handshake_attaches_input_pool_too() {
let mut s = WorkerSession::new();
let (hs, _out, _in) = parent_side(2, 256, true);
let resp = s.handle_line(&hs.to_string());
assert!(resp.is_none(), "unexpected error: {resp:?}");
assert!(s.has_pools());
let pools = s.pools.as_ref().unwrap();
assert!(pools.input_pool.is_some());
let in_pool = pools.input_pool.as_ref().unwrap();
assert_eq!(in_pool.slot_count(), 2);
assert_eq!(in_pool.slot_data_bytes(), 256);
}
#[test]
fn handshake_attach_failure_reports_error() {
let mut s = WorkerSession::new();
// A key that was never created.
let resp = s
.handle_line(
&json!({
"type": "handshake",
"protocol_version": 1,
"shm_key": format!("olive-rw-{}-missing", std::process::id()),
"output_slots": 4,
"slot_data_bytes": 4096,
})
.to_string(),
)
.unwrap();
assert_eq!(resp["type"], "error");
assert!(resp["message"]
.as_str()
.unwrap()
.starts_with("failed to attach shared memory: "));
assert!(!s.has_pools());
}
#[test]
fn handshake_rejects_non_pool_segment() {
let mut s = WorkerSession::new();
// A real segment of the right size that does not contain a pool
// (zeroed memory -> wrong magic).
let key = test_key("nopool");
let bytes = FrameSlotPool::bytes_needed(4, 4096);
let mut region = SharedMemoryRegion::new();
assert!(region.open(&key, bytes, ShmMode::Create));
let resp = s
.handle_line(
&json!({
"type": "handshake",
"protocol_version": 1,
"shm_key": key,
"output_slots": 4,
"slot_data_bytes": 4096,
})
.to_string(),
)
.unwrap();
assert_eq!(
resp["message"],
"shared memory does not contain a frame slot pool"
);
assert!(!s.has_pools());
}
#[test]
fn handshake_bad_json_shape_is_invalid_handshake() {
let mut s = WorkerSession::new();
let resp = s
.handle_line(r#"{"type":"handshake","protocol_version":"x"}"#)
.unwrap();
assert_eq!(resp["message"], "invalid handshake message");
}
#[test]
fn load_graph_file_checks_are_real_then_stub() {
let mut s = WorkerSession::new();
let missing = "/definitely/not/a/real/graph.ove";
let resp = s
.handle_line(&json!({ "type": "load_graph", "path": missing }).to_string())
.unwrap();
assert_eq!(
resp["message"],
format!("graph file does not exist: {missing}")
);
let empty = std::env::temp_dir().join("oak_worker_test_empty_graph.ove");
std::fs::write(&empty, b"").unwrap();
let resp = s
.handle_line(
&json!({ "type": "load_graph", "path": empty.display().to_string() }).to_string(),
)
.unwrap();
assert_eq!(
resp["message"],
format!("graph file is empty: {}", empty.display())
);
let _ = std::fs::remove_file(&empty);
let real = std::env::temp_dir().join("oak_worker_test_graph.ove");
std::fs::write(&real, b"<root/>").unwrap();
let resp = s
.handle_line(
&json!({ "type": "load_graph", "path": real.display().to_string() }).to_string(),
)
.unwrap();
assert!(resp["message"]
.as_str()
.unwrap()
.contains("node-graph deserialization is not yet available"));
let _ = std::fs::remove_file(&real);
}
#[test]
fn render_frame_reports_stub_with_ticket() {
let mut s = WorkerSession::new();
let resp = s
.handle_line(r#"{"type":"render_frame","ticket":123,"node":"abc"}"#)
.unwrap();
assert_eq!(resp["type"], "error");
assert_eq!(resp["ticket"], 123);
assert!(resp["message"]
.as_str()
.unwrap()
.contains("frame rendering is not yet available"));
}
}
crates/oak-worker/src/transport.rs
-124
View File
@@ -1,124 +0,0 @@
// 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/>.
//! Shared-memory frame-slot transport.
//!
//! The C++ worker exchanges bulk pixel data with the editor through named
//! shared-memory segments holding an `olive::ipc::FrameSlotPool` — a fixed
//! pool of frame slots whose free/ready queues are synchronized by the
//! lock-free single-producer/single-consumer `SpscRingBuffer`. That
//! machinery is implemented in the engine facade (behind the
//! `oakengine_ipc_*` C ABI in `engine/include/oakengine/ipc.h`) — this
//! module is the worker-side transport over it, attached through
//! [`crate::engine_ipc`]'s C-ABI wrappers.
//!
//! [`attach_pools`] mirrors the C++ `attach_output_pool()`: attach the
//! output segment in [`ShmMode::Attach`], map the [`FrameSlotPool`] it
//! contains (rejecting segments without the pool magic), and attach the
//! input pool when the handshake announces one. The returned
//! [`AttachedPools`] is owned by the session and dropped (unmapped) with
//! it.
//!
//! The node-graph and render-pipeline stubs below carry the same rationale
//! as before: `oaknode` is a `todo!()` skeleton and the oakrender crate
//! does not yet evaluate an arbitrary loaded graph to a frame.
//!
//! The production attach path is the engine's `OakWorkerSession`
//! `handle_handshake`; this module is the transport surface the in-process
//! session mirror ([`crate::session`]) uses, exercised by the unit tests.
#![allow(dead_code)]
use crate::engine_ipc::{FrameSlotPool, SharedMemoryRegion, ShmMode};
use crate::engine_ipc::HandshakeMsg;
/// Why `load_graph` answers "not yet available" (after the real file checks).
pub const GRAPH_STUB: &str = "load_graph: node-graph deserialization is not yet available in the \
Rust worker (the oaknode crate is a todo!() skeleton; see worker/rust/README.md)";
/// Why `render_frame` answers "not yet available".
pub const RENDER_STUB: &str = "render_frame: frame rendering is not yet available in the Rust \
worker (no node-graph or render-pipeline backing; the shm frame-slot transport is \
attached but there is no graph to render; see worker/rust/README.md)";
/// The shared-memory frame-slot pools attached by a successful handshake,
/// kept alive for the session's lifetime.
pub struct AttachedPools {
/// Worker->main output segment (unmapped on drop).
pub output_region: SharedMemoryRegion,
/// Output frame-slot pool view.
pub output_pool: FrameSlotPool,
/// Main->worker input segment, when the handshake announced one.
pub input_region: Option<SharedMemoryRegion>,
/// Input frame-slot pool view.
pub input_pool: Option<FrameSlotPool>,
}
/// Attach the handshake's shared-memory frame-slot pools — the real port of
/// worker.cpp `attach_output_pool()`.
///
/// The output segment must exist and contain a valid [`FrameSlotPool`]
/// (magic check); the input pool is attached when `input_slots > 0`.
/// Returns `Err(message)` describing the failure, matching the C++ error
/// strings.
pub fn attach_pools(hs: &HandshakeMsg) -> Result<AttachedPools, String> {
// Attach the worker->main output pool.
let bytes = FrameSlotPool::bytes_needed(hs.output_slots as u32, hs.slot_data_bytes as usize);
let mut output_region = SharedMemoryRegion::new();
if !output_region.open(&hs.shm_key, bytes, ShmMode::Attach) {
return Err(format!(
"failed to attach shared memory: {}",
output_region.error()
));
}
// SAFETY: `output_region` is a live mapping of at least `bytes` bytes
// (checked inside `open`).
let output_pool = unsafe { FrameSlotPool::attach(output_region.data()) };
if !output_pool.is_valid() {
return Err("shared memory does not contain a frame slot pool".to_string());
}
// Attach the main->worker input pool when the handshake announced one.
let mut input_region = None;
let mut input_pool = None;
if hs.input_slots > 0 {
let input_bytes =
FrameSlotPool::bytes_needed(hs.input_slots as u32, hs.input_slot_data_bytes as usize);
let mut region = SharedMemoryRegion::new();
if !region.open(&hs.input_shm_key, input_bytes, ShmMode::Attach) {
return Err(format!(
"failed to attach input shared memory: {}",
region.error()
));
}
// SAFETY: `region` is a live mapping of at least `input_bytes`
// bytes (checked inside `open`).
let pool = unsafe { FrameSlotPool::attach(region.data()) };
if !pool.is_valid() {
return Err("input shared memory does not contain a frame slot pool".to_string());
}
input_region = Some(region);
input_pool = Some(pool);
}
Ok(AttachedPools {
output_region,
output_pool,
input_region,
input_pool,
})
}
crates/oak-worker/src/worker.rs
+812
View File
@@ -0,0 +1,812 @@
// 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 render worker runtime — the Rust port of
//! `engine/src/capi/worker.cpp`, owned by the oak-worker binary since
//! M14 R2 (the facade keeps its own copy for the frozen
//! `oakengine_worker_*` C ABI).
//!
//! - **Backend selection.** [`Renderer::create`] initializes the render
//! backend through the oakrender crate's direct Rust API
//! ([`oakrender::backend::DisplayRenderer`]), falling back to the
//! direct OpenGL renderer exactly like the C++ `create_renderer()`
//! chain.
//! - **The session.** [`WorkerSession`] holds the renderer, the
//! shared-memory frame-slot pools ([`crate::ipc::FrameSlotPool`]) and
//! the shutdown flag, and answers one NDJSON control message at a time.
//! - **The main loop.** [`worker_main`] creates the session, loads the
//! runtime config, writes the startup handshake, and serves the
//! stdin/stdout NDJSON loop until a `shutdown` message or EOF.
//!
//! The control-plane protocol is the same NDJSON the C++ worker speaks
//! (`engine/render/ipc/ipcmessage.cpp`): one compact JSON object per line,
//! `"type"`-dispatched ([`crate::ipc`]), with `handshake` carrying the
//! shared-memory geometry the worker attaches to via the real
//! [`crate::ipc`] transport. `load_graph`/`render_frame` reproduce the
//! C++ validation and then answer with the documented "not yet available"
//! errors (the oaknode graph crate is still a skeleton).
use std::io::{self, BufRead, Write};
use serde_json::Value;
use oakrender::backend::{BackendKind, DisplayRenderer};
use crate::ipc::{
error_message, write_message, FrameSlotPool, HandshakeMsg, LoadGraphMsg, RenderFrameMsg,
SharedMemoryRegion, ShmMode, TYPE_CANCEL, TYPE_HANDSHAKE, TYPE_LOAD_GRAPH, TYPE_RENDER_FRAME,
TYPE_SHUTDOWN,
};
use crate::{log_error, PROTOCOL_VERSION};
/// Why `load_graph` answers "not yet available" (after the real file checks).
const GRAPH_STUB: &str = "load_graph: node-graph deserialization is not yet available in the \
Rust worker (the oaknode crate is a todo!() skeleton; see worker/rust/README.md)";
/// Why `render_frame` answers "not yet available".
const RENDER_STUB: &str = "render_frame: frame rendering is not yet available in the Rust \
worker (no node-graph or render-pipeline backing; the shm frame-slot transport is \
attached but there is no graph to render; see worker/rust/README.md)";
// ---------------------------------------------------------------------------
// Renderer (backend selection)
// ---------------------------------------------------------------------------
/// Whether `backend` requests no renderer (worker.cpp
/// `backend_requests_no_renderer()`: NULL, "" and "none").
pub fn is_no_backend(backend: &str) -> bool {
backend.is_empty() || backend.eq_ignore_ascii_case("none")
}
/// A live, initialized oakrender display renderer (destroyed on drop).
pub struct Renderer {
/// The oakrender crate's value-typed display renderer (single-lib
/// unification; the CHandle-based C ABI is deleted).
inner: DisplayRenderer,
}
impl Renderer {
/// Create and initialize a renderer through the oakrender crate's
/// direct Rust API, trying the named dynamic backend first and falling
/// back to the direct OpenGL renderer — the exact fallback chain of
/// worker.cpp `create_renderer()`.
pub fn create(backend: &str) -> Result<Renderer, String> {
match Self::create_dynamic(backend) {
Ok(r) => Ok(r),
Err(first) => {
log_error(&format!(
"failed to initialize dynamic {backend} backend: {first}; falling back to direct OpenGL renderer"
));
Self::create_opengl().map_err(|second| {
format!("{first}; direct OpenGL fallback also failed: {second}")
})
}
}
}
/// Try the named dynamic backend (`DisplayRenderer::new` +
/// `init`, the single-lib equivalent of
/// `oakrender_display_renderer_create_dynamic` + `_init`).
fn create_dynamic(backend: &str) -> Result<Renderer, String> {
let renderer = DisplayRenderer::new(BackendKind::from_config_string(backend));
Self::init_inner(renderer, &format!("dynamic {backend}"))
}
/// Fall back to the direct OpenGL renderer.
fn create_opengl() -> Result<Renderer, String> {
let renderer = DisplayRenderer::new(BackendKind::Gl);
Self::init_inner(renderer, "direct OpenGL")
}
/// Initialize a freshly created renderer.
fn init_inner(mut renderer: DisplayRenderer, what: &str) -> Result<Renderer, String> {
// NULL gl_context makes the backend use its default device/context
// path.
if let Err(e) = renderer.init(std::ptr::null_mut()) {
return Err(format!("failed to initialize {what} renderer ({e})"));
}
Ok(Renderer { inner: renderer })
}
/// 1 when the renderer is OpenGL-based (the C++ worker uses the GL
/// context to announce the negotiated GL version in the handshake).
///
/// Not called yet: the oakrender module exposes no GL context
/// version, so the startup handshake omits `gl_major`/`gl_minor`.
#[allow(dead_code)]
pub fn is_open_gl(&self) -> bool {
self.inner.is_open_gl()
}
}
// ---------------------------------------------------------------------------
// WorkerSession
// ---------------------------------------------------------------------------
/// The worker-side session state machine — the Rust mirror of
/// `OakWorkerSession` in worker.cpp. Holds the renderer, the attached
/// shared-memory frame-slot pools and the shutdown flag, and answers one
/// NDJSON control message at a time.
pub struct WorkerSession {
renderer: Option<Renderer>,
shutdown_requested: bool,
runtime_initialized: bool,
output_region: Option<SharedMemoryRegion>,
output_pool: Option<FrameSlotPool>,
input_region: Option<SharedMemoryRegion>,
input_pool: Option<FrameSlotPool>,
}
impl WorkerSession {
/// Create a session for `backend`, mirroring
/// `oakengine_worker_session_create()`: "none"/"" skips renderer
/// creation, anything else initializes the render backend through the
/// oakrender crate's direct Rust API (dynamic -> OpenGL fallback).
pub fn create(backend: &str) -> Result<WorkerSession, String> {
let renderer = if is_no_backend(backend) {
None
} else {
Some(Renderer::create(backend)?)
};
Ok(WorkerSession {
renderer,
shutdown_requested: false,
runtime_initialized: false,
output_region: None,
output_pool: None,
input_region: None,
input_pool: None,
})
}
/// 1 when the session holds a successfully initialized render backend.
pub fn has_renderer(&self) -> bool {
self.renderer.is_some()
}
/// 1 once a shutdown control message has been received.
pub fn shutdown_requested(&self) -> bool {
self.shutdown_requested
}
/// Load the runtime services the session depends on — the Rust analog
/// of the C++ `initialize_runtime()`. Of the C++ list (config, node
/// factory, color manager, frame/disk managers, project serializer)
/// only the color-manager default config has a Rust backing linked into
/// the worker binary; the rest are logged and skipped. Always returns
/// true (the C++ returns true unconditionally).
pub fn initialize_runtime(&mut self) -> bool {
if self.runtime_initialized {
return true;
}
log_error("runtime: loading color-manager default config");
if let Err(e) = oakrender::color::set_up_default_config() {
log_error(&format!(
"runtime: color-manager default config failed ({e}); continuing"
));
}
log_error(
"runtime: config / node factory / frame manager / disk manager / project \
serializer have no Rust backing in the worker binary; skipped",
);
self.runtime_initialized = true;
true
}
/// The startup handshake the worker sends to its parent
/// (`worker.cpp startup_handshake()`): protocol version 1 and empty
/// shared-memory geometry — the parent creates the segments and
/// announces their geometry in its handshake reply.
///
/// Deviation from the C++: `gl_major`/`gl_minor` are omitted because
/// the oakrender module exposes no GL context version.
pub fn startup_handshake(&self) -> Value {
HandshakeMsg {
protocol_version: PROTOCOL_VERSION,
shm_key: String::new(),
input_shm_key: String::new(),
input_slots: 0,
output_slots: 0,
slot_data_bytes: 0,
input_slot_data_bytes: 0,
}
.to_json()
}
/// Handle one complete NDJSON control line and produce the response, if
/// any — the port of worker.cpp `handle()`. A malformed line yields an
/// error response (the loop continues), never a failure.
pub fn handle_line(&mut self, line: &str) -> Option<Value> {
let msg: Value = match serde_json::from_str::<Value>(line) {
Ok(v) if v.is_object() => v,
_ => return Some(error_message("malformed control message", None)),
};
let typ = msg.get("type").and_then(Value::as_str).unwrap_or("");
match typ {
TYPE_HANDSHAKE => self.handle_handshake(&msg),
TYPE_LOAD_GRAPH => self.handle_load_graph(&msg),
TYPE_RENDER_FRAME => self.handle_render_frame(&msg),
// cancel: the worker does synchronous single-frame work
// (nothing in flight), so a cancel produces no response.
TYPE_CANCEL => None,
TYPE_SHUTDOWN => {
self.shutdown_requested = true;
None
}
other => Some(error_message(
&format!("unknown message type: {other}"),
None,
)),
}
}
/// `handshake`: validate and attach the shared-memory frame-slot pools
/// — the real port of worker.cpp `attach_output_pool()`.
fn handle_handshake(&mut self, msg: &Value) -> Option<Value> {
let hs: HandshakeMsg = match serde_json::from_value(msg.clone()) {
Ok(hs) => hs,
Err(_) => return Some(error_message("invalid handshake message", None)),
};
if hs.protocol_version != PROTOCOL_VERSION {
return Some(error_message(
&format!("unsupported protocol version {}", hs.protocol_version),
None,
));
}
if hs.shm_key.is_empty() || hs.output_slots <= 0 || hs.slot_data_bytes <= 0 {
return Some(error_message(
"handshake missing output shared-memory geometry",
None,
));
}
// A re-handshake replaces the pools (worker.cpp resets the input
// pool before attaching the output).
self.input_pool = None;
self.input_region = None;
self.output_pool = None;
self.output_region = None;
let bytes =
FrameSlotPool::bytes_needed(hs.output_slots as u32, hs.slot_data_bytes as usize);
let mut output_region = SharedMemoryRegion::new();
if !output_region.open(&hs.shm_key, bytes, ShmMode::Attach) {
return Some(error_message(
&format!("failed to attach shared memory: {}", output_region.error()),
None,
));
}
// SAFETY: `output_region` is a live mapping of at least `bytes`
// bytes (checked above).
let output_pool = unsafe { FrameSlotPool::attach(output_region.data()) };
if !output_pool.is_valid() {
return Some(error_message(
"shared memory does not contain a frame slot pool",
None,
));
}
self.output_region = Some(output_region);
self.output_pool = Some(output_pool);
if hs.input_slots > 0 {
if hs.input_shm_key.is_empty() || hs.input_slot_data_bytes <= 0 {
return Some(error_message(
"handshake missing input shared-memory geometry",
None,
));
}
let input_bytes = FrameSlotPool::bytes_needed(
hs.input_slots as u32,
hs.input_slot_data_bytes as usize,
);
let mut input_region = SharedMemoryRegion::new();
if !input_region.open(&hs.input_shm_key, input_bytes, ShmMode::Attach) {
return Some(error_message(
&format!(
"failed to attach input shared memory: {}",
input_region.error()
),
None,
));
}
// SAFETY: `input_region` is a live mapping of at least
// `input_bytes` bytes (checked above).
let input_pool = unsafe { FrameSlotPool::attach(input_region.data()) };
if !input_pool.is_valid() {
return Some(error_message(
"input shared memory does not contain a frame slot pool",
None,
));
}
self.input_region = Some(input_region);
self.input_pool = Some(input_pool);
}
// Success: no response (worker.cpp leaves `response` untouched).
None
}
/// `load_graph`: the file checks are real (mirror worker.cpp
/// `load_graph()`); the deserialization is the documented stub.
fn handle_load_graph(&mut self, msg: &Value) -> Option<Value> {
let load: LoadGraphMsg = match serde_json::from_value(msg.clone()) {
Ok(l) => l,
Err(_) => return Some(error_message("invalid load_graph message", None)),
};
match std::fs::metadata(&load.path) {
Err(_) => Some(error_message(
&format!("graph file does not exist: {}", load.path),
None,
)),
Ok(md) if md.len() == 0 => Some(error_message(
&format!("graph file is empty: {}", load.path),
None,
)),
Ok(md) => {
log_error(&format!(
"LoadGraph: loading {} ({} bytes)",
load.path,
md.len()
));
Some(error_message(GRAPH_STUB, None))
}
}
}
/// `render_frame`: the graph/render pipeline has no Rust backing, so a
/// render request is answered with a clear error carrying the ticket.
fn handle_render_frame(&mut self, msg: &Value) -> Option<Value> {
let render: RenderFrameMsg = match serde_json::from_value(msg.clone()) {
Ok(r) => r,
Err(_) => return Some(error_message("invalid render_frame message", None)),
};
Some(error_message(RENDER_STUB, Some(render.ticket)))
}
}
// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------
/// Full render-worker main, transport-agnostic in the backend name.
///
/// Mirrors `oakengine_worker_main()` in worker.cpp: create the session
/// (which initializes the render backend), load the runtime config, write
/// the startup handshake, then serve the NDJSON control loop on
/// stdin/stdout until a `shutdown` message or EOF. Returns the process
/// exit code.
pub fn worker_main(backend: &str) -> i32 {
// 1. Session creation initializes the render backend through the
// oakrender crate's direct Rust API
// (oakengine_worker_session_create()).
let mut session = match WorkerSession::create(backend) {
Ok(s) => s,
Err(msg) => {
log_error(&msg);
return 1;
}
};
if !session.has_renderer() {
// Mirrors oakengine_worker_main(): without a renderer the worker
// cannot do anything, so it exits 1. ("--backend none" lands here.)
log_error("no renderer initialized");
return 1;
}
// 2. Runtime services (config load etc.).
if !session.initialize_runtime() {
return 1;
}
// 3. Startup handshake before the loop (mirrors worker.cpp main).
let handshake = session.startup_handshake();
let stdout = io::stdout();
let mut out = io::BufWriter::new(stdout.lock());
if let Err(e) = write_message(&mut out, &handshake) {
log_error(&format!("failed to write startup handshake: {e}"));
return 1;
}
if let Err(e) = out.flush() {
log_error(&format!("failed to flush startup handshake: {e}"));
return 1;
}
// 4. NDJSON control loop until a shutdown message or EOF.
let stdin = io::stdin();
let mut reader = stdin.lock();
let mut line = String::new();
let mut exit_code = 0;
while !session.shutdown_requested() {
line.clear();
match reader.read_line(&mut line) {
Ok(0) => break, // EOF: the parent closed the control pipe.
Ok(_) => {}
Err(e) => {
log_error(&format!("failed to read control line: {e}"));
break;
}
}
if line.trim().is_empty() {
// Blank lines are skipped silently (read_message() semantics).
continue;
}
if let Some(response) = session.handle_line(&line) {
if let Err(e) = write_message(&mut out, &response) {
log_error(&format!("failed to write response: {e}"));
exit_code = 1;
break;
}
if let Err(e) = out.flush() {
log_error(&format!("failed to flush response: {e}"));
exit_code = 1;
break;
}
}
}
exit_code
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ipc::{FrameSlotPool, SharedMemoryRegion, ShmMode};
use serde_json::json;
use std::ptr;
fn test_key(name: &str) -> String {
static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
SharedMemoryRegion::make_key(i64::from(std::process::id()), (n & 0x7FFF) as i32)
+ &format!("-w-{name}")
}
/// The "parent" side of a handshake: create an output segment holding a
/// pool, optionally an input segment, and return the handshake message
/// plus the owner regions (kept alive by the caller).
fn parent_side(
slots: i32,
slot_bytes: i64,
input: bool,
) -> (Value, SharedMemoryRegion, Option<SharedMemoryRegion>) {
let out_key = test_key("out");
let out_bytes = FrameSlotPool::bytes_needed(slots as u32, slot_bytes as usize);
let mut out_region = SharedMemoryRegion::new();
assert!(
out_region.open(&out_key, out_bytes, ShmMode::Create),
"{}",
out_region.error()
);
// SAFETY: live mapping sized by bytes_needed.
let _pool =
unsafe { FrameSlotPool::create(out_region.data(), slots as u32, slot_bytes as usize) };
let (in_key, in_bytes, in_region) = if input {
let in_key = test_key("in");
let in_bytes = FrameSlotPool::bytes_needed(slots as u32, slot_bytes as usize);
let mut in_region = SharedMemoryRegion::new();
assert!(in_region.open(&in_key, in_bytes, ShmMode::Create));
// SAFETY: live mapping.
let _ = unsafe {
FrameSlotPool::create(in_region.data(), slots as u32, slot_bytes as usize)
};
(Some(in_key), Some(in_bytes), Some(in_region))
} else {
(None, None, None)
};
let hs = json!({
"type": "handshake",
"protocol_version": PROTOCOL_VERSION,
"shm_key": out_key,
"input_shm_key": in_key.unwrap_or_default(),
"input_slots": if input { slots } else { 0 },
"output_slots": slots,
"slot_data_bytes": slot_bytes,
"input_slot_data_bytes": in_bytes.unwrap_or(0),
});
(hs, out_region, in_region)
}
#[test]
fn no_backend_detection_matches_cpp() {
assert!(is_no_backend(""));
assert!(is_no_backend("none"));
assert!(is_no_backend("NONE"));
assert!(!is_no_backend("opengl"));
assert!(!is_no_backend("vulkan"));
}
#[test]
fn none_backend_session_has_no_renderer_but_serves_messages() {
let mut s = WorkerSession::create("none").unwrap();
assert!(!s.has_renderer());
let resp = s.handle_line(r#"{"type":"shutdown"}"#);
assert!(resp.is_none());
assert!(s.shutdown_requested());
}
#[test]
fn startup_handshake_is_protocol_version_1_with_empty_geometry() {
let s = WorkerSession::create("none").unwrap();
let hs = s.startup_handshake();
assert_eq!(
hs,
json!({
"type": "handshake",
"protocol_version": 1,
"shm_key": "",
"input_shm_key": "",
"input_slots": 0,
"output_slots": 0,
"slot_data_bytes": 0,
"input_slot_data_bytes": 0,
})
);
}
#[test]
fn malformed_line_yields_error_response() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s.handle_line("this is not json").unwrap();
assert_eq!(resp["type"], "error");
assert_eq!(resp["message"], "malformed control message");
}
#[test]
fn unknown_message_type_yields_error_response() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s.handle_line(r#"{"type":"frobnicate"}"#).unwrap();
assert_eq!(resp["message"], "unknown message type: frobnicate");
}
#[test]
fn cancel_and_shutdown_produce_no_response() {
let mut s = WorkerSession::create("none").unwrap();
assert!(s.handle_line(r#"{"type":"cancel","ticket":5}"#).is_none());
assert!(s.handle_line(r#"{"type":"shutdown"}"#).is_none());
assert!(s.shutdown_requested());
}
#[test]
fn handshake_wrong_protocol_version() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s
.handle_line(
r#"{"type":"handshake","protocol_version":99,"shm_key":"k","output_slots":1,"slot_data_bytes":16}"#,
)
.unwrap();
assert_eq!(resp["message"], "unsupported protocol version 99");
}
#[test]
fn handshake_missing_geometry() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s
.handle_line(r#"{"type":"handshake","protocol_version":1}"#)
.unwrap();
assert_eq!(
resp["message"],
"handshake missing output shared-memory geometry"
);
}
#[test]
fn handshake_attaches_real_output_pool() {
let mut s = WorkerSession::create("none").unwrap();
let (hs, out_region, _in) = parent_side(4, 4096, false);
let resp = s.handle_line(&hs.to_string());
assert!(resp.is_none(), "unexpected error: {resp:?}");
// The session now holds a real attached pool with the parent's
// geometry.
let out_pool = s.output_pool.as_ref().unwrap();
assert_eq!(out_pool.slot_count(), 4);
assert_eq!(out_pool.slot_data_bytes(), 4096);
// The two views share the same rings, not copies: the parent pops a
// free slot and the worker's pool sees the ring cursor move; the
// parent's publish lands in the worker's ready ring.
// SAFETY: `out_region` is a live mapping containing the pool the
// session attached to.
let parent_pool = unsafe { FrameSlotPool::attach(out_region.data()) };
let mut parent_slot = 0;
assert!(unsafe { parent_pool.acquire(&mut parent_slot) });
assert_eq!(parent_slot, 0);
let mut worker_slot = 0;
assert!(unsafe { out_pool.acquire(&mut worker_slot) });
assert_eq!(worker_slot, 1, "worker must see the parent's free-ring pop");
// SAFETY: `parent_slot` was acquired by the parent; slot_bytes
// writable.
unsafe {
ptr::write_bytes(parent_pool.slot_data(parent_slot), 0xAB, 64);
}
assert!(unsafe { parent_pool.publish(parent_slot) });
let mut consumed = 0;
assert!(unsafe { out_pool.consume(&mut consumed) });
assert_eq!(consumed, parent_slot);
// SAFETY: `consumed` was consumed by the worker's pool.
assert_eq!(unsafe { *out_pool.slot_data_const(consumed) }, 0xAB);
// Clean up so the region drop at test end unlinks cleanly.
unsafe { out_pool.release(consumed) };
unsafe { out_pool.release(worker_slot) };
}
#[test]
fn handshake_attaches_input_pool_too() {
let mut s = WorkerSession::create("none").unwrap();
let (hs, _out, _in) = parent_side(2, 256, true);
let resp = s.handle_line(&hs.to_string());
assert!(resp.is_none(), "unexpected error: {resp:?}");
assert!(s.input_pool.is_some());
let in_pool = s.input_pool.as_ref().unwrap();
assert_eq!(in_pool.slot_count(), 2);
assert_eq!(in_pool.slot_data_bytes(), 256);
}
#[test]
fn handshake_attach_failure_reports_error() {
let mut s = WorkerSession::create("none").unwrap();
// A key that was never created.
let resp = s
.handle_line(
&json!({
"type": "handshake",
"protocol_version": 1,
"shm_key": format!("olive-rw-{}-missing", std::process::id()),
"output_slots": 4,
"slot_data_bytes": 4096,
})
.to_string(),
)
.unwrap();
assert_eq!(resp["type"], "error");
assert!(resp["message"]
.as_str()
.unwrap()
.starts_with("failed to attach shared memory: "));
assert!(s.output_pool.is_none());
}
#[test]
fn handshake_rejects_non_pool_segment() {
let mut s = WorkerSession::create("none").unwrap();
// A real segment of the right size that does not contain a pool
// (zeroed memory → wrong magic). Sized so the attach size check
// passes and the magic check fires.
let key = test_key("nopool");
let bytes = FrameSlotPool::bytes_needed(4, 4096);
let mut region = SharedMemoryRegion::new();
assert!(region.open(&key, bytes, ShmMode::Create));
let resp = s
.handle_line(
&json!({
"type": "handshake",
"protocol_version": 1,
"shm_key": key,
"output_slots": 4,
"slot_data_bytes": 4096,
})
.to_string(),
)
.unwrap();
assert_eq!(
resp["message"],
"shared memory does not contain a frame slot pool"
);
}
#[test]
fn handshake_missing_input_geometry_is_an_error() {
let mut s = WorkerSession::create("none").unwrap();
let (mut hs, _out, _in) = parent_side(2, 256, false);
// Ask for input slots without announcing their geometry.
hs["input_slots"] = json!(2);
let resp = s.handle_line(&hs.to_string()).unwrap();
assert_eq!(
resp["message"],
"handshake missing input shared-memory geometry"
);
}
#[test]
fn load_graph_checks_are_real_then_stub() {
let mut s = WorkerSession::create("none").unwrap();
let missing = "/definitely/not/a/real/graph.ove";
let resp = s
.handle_line(&json!({ "type": "load_graph", "path": missing }).to_string())
.unwrap();
assert_eq!(
resp["message"],
format!("graph file does not exist: {missing}")
);
let empty = std::env::temp_dir().join("oak_worker_main_test_empty.ove");
std::fs::write(&empty, b"").unwrap();
let resp = s
.handle_line(
&json!({ "type": "load_graph", "path": empty.display().to_string() }).to_string(),
)
.unwrap();
assert_eq!(
resp["message"],
format!("graph file is empty: {}", empty.display())
);
let _ = std::fs::remove_file(&empty);
let real = std::env::temp_dir().join("oak_worker_main_test_graph.ove");
std::fs::write(&real, b"<root/>").unwrap();
let resp = s
.handle_line(
&json!({ "type": "load_graph", "path": real.display().to_string() }).to_string(),
)
.unwrap();
assert!(resp["message"]
.as_str()
.unwrap()
.contains("node-graph deserialization is not yet available"));
let _ = std::fs::remove_file(&real);
}
#[test]
fn render_frame_reports_stub_with_ticket() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s
.handle_line(r#"{"type":"render_frame","ticket":123,"node":"abc"}"#)
.unwrap();
assert_eq!(resp["type"], "error");
assert_eq!(resp["ticket"], 123);
assert!(resp["message"]
.as_str()
.unwrap()
.contains("frame rendering is not yet available"));
}
// ---- oak-worker's in-process session tests (M14 R2: folded from the
// ---- former src/session.rs mirror; the facade's production session
// ---- tests above cover the rest) -------------------------------------
#[test]
fn session_starts_without_pools() {
let s = WorkerSession::create("none").unwrap();
assert!(s.output_pool.is_none());
assert!(s.input_pool.is_none());
assert!(!s.shutdown_requested());
}
#[test]
fn non_object_json_yields_error_response() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s.handle_line("[1,2,3]").unwrap();
assert_eq!(resp["message"], "malformed control message");
}
#[test]
fn missing_type_field_yields_unknown_error() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s.handle_line(r#"{"hello":1}"#).unwrap();
assert_eq!(resp["message"], "unknown message type: ");
}
#[test]
fn handshake_bad_json_shape_is_invalid_handshake() {
let mut s = WorkerSession::create("none").unwrap();
let resp = s
.handle_line(r#"{"type":"handshake","protocol_version":"x"}"#)
.unwrap();
assert_eq!(resp["message"], "invalid handshake message");
}
}