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.
This commit is contained in:
can1357
2026-02-01 15:52:22 +01:00
parent 96b83875f4
commit 89df02edbc
8 changed files with 124 additions and 33 deletions
+9 -7
View File
@@ -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<Vec<u8>> {
/// 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<Option<ClipboardImage>> {
let result = spawn_blocking(move || -> Result<Option<ClipboardImage>> {
let result = launch_task(move || -> Result<Option<ClipboardImage>> {
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<Option<ClipboardImage>> {
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)
}
+4 -3
View File
@@ -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}")))?
}
+6 -5
View File
@@ -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<GrepMatch>,
>,
) -> Result<GrepResult> {
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<FuzzyFindResult> {
/// Matching file and directory entries.
#[napi(js_name = "fuzzyFind")]
pub async fn fuzzy_find(options: FuzzyFindOptions) -> Result<FuzzyFindResult> {
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}")))?
}
+5 -3
View File
@@ -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}")))?
}
+12 -10
View File
@@ -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<Self> {
let bytes = bytes.as_ref().to_vec();
let img = spawn_blocking(move || -> Result<DynamicImage> {
let img = launch_task(move || -> Result<DynamicImage> {
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<Uint8Array> {
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<u8> = 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<Self> {
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) })
}
}
+1
View File
@@ -31,3 +31,4 @@ pub mod ps;
pub mod shell;
pub mod system_info;
pub mod text;
pub mod work;
+9 -5
View File
@@ -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<String, ExecutionControl>;
type SessionMap = HashMap<String, Arc<TokioMutex<ShellSession>>>;
@@ -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(), &params)
@@ -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}")))
}
+78
View File
@@ -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<T> {
receiver: oneshot::Receiver<Result<T>>,
}
impl<T> WorkHandle<T> {
/// 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<T> {
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<F, T>(work: F) -> WorkHandle<T>
where
F: FnOnce() -> Result<T> + 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<ThreadPool> = OnceLock::new();
POOL.get_or_init(|| {
ThreadPoolBuilder::new()
.thread_name(|index| format!("pi-natives-{}", index))
.build()
.expect("Failed to build Rayon thread pool")
})
}