From 89df02edbc69db4e4f27a621ae0a7014e6f4cc67 Mon Sep 17 00:00:00 2001 From: can1357 Date: Sun, 1 Feb 2026 15:52:22 +0100 Subject: [PATCH] refactor(pi-natives): replaced tokio::task::spawn_blocking with unified launch_task abstraction - Replaced tokio::task::spawn_blocking with custom launch_task abstraction across clipboard, find, grep, html, image, and shell modules. - Introduced new work module providing Rayon-backed task scheduling for blocking operations in N-API exports. - Simplified error handling by removing redundant join error mapping and consolidating to single-level error propagation. - Unified task spawning pattern across all native modules to use consistent launch_task API with wait() method. --- crates/pi-natives/src/clipboard.rs | 16 +++--- crates/pi-natives/src/find.rs | 7 +-- crates/pi-natives/src/grep.rs | 11 +++-- crates/pi-natives/src/html.rs | 8 +-- crates/pi-natives/src/image.rs | 22 +++++---- crates/pi-natives/src/lib.rs | 1 + crates/pi-natives/src/shell.rs | 14 ++++-- crates/pi-natives/src/work.rs | 78 ++++++++++++++++++++++++++++++ 8 files changed, 124 insertions(+), 33 deletions(-) create mode 100644 crates/pi-natives/src/work.rs diff --git a/crates/pi-natives/src/clipboard.rs b/crates/pi-natives/src/clipboard.rs index 529f13d1c..41a797e3c 100644 --- a/crates/pi-natives/src/clipboard.rs +++ b/crates/pi-natives/src/clipboard.rs @@ -18,9 +18,11 @@ use std::io::Cursor; use arboard::{Clipboard, Error as ClipboardError, ImageData}; use image::{DynamicImage, ImageFormat, RgbaImage}; -use napi::{bindgen_prelude::*, tokio::task::spawn_blocking}; +use napi::bindgen_prelude::*; use napi_derive::napi; +use crate::work::launch_task; + /// Clipboard image payload encoded as PNG bytes. #[napi(object)] pub struct ClipboardImage { @@ -56,7 +58,7 @@ fn encode_png(image: ImageData<'_>) -> Result> { /// Returns an error if clipboard access fails. #[napi(js_name = "copyToClipboard")] pub async fn copy_to_clipboard(text: String) -> Result<()> { - spawn_blocking(move || -> Result<()> { + launch_task(move || -> Result<()> { let mut clipboard = Clipboard::new() .map_err(|err| Error::from_reason(format!("Failed to access clipboard: {err}")))?; clipboard @@ -64,8 +66,8 @@ pub async fn copy_to_clipboard(text: String) -> Result<()> { .map_err(|err| Error::from_reason(format!("Failed to copy to clipboard: {err}")))?; Ok(()) }) - .await - .map_err(|err| Error::from_reason(format!("Clipboard task failed: {err}")))??; + .wait() + .await?; Ok(()) } @@ -77,7 +79,7 @@ pub async fn copy_to_clipboard(text: String) -> Result<()> { /// Returns an error if clipboard access fails or image encoding fails. #[napi(js_name = "readImageFromClipboard")] pub async fn read_image_from_clipboard() -> Result> { - let result = spawn_blocking(move || -> Result> { + let result = launch_task(move || -> Result> { let mut clipboard = Clipboard::new() .map_err(|err| Error::from_reason(format!("Failed to access clipboard: {err}")))?; match clipboard.get_image() { @@ -92,7 +94,7 @@ pub async fn read_image_from_clipboard() -> Result> { Err(err) => Err(Error::from_reason(format!("Failed to read clipboard image: {err}"))), } }) - .await - .map_err(|err| Error::from_reason(format!("Clipboard task failed: {err}")))??; + .wait() + .await?; Ok(result) } diff --git a/crates/pi-natives/src/find.rs b/crates/pi-natives/src/find.rs index ab422f15d..8de9f4370 100644 --- a/crates/pi-natives/src/find.rs +++ b/crates/pi-natives/src/find.rs @@ -20,10 +20,11 @@ use ignore::WalkBuilder; use napi::{ bindgen_prelude::*, threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode}, - tokio::task, }; use napi_derive::napi; +use crate::work::launch_task; + /// Options for discovering files and directories. #[napi(object)] pub struct FindOptions { @@ -316,7 +317,7 @@ pub async fn find( let mentions_node_modules = pattern.contains("node_modules"); let sort_by_mtime = sort_by_mtime.unwrap_or(false); - task::spawn_blocking(move || { + launch_task(move || { let cancelled = AtomicBool::new(false); let config = FindConfig { root: search_path, @@ -330,6 +331,6 @@ pub async fn find( }; run_find(config, on_match.as_ref(), &cancelled) }) + .wait() .await - .map_err(|err| Error::from_reason(format!("Join error: {err}")))? } diff --git a/crates/pi-natives/src/grep.rs b/crates/pi-natives/src/grep.rs index fae332d45..3ea7cd549 100644 --- a/crates/pi-natives/src/grep.rs +++ b/crates/pi-natives/src/grep.rs @@ -25,11 +25,12 @@ use napi::{ JsString, bindgen_prelude::*, threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode}, - tokio::task, }; use napi_derive::napi; use rayon::prelude::*; +use crate::work::launch_task; + const MAX_FILE_BYTES: u64 = 4 * 1024 * 1024; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -1033,9 +1034,9 @@ pub async fn grep( ThreadsafeFunction, >, ) -> Result { - task::spawn_blocking(move || grep_sync(options, on_match.as_ref())) + launch_task(move || grep_sync(options, on_match.as_ref())) + .wait() .await - .map_err(|err| Error::from_reason(format!("Join error: {err}")))? } // ============================================================================= @@ -1155,7 +1156,7 @@ fn fuzzy_find_sync(options: FuzzyFindOptions) -> Result { /// Matching file and directory entries. #[napi(js_name = "fuzzyFind")] pub async fn fuzzy_find(options: FuzzyFindOptions) -> Result { - task::spawn_blocking(move || fuzzy_find_sync(options)) + launch_task(move || fuzzy_find_sync(options)) + .wait() .await - .map_err(|err| Error::from_reason(format!("Join error: {err}")))? } diff --git a/crates/pi-natives/src/html.rs b/crates/pi-natives/src/html.rs index b39c2fa64..f91dcf704 100644 --- a/crates/pi-natives/src/html.rs +++ b/crates/pi-natives/src/html.rs @@ -1,9 +1,11 @@ //! HTML to Markdown conversion. use html_to_markdown_rs::{ConversionOptions, PreprocessingOptions, PreprocessingPreset, convert}; -use napi::{bindgen_prelude::*, tokio::task::spawn_blocking}; +use napi::bindgen_prelude::*; use napi_derive::napi; +use crate::work::launch_task; + /// Options for HTML to Markdown conversion. #[napi(object)] #[derive(Debug, Default)] @@ -29,7 +31,7 @@ pub async fn html_to_markdown( let clean_content = options.clean_content.unwrap_or(false); let skip_images = options.skip_images.unwrap_or(false); - spawn_blocking(move || { + launch_task(move || { let conversion_opts = ConversionOptions { skip_images, preprocessing: PreprocessingOptions { @@ -44,6 +46,6 @@ pub async fn html_to_markdown( convert(html.as_str(), Some(conversion_opts)) .map_err(|err| Error::from_reason(format!("Conversion error: {err}"))) }) + .wait() .await - .map_err(|err| Error::from_reason(format!("HTML conversion task failed: {err}")))? } diff --git a/crates/pi-natives/src/image.rs b/crates/pi-natives/src/image.rs index ea46979fa..c20907f39 100644 --- a/crates/pi-natives/src/image.rs +++ b/crates/pi-natives/src/image.rs @@ -13,9 +13,11 @@ use image::{ codecs::{jpeg::JpegEncoder, webp::WebPEncoder}, imageops::FilterType, }; -use napi::{bindgen_prelude::*, tokio::task::spawn_blocking}; +use napi::bindgen_prelude::*; use napi_derive::napi; +use crate::work::launch_task; + /// Sampling filter for resize operations. #[napi] pub enum SamplingFilter { @@ -60,7 +62,7 @@ impl PhotonImage { #[napi(factory, js_name = "parse")] pub async fn parse(bytes: Uint8Array) -> Result { let bytes = bytes.as_ref().to_vec(); - let img = spawn_blocking(move || -> Result { + let img = launch_task(move || -> Result { let reader = ImageReader::new(Cursor::new(bytes)) .with_guessed_format() .map_err(|e| Error::from_reason(format!("Failed to detect image format: {e}")))?; @@ -71,8 +73,8 @@ impl PhotonImage { Ok(img) }) - .await - .map_err(|e| Error::from_reason(format!("Image decode task failed: {e}")))??; + .wait() + .await?; Ok(Self { img: Arc::new(img) }) } @@ -102,9 +104,9 @@ impl PhotonImage { #[napi(js_name = "encode")] pub async fn encode(&self, format: u8, quality: u8) -> Result { let img = Arc::clone(&self.img); - let buffer = spawn_blocking(move || encode_image(&img, format, quality)) - .await - .map_err(|e| Error::from_reason(format!("Encode task failed: {e}")))??; + let buffer: Vec = launch_task(move || encode_image(&img, format, quality)) + .wait() + .await?; Ok(Uint8Array::from(buffer)) } @@ -113,9 +115,9 @@ impl PhotonImage { #[napi(js_name = "resize")] pub async fn resize(&self, width: u32, height: u32, filter: SamplingFilter) -> Result { let img = Arc::clone(&self.img); - let resized = spawn_blocking(move || img.resize_exact(width, height, filter.into())) - .await - .map_err(|e| Error::from_reason(format!("Resize task failed: {e}")))?; + let resized = launch_task(move || Ok(img.resize_exact(width, height, filter.into()))) + .wait() + .await?; Ok(Self { img: Arc::new(resized) }) } } diff --git a/crates/pi-natives/src/lib.rs b/crates/pi-natives/src/lib.rs index 63e403b6b..18bb67ae2 100644 --- a/crates/pi-natives/src/lib.rs +++ b/crates/pi-natives/src/lib.rs @@ -31,3 +31,4 @@ pub mod ps; pub mod shell; pub mod system_info; pub mod text; +pub mod work; diff --git a/crates/pi-natives/src/shell.rs b/crates/pi-natives/src/shell.rs index 1fb7dd671..54941cc3d 100644 --- a/crates/pi-natives/src/shell.rs +++ b/crates/pi-natives/src/shell.rs @@ -36,7 +36,7 @@ use napi::{ use napi_derive::napi; use parking_lot::Mutex; -use crate::ps; +use crate::{ps, work::launch_task}; type ExecutionMap = HashMap; type SessionMap = HashMap>>; @@ -161,8 +161,9 @@ impl Shell { /// Run a shell command using the provided options. /// - /// The `on_chunk` callback receives streamed stdout/stderr output. Returns the - /// exit code when the command completes, or flags when cancelled or timed out. + /// The `on_chunk` callback receives streamed stdout/stderr output. Returns + /// the exit code when the command completes, or flags when cancelled or + /// timed out. #[napi] pub async fn run( &self, @@ -496,7 +497,10 @@ async fn run_shell_command( } } - let reader_handle = task::spawn_blocking(move || read_output(reader_file, on_chunk)); + let reader_handle = launch_task(move || -> Result<()> { + read_output(reader_file, on_chunk); + Ok(()) + }); let result = session .shell .run_string(options.command.clone(), ¶ms) @@ -512,7 +516,7 @@ async fn run_shell_command( drop(params); - let _ = reader_handle.await; + let _ = reader_handle.wait().await; result.map_err(|err| Error::from_reason(format!("Shell execution failed: {err}"))) } diff --git a/crates/pi-natives/src/work.rs b/crates/pi-natives/src/work.rs new file mode 100644 index 000000000..4227f9295 --- /dev/null +++ b/crates/pi-natives/src/work.rs @@ -0,0 +1,78 @@ +//! Rayon-backed scheduling for blocking work in N-API exports. +//! +//! # Overview +//! Runs CPU-bound or blocking Rust work on a shared Rayon thread pool instead +//! of Tokio's limited blocking workers. +//! +//! # Example +//! ```ignore +//! use pi_natives::work::launch_task; +//! +//! # async fn demo() -> napi::Result<()> { +//! let handle = launch_task(|| Ok(42)); +//! let value = handle.wait().await?; +//! assert_eq!(value, 42); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Architecture +//! ```text +//! JS async -> N-API -> launch_task -> Rayon thread pool +//! ``` + +use std::{ + panic::{AssertUnwindSafe, catch_unwind}, + sync::OnceLock, +}; + +use napi::{Error, Result}; +use rayon::{ThreadPool, ThreadPoolBuilder}; +use tokio::sync::oneshot; + +/// Handle for a scheduled blocking task. +pub struct WorkHandle { + receiver: oneshot::Receiver>, +} + +impl WorkHandle { + /// Await completion of the scheduled work. + /// + /// # Errors + /// Returns an error if the task panics or the channel is cancelled. + pub async fn wait(self) -> Result { + match self.receiver.await { + Ok(result) => result, + Err(_) => Err(Error::from_reason("Rayon task cancelled")), + } + } +} + +/// Schedule blocking work on the shared Rayon pool. +/// +/// # Errors +/// The returned handle resolves to an error if the task panics or is cancelled. +pub fn launch_task(work: F) -> WorkHandle +where + F: FnOnce() -> Result + Send + 'static, + T: Send + 'static, +{ + let pool = thread_pool(); + let (sender, receiver) = oneshot::channel(); + pool.spawn(move || { + let result = catch_unwind(AssertUnwindSafe(work)) + .unwrap_or_else(|_| Err(Error::from_reason("Rayon task panicked"))); + let _ = sender.send(result); + }); + WorkHandle { receiver } +} + +fn thread_pool() -> &'static ThreadPool { + static POOL: OnceLock = OnceLock::new(); + POOL.get_or_init(|| { + ThreadPoolBuilder::new() + .thread_name(|index| format!("pi-natives-{}", index)) + .build() + .expect("Failed to build Rayon thread pool") + }) +}