diff --git a/frb_rust/src/for_generated/mod.rs b/frb_rust/src/for_generated/mod.rs index 00389e65b5d..b73587ce79a 100644 --- a/frb_rust/src/for_generated/mod.rs +++ b/frb_rust/src/for_generated/mod.rs @@ -26,10 +26,12 @@ pub use crate::handler::error_listener::ErrorListener; pub use crate::handler::executor::Executor; pub use crate::handler::handler::{FfiCallMode, TaskInfo, Thread}; pub use crate::handler::handler::{TaskContext, TaskRetFutTrait}; -pub use crate::handler::implementation::thread_executor::{ThreadExecutor, ThreadJob, ThreadRouter}; pub use crate::handler::implementation::error_listener::NoOpErrorListener; pub use crate::handler::implementation::executor::SimpleExecutor; pub use crate::handler::implementation::handler::SimpleHandler; +pub use crate::handler::implementation::thread_executor::{ + ThreadExecutor, ThreadJob, ThreadRouter, +}; pub use crate::lifetimeable::lifetime_changer::{ ouroboros_change_lifetime, ouroboros_change_lifetime_mut, }; @@ -63,9 +65,9 @@ pub use crate::rust_auto_opaque::rust2dart_explicit::rust_auto_opaque_explicit_e pub use crate::rust_auto_opaque::{inner::RustAutoOpaqueInner, RustAutoOpaqueBase}; pub use crate::rust_opaque::{dart2rust::decode_rust_opaque_nom, RustOpaqueBase}; pub use crate::stream::stream_sink::StreamSinkBase; -pub use crate::thread_pool::{BaseThreadPool, SimpleThreadPool}; #[cfg(all(feature = "rust-async", not(target_family = "wasm")))] pub use crate::thread_pool::SingleThreadAsyncLane; +pub use crate::thread_pool::{BaseThreadPool, SimpleThreadPool}; #[cfg(target_family = "wasm")] pub use crate::web_transfer::transfer_closure::TransferClosure; #[cfg(feature = "anyhow")] diff --git a/frb_rust/src/handler/implementation/thread_executor.rs b/frb_rust/src/handler/implementation/thread_executor.rs index b5c36d12165..b1fb2f9e380 100644 --- a/frb_rust/src/handler/implementation/thread_executor.rs +++ b/frb_rust/src/handler/implementation/thread_executor.rs @@ -51,8 +51,7 @@ pub trait ThreadRouter { #[cfg(not(target_family = "wasm"))] pub type ThreadJob = Box; #[cfg(target_family = "wasm")] -pub type ThreadJob = - crate::web_transfer::transfer_closure::TransferClosure; +pub type ThreadJob = crate::web_transfer::transfer_closure::TransferClosure; /// Build a [`ThreadJob`] from a `transfer!`-style closure. On native it boxes /// the closure into `Box`; on web `transfer!` already @@ -73,7 +72,9 @@ macro_rules! thread_job { /// An [`Executor`] that routes calls to per-thread lanes. /// -/// - `execute_normal` / `execute_async`: build the same closure +/// - wasm `Thread::Main` normal/async calls run inline on the caller so app +/// startup does not prewarm a general worker pool for main-thread-safe work. +/// - other `execute_normal` / `execute_async` calls build the same closure /// [`SimpleExecutor`](super::executor::SimpleExecutor) builds, then route it /// to the `thread`'s lane via [`ThreadRouter::execute_on`]. /// - `execute_sync`: run inline on the caller (main thread), exactly like @@ -112,6 +113,26 @@ impl Executor for ThreadExecutor(ret, sender, el2); + })); + + if let Err(error) = thread_result { + handle_non_sync_panic_error::(el, port, error); + } + return; + } + // Identical body to SimpleExecutor::execute_normal (executor.rs), only // the destination differs: route to the thread lane instead of "any // idle worker". @@ -173,6 +194,32 @@ impl Executor for ThreadExecutor(ret, sender, el2); + }) + .catch_unwind() + .await; + + if let Err(err) = async_result { + let err = CatchUnwindWithBacktrace::new(err, PanicBacktrace::take_last()); + handle_non_sync_panic_error::(el, port, err); + } + }); + return; + } + // Off-main async: transfer the `Send` closure (capturing the // transferable `MessagePort`, exactly like the normal branch) to the // thread lane; the lane worker drives the (possibly non-`Send`) @@ -262,7 +309,12 @@ impl ThreadExecuteUtils { // This routes two async tasks to one single-worker lane with a cross-task // dependency: cooperative scheduling completes both; a parking driver would // deadlock (A holds the worker awaiting B, which can never start). -#[cfg(all(test, feature = "rust-async", feature = "thread-pool", not(target_family = "wasm")))] +#[cfg(all( + test, + feature = "rust-async", + feature = "thread-pool", + not(target_family = "wasm") +))] mod single_worker_lane_pinning_tests { use super::*; use crate::codec::dco::DcoCodec; diff --git a/frb_rust/src/handler/mod.rs b/frb_rust/src/handler/mod.rs index ed36f57e918..bc0b88edc50 100644 --- a/frb_rust/src/handler/mod.rs +++ b/frb_rust/src/handler/mod.rs @@ -10,7 +10,7 @@ pub use error_listener::ErrorListener; pub use executor::Executor; pub use handler::{FfiCallMode, TaskInfo, Thread}; pub use handler::{TaskContext, TaskRetFutTrait}; -pub use implementation::thread_executor::{ThreadExecutor, ThreadJob, ThreadRouter}; pub use implementation::error_listener::NoOpErrorListener; pub use implementation::executor::SimpleExecutor; pub use implementation::handler::SimpleHandler; +pub use implementation::thread_executor::{ThreadExecutor, ThreadJob, ThreadRouter}; diff --git a/frb_rust/src/third_party/wasm_bindgen/worker_pool.rs b/frb_rust/src/third_party/wasm_bindgen/worker_pool.rs index 7f44a49ba30..5a86cb1ae59 100644 --- a/frb_rust/src/third_party/wasm_bindgen/worker_pool.rs +++ b/frb_rust/src/third_party/wasm_bindgen/worker_pool.rs @@ -5,7 +5,8 @@ use crate::misc::web_utils::script_path; use crate::web_transfer::transfer_closure::TransferClosure; use js_sys::{Array, Object, Reflect}; -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; +use std::collections::VecDeque; use std::iter::FromIterator; use std::rc::Rc; use wasm_bindgen::prelude::*; @@ -25,7 +26,14 @@ pub struct WorkerPool { } struct PoolState { + /// Idle workers ready to accept a new job. workers: RefCell>, + /// FIFO queue used once all workers up to `max_workers` are busy. + pending: RefCell>>, + /// Number of workers spawned by this pool, idle + busy. + live_workers: Cell, + /// Hard upper bound for workers owned by this pool. + max_workers: usize, callback: Closure, } @@ -37,8 +45,37 @@ impl WorkerPool { worker_js_preamble: Option, wasm_bindgen_name: Option, ) -> Result { - Self::new_raw( - initial.unwrap_or_else(get_wasm_hardware_concurrency), + let initial = initial.unwrap_or_else(get_wasm_hardware_concurrency); + let max_workers = if initial == 0 { + get_wasm_hardware_concurrency() + } else { + initial + }; + Self::new_with_max( + initial, + max_workers, + script_src, + worker_js_preamble, + wasm_bindgen_name, + ) + } + + /// Creates a pool with separate prewarm and hard-cap counts. + /// + /// This is useful for embedders that want a hard one-worker lane but do not + /// want to instantiate the large wasm module on page load. For example, + /// `initial = 0, max_workers = 1` creates no worker until first use, then + /// queues FIFO once that single worker is busy. + pub fn new_with_max( + initial: usize, + max_workers: usize, + script_src: Option, + worker_js_preamble: Option, + wasm_bindgen_name: Option, + ) -> Result { + Self::new_raw_with_max( + initial, + max_workers, script_src.unwrap_or_else(|| script_path().expect("fail to get script path")), worker_js_preamble.unwrap_or_default(), wasm_bindgen_name.unwrap_or_else(|| "wasm_bindgen".to_owned()), @@ -47,9 +84,15 @@ impl WorkerPool { /// Creates a new `WorkerPool` which immediately creates `initial` workers. /// - /// The pool created here can be used over a long period of time, and it - /// will be initially primed with `initial` workers. Currently workers are - /// never released or gc'd until the whole pool is destroyed. + /// `initial` also acts as this pool's hard worker cap, except `initial == 0` + /// means "spawn lazily, capped at `navigator.hardwareConcurrency`". When all + /// workers are busy and the cap has been reached, additional jobs are queued + /// FIFO instead of spawning more web workers. Use [`WorkerPool::new_raw_with_max`] + /// when prewarm and hard-cap counts must differ. + /// + /// The pool created here can be used over a long period of time. Workers are + /// retained for reuse until the whole pool is destroyed, but the pool will + /// never retain more than the hard cap. /// /// # Errors /// @@ -62,10 +105,37 @@ impl WorkerPool { worker_js_preamble: String, wasm_bindgen_name: String, ) -> Result { + let max_workers = if initial == 0 { + get_wasm_hardware_concurrency() + } else { + initial + }; + Self::new_raw_with_max( + initial, + max_workers, + script_src, + worker_js_preamble, + wasm_bindgen_name, + ) + } + + pub fn new_raw_with_max( + initial: usize, + max_workers: usize, + script_src: String, + worker_js_preamble: String, + wasm_bindgen_name: String, + ) -> Result { + let max_workers = max_workers.max(1); + let initial = initial.min(max_workers); + let pool = WorkerPool { script_src, state: Rc::new(PoolState { workers: RefCell::new(Vec::with_capacity(initial)), + pending: RefCell::new(VecDeque::new()), + live_workers: Cell::new(0), + max_workers, callback: Closure::new(|event: Event| { if let Some(event) = event.dyn_ref::() { crate::console_error!("Dropped data:: {:?}", event.data()); @@ -78,14 +148,14 @@ impl WorkerPool { wasm_bindgen_name, }; for _ in 0..initial { - let worker = pool.spawn()?; + let worker = pool.spawn_for_pool()?; pool.state.push(worker); } Ok(pool) } - /// Unconditionally spawns a new worker + /// Unconditionally spawns a new worker. /// /// The worker isn't registered with this `WorkerPool` but is capable of /// executing work for this wasm module. @@ -129,7 +199,9 @@ impl WorkerPool { BlobPropertyBag::new().type_("text/javascript"), )?; let url = Url::create_object_url_with_blob(&blob)?; - let worker: Worker = Worker::new(&url)?; + let worker = Worker::new(&url); + let _ = Url::revoke_object_url(&url); + let worker: Worker = worker?; // With a worker spun up send it the module/memory so it can start // instantiating the wasm module. Later it might receive further @@ -150,81 +222,49 @@ impl WorkerPool { Ok(worker) } - /// Fetches a worker from this pool, spawning one if necessary. - /// - /// This will attempt to pull an already-spawned web worker from our cache - /// if one is available, otherwise it will spawn a new worker and return the - /// newly spawned worker. - /// - /// # Errors - /// - /// Returns any error that may happen while a JS web worker is created and a - /// message is sent to it. - fn worker(&self) -> Result { - match self.state.workers.borrow_mut().pop() { - Some(worker) => Ok(worker), - None => self.spawn(), - } - } - - /// Executes the work `f` in a web worker, spawning a web worker if - /// necessary. - /// - /// This will acquire a web worker and then send the closure `f` to the - /// worker to execute. The worker won't be usable for anything else while - /// `f` is executing, and no callbacks are registered for when the worker - /// finishes. - /// - /// ## Errors - /// - /// Returns any error that may happen while a JS web worker is created and a - /// message is sent to it. - // NOTE: It is originally named `execute`, but rename to align with crate `threadpool` - fn execute_raw(&self, closure: TransferClosure) -> Result { - let worker = self.worker()?; - closure.apply(&worker).map(|_| worker) + fn spawn_for_pool(&self) -> Result { + let worker = self.spawn()?; + self.state + .live_workers + .set(self.state.live_workers.get() + 1); + Ok(worker) } - /// Configures an `onmessage` callback for the `worker` specified for the - /// web worker to be reclaimed and re-inserted into this pool when a message - /// is received. - /// - /// Currently this `WorkerPool` abstraction is intended to execute one-off - /// style work where the work itself doesn't send any notifications and - /// when it's done the worker is ready to execute more work. This method is - /// used for all spawned workers to ensure that when the work is finished - /// the worker is reclaimed back into this pool. - fn reclaim_on_message(&self, worker: &Worker) { - let state = Rc::downgrade(&self.state); - let worker2 = worker.clone(); - let reclaim_slot = Rc::new(RefCell::new(None)); - let slot2 = reclaim_slot.clone(); - let reclaim = Closure::::new(move |_: MessageEvent| { - if let Some(state) = state.upgrade() { - state.push(worker2.clone()); - } - *slot2.borrow_mut() = None; - }); - worker.set_onmessage(Some(reclaim.as_ref().unchecked_ref())); - *reclaim_slot.borrow_mut() = Some(reclaim); + /// Dispatches work to an idle/new worker and arranges for it to be reclaimed + /// on completion. + fn dispatch_to_worker( + &self, + worker: Worker, + closure: TransferClosure, + ) -> Result<(), JsValue> { + PoolState::dispatch(&self.state, worker.clone(), closure).map_err(|err| { + // Preserve the worker if the post failed synchronously. This mirrors + // native thread pools: the failed job is reported to the caller, but + // the pool itself remains usable for later jobs. + self.state.push(worker); + err + }) } } impl WorkerPool { /// Executes `f` in a web worker. /// - /// This pool manages a set of web workers to draw from, and `f` will be - /// spawned quickly into one if the worker is idle. If no idle workers are - /// available then a new web worker will be spawned. + /// This pool manages a capped set of web workers. `f` will be spawned + /// quickly into an idle worker if one is available. If no idle worker is + /// available and the hard cap has not been reached, a new worker is spawned. + /// If the cap has been reached, `f` is queued FIFO until a worker completes + /// its current job. /// - /// Once `f` returns the worker assigned to `f` is automatically reclaimed - /// by this `WorkerPool`. This method provides no method of learning when - /// `f` completes, and for that you'll need to use `run_notify`. + /// Once `f` returns the worker assigned to `f` is automatically reclaimed by + /// this `WorkerPool`. This method provides no method of learning when `f` + /// completes, and for that you'll need to use `run_notify`. /// /// ## Errors /// /// If an error happens while spawning a web worker or sending a message to - /// a web worker, that error is returned. + /// a web worker immediately, that error is returned. Queued jobs return + /// `Ok(())` once enqueued. /// /// ## Transferrables /// Items put inside `transfer` will have their ownership transferred from @@ -237,13 +277,58 @@ impl WorkerPool { /// a `post_message`. Examples are `Buffer`s, `MessagePort`s, etc... // NOTE: It is originally named `run`, but rename to align with crate `threadpool` pub fn execute(&self, closure: TransferClosure) -> Result<(), JsValue> { - let worker = self.execute_raw(closure)?; - self.reclaim_on_message(&worker); + let idle_worker = { self.state.workers.borrow_mut().pop() }; + if let Some(worker) = idle_worker { + return self.dispatch_to_worker(worker, closure); + } + + if self.state.live_workers.get() < self.state.max_workers { + let worker = self.spawn_for_pool()?; + return self.dispatch_to_worker(worker, closure); + } + + self.state.pending.borrow_mut().push_back(closure); Ok(()) } } impl PoolState { + fn dispatch( + state: &Rc, + worker: Worker, + closure: TransferClosure, + ) -> Result<(), JsValue> { + let weak_state = Rc::downgrade(state); + let worker2 = worker.clone(); + let reclaim_slot = Rc::new(RefCell::new(None)); + let slot2 = reclaim_slot.clone(); + let reclaim = Closure::::new(move |_: MessageEvent| { + if let Some(state) = weak_state.upgrade() { + Self::complete(&state, worker2.clone()); + } + *slot2.borrow_mut() = None; + }); + worker.set_onmessage(Some(reclaim.as_ref().unchecked_ref())); + *reclaim_slot.borrow_mut() = Some(reclaim); + + closure.apply(&worker).map_err(|err| { + *reclaim_slot.borrow_mut() = None; + err + }) + } + + fn complete(state: &Rc, worker: Worker) { + let next = { state.pending.borrow_mut().pop_front() }; + if let Some(closure) = next { + if let Err(err) = Self::dispatch(state, worker.clone(), closure) { + crate::console_error!("Failed to dispatch queued worker job: {:?}", err); + state.push(worker); + } + } else { + state.push(worker); + } + } + fn push(&self, worker: Worker) { worker.set_onmessage(Some(self.callback.as_ref().unchecked_ref())); worker.set_onerror(Some(self.callback.as_ref().unchecked_ref())); @@ -260,9 +345,9 @@ impl PoolState { impl Default for WorkerPool { fn default() -> Self { // Use `Some(0)` instead of `None` so workers are spawned lazily on first - // FFI dispatch rather than eagerly at module init (sized by - // `navigator.hardwareConcurrency`). This avoids spawning N web workers - // on cold start when the app may not need any. + // FFI dispatch rather than eagerly at module init. The hard cap is still + // `navigator.hardwareConcurrency`, so a lazy default can grow under load + // but cannot burst beyond the browser-reported CPU parallelism. Self::new(Some(0), None, None, None).expect("fail to create WorkerPool") } } @@ -275,5 +360,5 @@ fn get_wasm_hardware_concurrency() -> usize { let navigator = js_sys::Reflect::get(global, &key).unwrap(); key = wasm_bindgen::JsValue::from_str("hardwareConcurrency"); let hardware_concurrency = js_sys::Reflect::get(&navigator, &key).unwrap(); - hardware_concurrency.as_f64().unwrap() as usize + (hardware_concurrency.as_f64().unwrap() as usize).max(1) }