From 8eb199bfc73249bf371d15dd6a9ab4f4348bc11c Mon Sep 17 00:00:00 2001 From: Sebastian <106036+schoeller@users.noreply.github.com> Date: Wed, 24 Jun 2026 10:27:07 +0200 Subject: [PATCH] Fix 2: optimize src --- crates/ocs_plugin_api/src/lib.rs | 2 +- crates/ocs_plugin_api/src/process.rs | 120 ++++++---- crates/ocs_plugin_api/src/process/manager.rs | 223 +++++++++++++++++++ src/plugin/external.rs | 98 +++----- src/plugin/registry.rs | 74 ++---- 5 files changed, 360 insertions(+), 157 deletions(-) create mode 100644 crates/ocs_plugin_api/src/process/manager.rs diff --git a/crates/ocs_plugin_api/src/lib.rs b/crates/ocs_plugin_api/src/lib.rs index b9ec12814..4140cc5e1 100644 --- a/crates/ocs_plugin_api/src/lib.rs +++ b/crates/ocs_plugin_api/src/lib.rs @@ -43,4 +43,4 @@ pub use ribbon::{ }; #[cfg(feature = "host")] -pub use process::{PluginError, PluginProcess}; +pub use process::{DispatchResult, PluginError, PluginManager, PluginProcess}; diff --git a/crates/ocs_plugin_api/src/process.rs b/crates/ocs_plugin_api/src/process.rs index 007fcdd83..5424fa055 100644 --- a/crates/ocs_plugin_api/src/process.rs +++ b/crates/ocs_plugin_api/src/process.rs @@ -18,6 +18,11 @@ use crate::ipc::server::handle_plugin_request; use crate::ipc::transport::{recv, send}; use crate::ribbon::owned::{OwnedPluginManifest, OwnedRibbonGroup as OwnedRibbonGroupAlias}; +use serde::de::DeserializeOwned; + +mod manager; +pub use manager::{DispatchResult, PluginManager}; + /// Maximum time to wait for the plugin runner to connect back to the host. const SPAWN_TIMEOUT: Duration = Duration::from_secs(10); @@ -47,8 +52,8 @@ pub enum PluginError { /// One spawned plugin process. pub struct PluginProcess { - stream: Mutex, - child: Mutex, + stream: Mutex>, + child: Mutex>, id: String, manifest: OwnedPluginManifest, ribbon: Vec, @@ -88,7 +93,7 @@ impl PluginProcess { let stream = match rx.recv_timeout(spawn_timeout()) { Ok(Ok(stream)) => { eprintln!("[plugin] runner connected"); - Mutex::new(stream) + Mutex::new(Some(stream)) } Ok(Err(e)) => return Err(e.into()), Err(mpsc::RecvTimeoutError::Timeout) => { @@ -116,7 +121,7 @@ impl PluginProcess { let id = manifest.id.clone(); Ok(Self { stream, - child: Mutex::new(child), + child: Mutex::new(Some(child)), id, manifest, ribbon, @@ -165,12 +170,9 @@ impl PluginProcess { command_id: u64, event: InteractiveEvent, ) -> Result { - send( - &mut self.stream.lock().unwrap(), - &HostToPlugin::Request(HostRequest::InteractiveEvent { command_id, event }), - )?; + self.send_request(HostRequest::InteractiveEvent { command_id, event })?; loop { - match recv::(&mut self.stream.lock().unwrap())? { + match self.recv_response::()? { PluginToHost::Response(HostResponse::CommandStep(s)) => return Ok(s), PluginToHost::Response(other) => { return Err(PluginError::UnexpectedResponse(other)) @@ -179,7 +181,7 @@ impl PluginProcess { let resp = crate::ipc::protocol::PluginResponse::Error(format!( "unexpected nested request during interactive event: {req:?}" )); - send(&mut self.stream.lock().unwrap(), &HostToPlugin::Response(resp))?; + self.send_response(resp)?; } } } @@ -187,12 +189,9 @@ impl PluginProcess { /// Ask the plugin process for the current prompt of an interactive command. pub fn get_prompt(&self, command_id: u64) -> Result { - send( - &mut self.stream.lock().unwrap(), - &HostToPlugin::Request(HostRequest::GetPrompt { command_id }), - )?; + self.send_request(HostRequest::GetPrompt { command_id })?; loop { - match recv::(&mut self.stream.lock().unwrap())? { + match self.recv_response::()? { PluginToHost::Response(HostResponse::Text(s)) => return Ok(s), PluginToHost::Response(other) => { return Err(PluginError::UnexpectedResponse(other)) @@ -201,7 +200,7 @@ impl PluginProcess { let resp = crate::ipc::protocol::PluginResponse::Error(format!( "unexpected nested request during get_prompt: {req:?}" )); - send(&mut self.stream.lock().unwrap(), &HostToPlugin::Response(resp))?; + self.send_response(resp)?; } } } @@ -209,12 +208,9 @@ impl PluginProcess { /// Ask the plugin process whether an interactive command wants object picks. pub fn needs_entity_pick(&self, command_id: u64) -> Result { - send( - &mut self.stream.lock().unwrap(), - &HostToPlugin::Request(HostRequest::NeedsEntityPick { command_id }), - )?; + self.send_request(HostRequest::NeedsEntityPick { command_id })?; loop { - match recv::(&mut self.stream.lock().unwrap())? { + match self.recv_response::()? { PluginToHost::Response(HostResponse::Bool(b)) => return Ok(b), PluginToHost::Response(other) => { return Err(PluginError::UnexpectedResponse(other)) @@ -223,63 +219,105 @@ impl PluginProcess { let resp = crate::ipc::protocol::PluginResponse::Error(format!( "unexpected nested request during needs_entity_pick: {req:?}" )); - send(&mut self.stream.lock().unwrap(), &HostToPlugin::Response(resp))?; + self.send_response(resp)?; } } } } pub fn is_alive(&self) -> bool { - match self.child.lock().unwrap().try_wait() { - Ok(None) => true, - Ok(Some(_)) | Err(_) => false, + let mut guard = self.child.lock().unwrap_or_else(|e| e.into_inner()); + match guard.as_mut() { + Some(child) => match child.try_wait() { + Ok(None) => true, + Ok(Some(_)) | Err(_) => false, + }, + None => false, } } - pub fn kill(&self) -> std::io::Result<()> { - let _ = call_no_host(&self.stream, HostRequest::Shutdown); - self.child.lock().unwrap().kill() + /// Tear down the plugin process without blocking the caller. The stream is + /// closed and the child is killed and reaped in a detached background thread. + pub fn shutdown(&self) { + let stream = self.stream.lock().unwrap_or_else(|e| e.into_inner()).take(); + let child = self.child.lock().unwrap_or_else(|e| e.into_inner()).take(); + std::thread::spawn(move || { + drop(stream); + if let Some(mut child) = child { + let _ = child.kill(); + let _ = child.wait(); + } + }); } } impl Drop for PluginProcess { fn drop(&mut self) { - let _ = self.kill(); + self.shutdown(); } } +impl PluginProcess { + fn send_request(&self, req: HostRequest) -> Result<(), PluginError> { + let mut guard = self.stream.lock().unwrap_or_else(|e| e.into_inner()); + let stream = guard.as_mut().ok_or_else(shutdown_error)?; + send(stream, &HostToPlugin::Request(req)).map_err(Into::into) + } + + fn send_response(&self, resp: crate::ipc::protocol::PluginResponse) -> Result<(), PluginError> { + let mut guard = self.stream.lock().unwrap_or_else(|e| e.into_inner()); + let stream = guard.as_mut().ok_or_else(shutdown_error)?; + send(stream, &HostToPlugin::Response(resp)).map_err(Into::into) + } + + fn recv_response(&self) -> Result { + let mut guard = self.stream.lock().unwrap_or_else(|e| e.into_inner()); + let stream = guard.as_mut().ok_or_else(shutdown_error)?; + recv(stream).map_err(Into::into) + } +} + +fn shutdown_error() -> PluginError { + PluginError::Io(std::io::Error::new( + std::io::ErrorKind::NotConnected, + "plugin process has been shut down", + )) +} + /// Send a host request and wait for the response, handling any nested plugin /// requests inline using the supplied `HostApi`. fn call( - stream: &Mutex, + stream: &Mutex>, host: &mut dyn HostApi, req: HostRequest, on_start_interactive: &mut dyn FnMut(u64), ) -> Result { eprintln!("[plugin] host -> runner: {req:?}"); - send(&mut stream.lock().unwrap(), &HostToPlugin::Request(req))?; + { + let mut guard = stream.lock().unwrap_or_else(|e| e.into_inner()); + let stream = guard.as_mut().ok_or_else(shutdown_error)?; + send(stream, &HostToPlugin::Request(req))?; + } loop { - let msg = recv::(&mut stream.lock().unwrap())?; + let msg = { + let mut guard = stream.lock().unwrap_or_else(|e| e.into_inner()); + let stream = guard.as_mut().ok_or_else(shutdown_error)?; + recv::(stream)? + }; eprintln!("[plugin] runner -> host: {msg:?}"); match msg { PluginToHost::Response(resp) => return Ok(resp), PluginToHost::Request(plugin_req) => { let resp = handle_plugin_request(host, plugin_req, on_start_interactive); eprintln!("[plugin] host -> runner response: {resp:?}"); - send(&mut stream.lock().unwrap(), &HostToPlugin::Response(resp))?; + let mut guard = stream.lock().unwrap_or_else(|e| e.into_inner()); + let stream = guard.as_mut().ok_or_else(shutdown_error)?; + send(stream, &HostToPlugin::Response(resp))?; } } } } -/// Best-effort shutdown request that does not wait for a response. -fn call_no_host( - stream: &Mutex, - req: HostRequest, -) -> Result<(), crate::ipc::transport::TransportError> { - send(&mut stream.lock().unwrap(), &HostToPlugin::Request(req)) -} - /// Locate the executable to spawn for running a plugin. /// /// The host spawns *itself* in runner mode (`--ocs-plugin-runner`), so the diff --git a/crates/ocs_plugin_api/src/process/manager.rs b/crates/ocs_plugin_api/src/process/manager.rs new file mode 100644 index 000000000..06ca1bf75 --- /dev/null +++ b/crates/ocs_plugin_api/src/process/manager.rs @@ -0,0 +1,223 @@ +//! Process manager for out-of-process plugins. + +use std::path::Path; +use std::sync::Arc; + +use crate::host::HostApi; +use crate::process::{PluginError, PluginProcess}; +use crate::ribbon::owned::{to_shared_module, SharedCadModule}; + +/// Owner of every spawned plugin process. +pub struct PluginManager { + plugins: Vec, +} + +struct LoadedPlugin { + process: Arc, + module: SharedCadModule, +} + +impl Default for PluginManager { + fn default() -> Self { + Self::new() + } +} + +/// Outcome of [`PluginManager::dispatch`]. +#[derive(Default)] +pub struct DispatchResult { + /// A plugin handled the command. + pub handled: bool, + /// An interactive command was started by a plugin. + pub started: Option<(Arc, u64)>, + /// Plugins whose process died before or during dispatch. + pub dead_plugins: Vec, + /// Plugins that returned an error while trying to handle the command. + pub errors: Vec<(String, String)>, +} + +impl PluginManager { + /// Create an empty manager. + pub fn new() -> Self { + Self { plugins: Vec::new() } + } + + /// Spawn `cdylib_path` as a separate plugin process, build its ribbon + /// module, and store it. Returns the plugin id on success. + pub fn load( + &mut self, + cdylib_path: &Path, + host: &mut dyn HostApi, + ) -> Result { + let process = PluginProcess::spawn(cdylib_path, host)?; + let id = process.id().to_string(); + let name = process.manifest().name.clone(); + let module = to_shared_module(id.clone(), name, process.ribbon().to_vec()); + self.plugins.push(LoadedPlugin { + process: Arc::new(process), + module, + }); + Ok(id) + } + + /// Ribbon modules for alive, non-disabled plugins, sorted by `ribbon_order`. + pub fn ribbon_modules bool>( + &self, + is_disabled: F, + ) -> Vec<(i32, SharedCadModule)> { + let mut out: Vec<(i32, SharedCadModule)> = self + .plugins + .iter() + .filter(|p| !is_disabled(p.process.id()) && p.process.is_alive()) + .map(|p| (p.process.manifest().ribbon_order, p.module.clone())) + .collect(); + out.sort_by_key(|(order, _)| *order); + out + } + + /// Dispatch `cmd` to each plugin until one handles it. + /// + /// `is_disabled` is called for each plugin id so the host can filter + /// disabled plugins without exposing its set type to the crate. + pub fn dispatch bool>( + &self, + host: &mut dyn HostApi, + cmd: &str, + is_disabled: F, + ) -> DispatchResult { + let mut result = DispatchResult::default(); + for p in &self.plugins { + let id = p.process.id().to_string(); + if is_disabled(&id) { + continue; + } + if !p.process.is_alive() { + result.dead_plugins.push(id); + continue; + } + let process = Arc::clone(&p.process); + let mut on_start = |command_id: u64| { + result.started = Some((Arc::clone(&process), command_id)); + }; + match p.process.dispatch(host, cmd, &mut on_start) { + Ok(true) => { + result.handled = true; + return result; + } + Ok(false) => {} + Err(e) => result.errors.push((id, e.to_string())), + } + } + result + } + + /// Plugin ids currently loaded. + pub fn ids(&self) -> Vec { + self.plugins + .iter() + .map(|p| p.process.id().to_string()) + .collect() + } + + /// Begin asynchronous shutdown of every plugin process. + pub fn shutdown_all(&mut self) { + let plugins = std::mem::take(&mut self.plugins); + for p in plugins { + p.process.shutdown(); + } + } +} + +impl Drop for PluginManager { + fn drop(&mut self) { + self.shutdown_all(); + } +} + +#[cfg(all(test, feature = "host"))] +mod tests { + use super::*; + + #[test] + fn empty_manager_has_no_ribbon_modules() { + let manager = PluginManager::new(); + assert!(manager.ribbon_modules(|_| false).is_empty()); + assert!(manager.ids().is_empty()); + } + + #[test] + fn dispatch_with_no_plugins_is_not_handled() { + struct DummyHost; + impl HostApi for DummyHost { + fn tab_index(&self) -> usize { + 0 + } + fn document(&self) -> &acadrust::CadDocument { + panic!("not used") + } + fn document_mut(&mut self) -> &mut acadrust::CadDocument { + panic!("not used") + } + fn add_entity(&mut self, _entity: acadrust::EntityType) -> acadrust::Handle { + panic!("not used") + } + fn bump_geometry(&mut self) {} + fn read_record( + &self, + _handle: acadrust::Handle, + _app_name: &str, + ) -> Option<&acadrust::xdata::ExtendedDataRecord> { + None + } + fn write_record( + &mut self, + _handle: acadrust::Handle, + _record: acadrust::xdata::ExtendedDataRecord, + ) -> bool { + false + } + fn remove_record( + &mut self, + _handle: acadrust::Handle, + _app_name: &str, + ) -> bool { + false + } + fn push_undo(&mut self, _label: &str) {} + fn set_dirty(&mut self) {} + fn push_info(&mut self, _msg: &str) {} + fn push_output(&mut self, _msg: &str) {} + fn push_error(&mut self, _msg: &str) {} + fn start_interactive( + &mut self, + _command: Box, + ) { + } + fn plugin_state_any( + &self, + _plugin_id: &str, + ) -> Option<&(dyn std::any::Any + Send + Sync)> { + None + } + fn plugin_state_any_mut( + &mut self, + _plugin_id: &str, + ) -> Option<&mut (dyn std::any::Any + Send + Sync)> { + None + } + fn ensure_plugin_state_any( + &mut self, + _plugin_id: &'static str, + _init: &mut dyn FnMut() -> Box, + ) -> &mut (dyn std::any::Any + Send + Sync) { + panic!("not used") + } + } + + let manager = PluginManager::new(); + let mut host = DummyHost; + let result = manager.dispatch(&mut host, "FOO", |_| false); + assert!(!result.handled); + assert!(result.started.is_none()); + } +} diff --git a/src/plugin/external.rs b/src/plugin/external.rs index d270c1ff2..360d0fd7f 100644 --- a/src/plugin/external.rs +++ b/src/plugin/external.rs @@ -221,7 +221,7 @@ fn parse_string_array(s: &str) -> Vec { // ── Runtime loading (desktop only) ────────────────────────────────────────── #[cfg(not(target_arch = "wasm32"))] -pub(crate) use loader::with_loaded; +pub(crate) use loader::with_manager; #[cfg(all(not(target_arch = "wasm32"), not(test)))] pub(crate) use loader::{load_at_startup, loaded_ids}; @@ -229,61 +229,60 @@ pub(crate) use loader::{load_at_startup, loaded_ids}; #[cfg(not(target_arch = "wasm32"))] #[cfg_attr(test, allow(dead_code))] mod loader { - use super::{lib_extension, ExternalPlugin}; - use std::path::{Path, PathBuf}; - use std::sync::Arc; - - /// A loaded external plugin. Holds the spawned process and a shareable - /// ribbon module built from the process's cached ribbon data. - pub struct LoadedPlugin { - pub process: Arc, - pub module: ocs_plugin_api::ribbon::owned::SharedCadModule, - pub id: String, - } - + use super::lib_extension; + use ocs_plugin_api::process::PluginManager; use std::cell::RefCell; + use std::path::{Path, PathBuf}; - // Process-wide store of spawned external plugins. The runner processes stay - // alive for the whole session; this is filled once at startup. + // Process-wide plugin manager. Drop kills every runner process asynchronously + // so host shutdown is never delayed by a plugin. thread_local! { - static LOADED: RefCell> = const { RefCell::new(Vec::new()) }; + static MANAGER: RefCell> = const { RefCell::new(None) }; } /// Discover packages and spawn every API-compatible one as a separate /// process. Call once at startup. Returns per-id results so the host can /// report load failures. - pub(crate) fn load_at_startup(app: &mut crate::app::OpenCADStudio) -> Vec<(String, Result<(), String>)> { + pub(crate) fn load_at_startup( + app: &mut crate::app::OpenCADStudio, + ) -> Vec<(String, Result<(), String>)> { let discovered = super::discover(); + let mut manager = PluginManager::new(); let mut out = Vec::new(); - LOADED.with(|cell| { - let mut store = cell.borrow_mut(); - if !store.is_empty() { - return; // already loaded this session + for d in &discovered { + if !d.api_compatible() || !d.lib_present { + continue; } - for d in &discovered { - if !d.api_compatible() || !d.lib_present { - continue; - } - match load(d, app) { - Ok(lp) => { - out.push((lp.id.clone(), Ok(()))); - store.push(lp); - } - Err(e) => out.push((d.id.clone(), Err(e))), - } + let Some(path) = lib_file(&d.dir) else { + out.push((d.id.clone(), Err("no native library in package".to_string()))); + continue; + }; + let mut host = crate::app::plugin_host::HostSession::new(app, 0); + match manager.load(&path, &mut host) { + Ok(id) => out.push((id, Ok(()))), + Err(e) => out.push((d.id.clone(), Err(e.to_string()))), } - }); + } + MANAGER.with(|m| *m.borrow_mut() = Some(manager)); out } /// Ids of the plugins currently loaded in the process store. pub fn loaded_ids() -> Vec { - LOADED.with(|c| c.borrow().iter().map(|lp| lp.id.clone()).collect()) + MANAGER.with(|m| m.borrow().as_ref().map(|mgr| mgr.ids()).unwrap_or_default()) } - /// Run `f` over the loaded plugins (borrowing the store). - pub fn with_loaded(f: impl FnOnce(&[LoadedPlugin]) -> R) -> R { - LOADED.with(|c| f(&c.borrow())) + /// Run `f` with a reference to the loaded plugin manager. + pub fn with_manager(f: impl FnOnce(&PluginManager) -> R) -> R { + MANAGER.with(|m| { + let guard = m.borrow(); + if let Some(manager) = guard.as_ref() { + return f(manager); + } + drop(guard); + let empty = PluginManager::new(); + f(&empty) + }) } /// Path to the native library beside `plugin.toml`, if any. @@ -294,31 +293,6 @@ mod loader { (p.extension().and_then(|s| s.to_str()) == Some(ext)).then_some(p) }) } - - /// Spawn a discovered package's `cdylib` in a separate process and cache - /// its ribbon module. The runner performs the API version gate before any - /// plugin code runs. - pub fn load( - p: &ExternalPlugin, - app: &mut crate::app::OpenCADStudio, - ) -> Result { - let path = lib_file(&p.dir).ok_or("no native library in package")?; - let mut host = crate::app::plugin_host::HostSession::new(app, 0); - let process = - ocs_plugin_api::process::PluginProcess::spawn(&path, &mut host).map_err(|e| e.to_string())?; - let id = process.id().to_string(); - let name = process.manifest().name.clone(); - let module = ocs_plugin_api::ribbon::owned::to_shared_module( - id.clone(), - name, - process.ribbon().to_vec(), - ); - Ok(LoadedPlugin { - process: Arc::new(process), - module, - id, - }) - } } #[cfg(test)] diff --git a/src/plugin/registry.rs b/src/plugin/registry.rs index 6372c5f03..756405190 100644 --- a/src/plugin/registry.rs +++ b/src/plugin/registry.rs @@ -17,24 +17,15 @@ pub fn ribbon_modules_enabled( ) -> Vec> { #[cfg_attr(target_arch = "wasm32", allow(unused_mut))] let mut core = core_registry::all_modules(); - // Dynamically-loaded external plugins contribute tabs (their libraries stay - // resident for the session, so these vtables remain valid). + // Dynamically-loaded external plugins contribute tabs via the crate manager. #[cfg(not(target_arch = "wasm32"))] { - let mut addons: Vec<(i32, Box)> = Vec::new(); - crate::plugin::external::with_loaded(|loaded| { - for lp in loaded { - if disabled.contains(lp.id.as_str()) || !lp.process.is_alive() { - continue; - } - addons.push(( - lp.process.manifest().ribbon_order, - Box::new(lp.module.clone()) as Box, - )); - } + crate::plugin::external::with_manager(|manager| { + let addons = manager.ribbon_modules(|id| disabled.contains(id)); + core.extend(addons.into_iter().map(|(_, module)| { + Box::new(module) as Box + })); }); - addons.sort_by_key(|(order, _)| *order); - core.extend(addons.into_iter().map(|(_, ribbon)| ribbon)); } let _ = disabled; core @@ -47,44 +38,20 @@ pub(crate) fn try_dispatch(app: &mut OpenCADStudio, tab: usize, cmd: &str) -> bo { use super::host::HostSession; let disabled = app.disabled_plugin_ids(); - let mut started: Option<(u64, std::sync::Arc)> = None; - let mut dead_plugins: Vec = Vec::new(); - let mut dispatch_errors: Vec<(String, String)> = Vec::new(); - let handled = crate::plugin::external::with_loaded(|loaded| { + let result = { let mut host = HostSession::new(app, tab); - for lp in loaded { - if disabled.contains(lp.id.as_str()) { - continue; - } - if !lp.process.is_alive() { - dead_plugins.push(lp.id.clone()); - continue; - } - let process = std::sync::Arc::clone(&lp.process); - let mut start = |command_id: u64| { - started = Some((command_id, std::sync::Arc::clone(&process))); - }; - match crate::plugin::guard("dispatch", || lp.process.dispatch(&mut host, cmd, &mut start)) { - Some(Ok(true)) => return true, - Some(Ok(false)) => {} - Some(Err(e)) => { - eprintln!("[plugin] dispatch error for '{}': {e}", lp.id); - dispatch_errors.push((lp.id.clone(), e.to_string())); - } - None => { - // Panic already logged by guard. - } - } - } - false - }); - for id in dead_plugins { + crate::plugin::external::with_manager(|manager| { + manager.dispatch(&mut host, cmd, |id| disabled.contains(id)) + }) + }; + + for id in result.dead_plugins { app.push_plugin_error(&format!("Plugin '{id}' process died; skipping dispatch")); } - for (id, err) in dispatch_errors { + for (id, err) in result.errors { app.push_plugin_error(&format!("Plugin '{id}' dispatch error: {err}")); } - if let Some((command_id, process)) = started { + if let Some((process, command_id)) = result.started { app.set_active_command( tab, Box::new(crate::app::plugin_host::PluginProcessInteractiveAdapter::new( @@ -93,12 +60,13 @@ pub(crate) fn try_dispatch(app: &mut OpenCADStudio, tab: usize, cmd: &str) -> bo )), ); } - if handled { - return true; - } + return result.handled; + } + #[cfg(target_arch = "wasm32")] + { + let _ = (app, tab, cmd); + false } - let _ = (app, tab, cmd); - false } #[cfg(test)]