From 2a962e0400a4764c4de7502b7d42c173431ba93c Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 5 Jan 2026 12:24:37 +1100 Subject: [PATCH 1/6] Initial --- crates/ion/src/env.rs | 18 -- crates/ion/src/js_context.rs | 45 +++-- crates/ion/src/js_runtime.rs | 7 +- crates/ion/src/js_worker.rs | 58 ++++-- crates/ion/src/platform/callback_registry.rs | 100 ++++++++++ crates/ion/src/platform/mod.rs | 1 + crates/ion/src/platform/platform.rs | 4 + crates/ion/src/platform/realm.rs | 12 +- crates/ion/src/platform/worker.rs | 185 +++++++++++-------- 9 files changed, 290 insertions(+), 140 deletions(-) create mode 100644 crates/ion/src/platform/callback_registry.rs diff --git a/crates/ion/src/env.rs b/crates/ion/src/env.rs index b74778b..735cf90 100644 --- a/crates/ion/src/env.rs +++ b/crates/ion/src/env.rs @@ -1,6 +1,4 @@ -use std::cell::RefCell; use std::future::Future; -use std::rc::Rc; use std::sync::Arc; use flume::Sender; @@ -27,7 +25,6 @@ pub struct Env { pub(crate) context: sys::GlobalContext, pub(crate) background_task_manager: Arc, pub(crate) global_refs: RefCounter, - pub(crate) shutdown_requested: Rc>, pub(crate) tx: Sender, pub(crate) finalizer_registry: FinalizerRegistery, pub(crate) global_this: sys::GlobalThis, @@ -40,7 +37,6 @@ impl Env { context: sys::GlobalContext, background_task_manager: Arc, global_refs: RefCounter, - shutdown_requested: Rc>, tx: Sender, finalizer_registry: FinalizerRegistery, global_this: sys::GlobalThis, @@ -53,7 +49,6 @@ impl Env { background_task_manager, inner: std::ptr::null_mut(), global_refs, - shutdown_requested, finalizer_registry, tx, }); @@ -85,19 +80,6 @@ impl Env { pub fn dec_ref(&self) { self.global_refs.dec(); - let shutdown_requested = { - let shutdown_requested = self.shutdown_requested.borrow(); - *shutdown_requested - }; - - if self.global_refs.count() == 0 && shutdown_requested { - self.tx - .try_send(JsWorkerEvent::RequestContextShutdown { - id: self.realm_id, - resolve: None, - }) - .unwrap(); - } } pub fn ref_count(&self) -> usize { diff --git a/crates/ion/src/js_context.rs b/crates/ion/src/js_context.rs index 5947dc4..be70bef 100644 --- a/crates/ion/src/js_context.rs +++ b/crates/ion/src/js_context.rs @@ -1,15 +1,19 @@ +use std::sync::Arc; + use flume::Sender; use flume::bounded; use crate::Env; use crate::Error; use crate::JsUnknown; +use crate::platform::callback_registry::CallbackRegistry; use crate::platform::worker::JsWorkerEvent; use crate::utils::channel::oneshot; /// This is a handle to a v8::Context #[derive(Debug, Clone)] pub struct JsContext { + pub(crate) callback_registry: Arc, pub(crate) id: usize, pub(crate) tx: Sender, } @@ -83,25 +87,40 @@ impl JsContext { let specifier = specifier.as_ref().to_string(); self.exec_blocking(move |env| env.import(specifier)) } -} -impl Drop for JsContext { - fn drop(&mut self) { + /// Wait for the context to complete all activity + pub fn join_blocking(self) -> crate::Result<()> { + if !self.callback_registry.worker_handle_active() { + return Ok(()); + } + + self.callback_registry + .context_handle_set_status(&self.id, false); + let (tx, rx) = oneshot(); + self.callback_registry + .add_context_shutdown_callback(self.id.clone(), move || tx.try_send(()).unwrap()); - if self - .tx - .send(JsWorkerEvent::RequestContextShutdown { - id: self.id, - resolve: Some(tx), - }) - .is_err() - { - panic!("Cannot drop JsContext 1") - }; + drop(self.tx.send(JsWorkerEvent::ContextHandleDeactivated { + id: self.id.clone(), + })); if rx.recv().is_err() { panic!("Cannot drop JsContext 2") } + + Ok(()) + } + + /// Wait for the context to complete all activity + pub async fn join_async(&self) -> crate::Result<()> { + self.callback_registry + .context_handle_set_status(&self.id, false); + self.tx + .send(JsWorkerEvent::ContextHandleDropped { + id: self.id.clone(), + }) + .unwrap(); + Ok(()) } } diff --git a/crates/ion/src/js_runtime.rs b/crates/ion/src/js_runtime.rs index 25b1b64..628f496 100644 --- a/crates/ion/src/js_runtime.rs +++ b/crates/ion/src/js_runtime.rs @@ -12,6 +12,7 @@ use crate::JsResolver; use crate::JsTransformer; use crate::JsWorker; use crate::JsWorkerOptions; +use crate::platform::callback_registry::CallbackRegistry; use crate::platform::platform::HAS_INIT; use crate::platform::platform::PLATFORM; @@ -173,12 +174,14 @@ impl JsRuntime { pub fn spawn_worker( &self, options: JsWorkerOptions, - ) -> crate::Result> { + ) -> crate::Result { let (tx, rx) = bounded(1); + let callback_registry = Arc::new(CallbackRegistry::default()); if self .tx .send(PlatformEvent::SpawnWorker { + callback_registry: Arc::clone(&callback_registry), extensions: options.extensions, transformers: options.transformers, resolvers: options.resolvers, @@ -193,7 +196,7 @@ impl JsRuntime { return Err(Error::WorkerInitializeError); }; - Ok(Arc::new(JsWorker::new(tx, handle))) + Ok(JsWorker::new(callback_registry, tx, Arc::new(handle))) } } diff --git a/crates/ion/src/js_worker.rs b/crates/ion/src/js_worker.rs index 7272b76..d162d16 100644 --- a/crates/ion/src/js_worker.rs +++ b/crates/ion/src/js_worker.rs @@ -10,6 +10,7 @@ use crate::Error; use crate::JsExtension; use crate::JsResolver; use crate::JsTransformer; +use crate::platform::callback_registry::CallbackRegistry; use crate::platform::worker::JsWorkerEvent; use crate::utils::channel::oneshot; @@ -30,20 +31,26 @@ pub struct JsWorkerOptions { /// to be used to execute JavaScript #[derive(Debug)] pub struct JsWorker { + callback_registry: Arc, tx: Sender, - handle: Mutex>>>, + handle: Arc>>>>, } impl JsWorker { pub(crate) fn new( + callback_registry: Arc, tx: Sender, - handle: Mutex>>>, + handle: Arc>>>>, ) -> Self { - JsWorker { tx, handle } + JsWorker { + tx, + handle, + callback_registry, + } } /// Create a handle to a v8::Context associated with this v8::Isolate - pub fn create_context(&self) -> crate::Result> { + pub fn create_context(&self) -> crate::Result { let (tx, rx) = bounded(1); if self @@ -58,7 +65,13 @@ impl JsWorker { return Err(Error::WorkerInitializeError); }; - Ok(Arc::new(JsContext { id, tx })) + self.callback_registry.context_handle_set_status(&id, true); + + Ok(JsContext { + id, + tx, + callback_registry: Arc::clone(&self.callback_registry), + }) } pub fn run_garbage_collection_for_testing(&self) -> crate::Result<()> { @@ -74,19 +87,20 @@ impl JsWorker { Ok(rx.recv()?) } -} -impl Drop for JsWorker { - fn drop(&mut self) { + /// Wait for all of the contexts within the worker to complete all activity + pub fn join_blocking(self) -> crate::Result<()> { + self.callback_registry.worker_handle_deactivate(); + let (tx, rx) = oneshot(); + self.callback_registry + .add_worker_shutdown_callback(move || { + tx.send(()).unwrap(); + }); - if self - .tx - .send(JsWorkerEvent::RequestShutdown { resolve: tx }) - .is_err() - { - panic!("Cannot drop JsWorker 1"); - }; + self.tx + .send(JsWorkerEvent::WorkerHandleDeactivated) + .unwrap(); if rx.recv().is_err() { panic!("Cannot drop JsWorker 2"); @@ -99,5 +113,19 @@ impl Drop for JsWorker { if let Some(handle) = handle.take() { drop(handle.join().unwrap()); } + + Ok(()) + } + + /// Wait for all of the contexts within the worker to complete all activity + pub async fn join_async(self) -> crate::Result<()> { + Ok(()) + } +} + +impl Drop for JsWorker { + fn drop(&mut self) { + self.callback_registry.worker_handle_deactivate(); + drop(self.tx.try_send(JsWorkerEvent::WorkerHandleDropped)); } } diff --git a/crates/ion/src/platform/callback_registry.rs b/crates/ion/src/platform/callback_registry.rs new file mode 100644 index 0000000..5e29dd9 --- /dev/null +++ b/crates/ion/src/platform/callback_registry.rs @@ -0,0 +1,100 @@ +use std::collections::HashMap; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; + +use parking_lot::Mutex; +use parking_lot::RwLock; + +pub type WorkerShutdownCallback = Box; +pub type ContextShutdownCallback = Box; + +pub struct CallbackRegistry { + pub worker_handle_active: AtomicBool, + pub context_handle_active: RwLock>, + pub worker_shutdown: Mutex>, + pub context_shutdown: Mutex>>, +} + +impl Default for CallbackRegistry { + fn default() -> Self { + Self { + worker_handle_active: AtomicBool::new(true), + context_handle_active: Default::default(), + worker_shutdown: Default::default(), + context_shutdown: Default::default(), + } + } +} + +impl CallbackRegistry { + pub(crate) fn worker_handle_active(&self) -> bool { + self.worker_handle_active.load(Ordering::Relaxed) + } + + pub(crate) fn worker_handle_deactivate(&self) { + self.worker_handle_active.swap(false, Ordering::Relaxed); + } + + pub(crate) fn context_handle_active( + &self, + id: &usize, + ) -> bool { + self.context_handle_active + .read() + .get(id) + .unwrap_or(&false) + .clone() + } + + pub(crate) fn context_handle_set_status( + &self, + id: &usize, + status: bool, + ) { + self.context_handle_active + .write() + .insert(id.clone(), status); + } + + pub(crate) fn add_worker_shutdown_callback( + &self, + callback: impl 'static + Send + Sync + FnOnce(), + ) { + self.worker_shutdown.lock().push(Box::new(callback)); + } + + pub(crate) fn add_context_shutdown_callback( + &self, + id: usize, + callback: impl 'static + Send + Sync + FnOnce(), + ) { + self.context_shutdown + .lock() + .entry(id) + .or_default() + .push(Box::new(callback)); + } + + pub(crate) fn take_worker_shutdown_callbacks(&self) -> Vec { + std::mem::take(&mut *self.worker_shutdown.lock()) + } + + pub(crate) fn take_context_shutdown_callbacks( + &self, + id: usize, + ) -> Vec { + std::mem::take(&mut *self.context_shutdown.lock().entry(id).or_default()) + } +} + +impl std::fmt::Debug for CallbackRegistry { + fn fmt( + &self, + f: &mut std::fmt::Formatter<'_>, + ) -> std::fmt::Result { + f.debug_struct("CallbackRegistry") + .field("worker_shutdown", &self.worker_shutdown.lock().len()) + .field("context_shutdown", &self.context_shutdown.lock().len()) + .finish() + } +} diff --git a/crates/ion/src/platform/mod.rs b/crates/ion/src/platform/mod.rs index e0088a9..1833fb8 100644 --- a/crates/ion/src/platform/mod.rs +++ b/crates/ion/src/platform/mod.rs @@ -1,5 +1,6 @@ #![allow(clippy::module_inception)] pub mod background_worker; +pub(crate) mod callback_registry; pub(crate) mod extension; pub(crate) mod finalizer_registry; pub mod module; diff --git a/crates/ion/src/platform/platform.rs b/crates/ion/src/platform/platform.rs index 347e598..9b5e5af 100644 --- a/crates/ion/src/platform/platform.rs +++ b/crates/ion/src/platform/platform.rs @@ -14,6 +14,7 @@ use crate::JsExtension; use crate::JsResolver; use crate::JsTransformer; use crate::platform::background_worker::BackgroundTaskManager; +use crate::platform::callback_registry::CallbackRegistry; use crate::platform::worker::JsWorkerEvent; use crate::platform::worker::start_js_worker_thread; @@ -26,6 +27,7 @@ pub(crate) enum PlatformEvent { transformers: Vec, }, SpawnWorker { + callback_registry: Arc, extensions: Vec, resolvers: Vec, transformers: Vec, @@ -96,6 +98,7 @@ pub(crate) static PLATFORM: LazyLock> = LazyLock::new(|| { } } PlatformEvent::SpawnWorker { + callback_registry, resolve, extensions: init_extensions, resolvers: init_resolvers, @@ -117,6 +120,7 @@ pub(crate) static PLATFORM: LazyLock> = LazyLock::new(|| { } let (tx, handle) = start_js_worker_thread( + callback_registry, background_task_manager.clone(), worker_extensions, worker_resolvers, diff --git a/crates/ion/src/platform/realm.rs b/crates/ion/src/platform/realm.rs index 8b51201..6c154c7 100644 --- a/crates/ion/src/platform/realm.rs +++ b/crates/ion/src/platform/realm.rs @@ -1,6 +1,4 @@ -use std::cell::RefCell; use std::collections::HashMap; -use std::rc::Rc; use std::sync::Arc; use flume::Sender; @@ -31,9 +29,7 @@ pub struct JsRealm { /// Used to tell the Worker if there are any long-lived async tasks /// that should prevent the context from being shutdown pub(crate) global_refs: RefCounter, - pub(crate) shutdown_requested: Rc>, pub(crate) modules: ModuleMap, - pub(crate) tx: Sender, pub(crate) global_this: sys::GlobalThis, } @@ -50,7 +46,6 @@ impl JsRealm { let global_this = sys::GlobalThis::new(&context); let global_refs = RefCounter::new(0); - let shutdown_requested = Rc::new(RefCell::new(false)); let finalizer_registry = FinalizerRegistery::new(isolate); // TODO make these RefCells @@ -61,7 +56,6 @@ impl JsRealm { context.clone(), Arc::clone(&background_task_manager), global_refs.clone(), - Rc::clone(&shutdown_requested), tx.clone(), finalizer_registry.clone(), global_this.clone(), @@ -76,10 +70,8 @@ impl JsRealm { resolvers, transformers, global_refs, - shutdown_requested, finalizer_registry, global_this, - tx, }); let realm_ptr = realm.as_mut() as *mut JsRealm; @@ -117,13 +109,11 @@ impl JsRealm { &self, fut: impl 'static + Send + Sync + Future>, ) -> crate::Result<()> { - let tx = self.tx.clone(); - let id = self.id; self.background_task_manager.spawn(async move { if let Err(_error) = fut.await { todo!("Missing global error handler") }; - Ok(tx.try_send(JsWorkerEvent::BackgroundTaskComplete { id })?) + Ok(()) }) } diff --git a/crates/ion/src/platform/worker.rs b/crates/ion/src/platform/worker.rs index 5f8a67a..1b04074 100644 --- a/crates/ion/src/platform/worker.rs +++ b/crates/ion/src/platform/worker.rs @@ -18,6 +18,7 @@ use crate::JsResolver; use crate::JsTransformer; use crate::fs::FileSystem; use crate::platform::background_worker::BackgroundTaskManager; +use crate::platform::callback_registry::CallbackRegistry; use crate::utils::HashMapExt; use crate::utils::PathExt; @@ -25,13 +26,6 @@ pub(crate) enum JsWorkerEvent { CreateContext { resolve: Sender<(usize, Sender)>, }, - BackgroundTaskComplete { - id: usize, - }, - RequestContextShutdown { - resolve: Option>, - id: usize, - }, Exec { id: usize, #[allow(clippy::type_complexity)] @@ -42,17 +36,27 @@ pub(crate) enum JsWorkerEvent { id: usize, specifier: String, }, - RequestShutdown { - resolve: Sender<()>, + TryShutdownContext { + id: usize, + force: bool, }, RunGarbageCollectionForTesting { resolve: Sender<()>, }, + WorkerHandleDropped, + WorkerHandleDeactivated, + ContextHandleDropped { + id: usize, + }, + ContextHandleDeactivated { + id: usize, + }, } // Create a dedicated thread to host the isolate #[allow(clippy::type_complexity)] pub(crate) fn start_js_worker_thread( + callback_registry: Arc, background_task_manager: Arc, extensions: Vec>, resolvers: Vec, @@ -68,6 +72,7 @@ pub(crate) fn start_js_worker_thread( let tx: Sender = tx.clone(); move || { worker_thread( + callback_registry, tx, rx, background_task_manager, @@ -82,6 +87,7 @@ pub(crate) fn start_js_worker_thread( } fn worker_thread( + callback_registry: Arc, tx: Sender, rx: Receiver, background_task_manager: Arc, @@ -98,13 +104,13 @@ fn worker_thread( // Maintain a store of Global to help with cleanup on shutdown. let mut realms = HashMap::>::new(); - // Cleanup hooks - let mut shutdown_context_senders = HashMap::>>::new(); - let mut shutdown_senders = Vec::>::new(); - let mut shutdown_requested = false; - while let Ok(event) = rx.recv() { - // println!("{:?} {:?}", active_context, event); + // eprintln!("{:?}", event); + + if realms.len() == 0 && !callback_registry.worker_handle_active() { + break; + } + match event { JsWorkerEvent::CreateContext { resolve } => { let realm = JsRealm::new( @@ -122,45 +128,6 @@ fn worker_thread( realms.insert(realm_id, realm); resolve.try_send((realm_id, tx.clone()))?; } - JsWorkerEvent::RequestContextShutdown { id, resolve } => { - // Store shutdown resolvers for when the context is closed - if let Some(resolve) = resolve { - shutdown_context_senders - .entry(id) - .or_default() - .push(resolve); - } - - // If there are async tasks pending then wait for them to complete - { - let realm = realms.try_get_mut(&id)?; - let mut realm_shutdown_requested = realm.shutdown_requested.borrow_mut(); - (*realm_shutdown_requested) = true; - if realm.global_refs.count() != 0 { - continue; - } - }; - - // If there are no async tasks then shutdown the context - let Some(realm) = realms.remove(&id) else { - continue; - }; - - let finalizer_registry = realm.finalizer_registry; - finalizer_registry.clear(); - drop(finalizer_registry); - - for resolver in shutdown_context_senders.remove(&id).unwrap_or_default() { - let _ = resolver.try_send(()); - } - - if shutdown_requested && realms.is_empty() { - for sender in shutdown_senders { - let _ = sender.try_send(()); - } - break; - } - } JsWorkerEvent::Exec { id, callback, span } => { let realm = realms.try_get(&id)?; @@ -169,13 +136,9 @@ fn worker_thread( // TODO global error handler panic!("Callback errored {:?}", err) }; - } - JsWorkerEvent::BackgroundTaskComplete { id } => { - let realm = realms.try_get(&id)?; - let realm_shutdown_requested = realm.shutdown_requested.borrow(); - if *realm_shutdown_requested && realm.global_refs.count() == 0 { - tx.try_send(JsWorkerEvent::RequestContextShutdown { id, resolve: None })?; - } + + tx.try_send(JsWorkerEvent::TryShutdownContext { id, force: false }) + .unwrap(); } JsWorkerEvent::Import { id, specifier } => { Module::v8_initialize( @@ -185,45 +148,105 @@ fn worker_thread( std::env::current_dir()?.try_to_string()?, )?; } - JsWorkerEvent::RequestShutdown { resolve } => { - shutdown_senders.push(resolve); - shutdown_requested = true; - if !realms.is_empty() { - continue; - } + JsWorkerEvent::RunGarbageCollectionForTesting { resolve } => { + isolate.request_garbage_collection_for_testing(v8::GarbageCollectionType::Full); + resolve.try_send(())?; + } + JsWorkerEvent::WorkerHandleDropped => { + for id in realms.keys().cloned().collect::>() { + let Some(realm) = realms.remove(&id) else { + continue; + }; - for sender in shutdown_senders { - let _ = sender.try_send(()); + let finalizer_registry = realm.finalizer_registry; + finalizer_registry.clear(); + drop(finalizer_registry); + + for shutdown_callback in callback_registry + .take_context_shutdown_callbacks(id.clone()) + .into_iter() + { + shutdown_callback(); + } } break; } - JsWorkerEvent::RunGarbageCollectionForTesting { resolve } => { - isolate.request_garbage_collection_for_testing(v8::GarbageCollectionType::Full); - resolve.try_send(())?; + JsWorkerEvent::WorkerHandleDeactivated => { + for id in realms.keys() { + tx.try_send(JsWorkerEvent::TryShutdownContext { + id: id.clone(), + force: true, + }) + .unwrap(); + } + } + JsWorkerEvent::ContextHandleDeactivated { id } => { + tx.try_send(JsWorkerEvent::TryShutdownContext { + id: id.clone(), + force: false, + }) + .unwrap(); + } + JsWorkerEvent::ContextHandleDropped { id } => { + tx.try_send(JsWorkerEvent::TryShutdownContext { + id: id.clone(), + force: true, + }) + .unwrap(); + } + JsWorkerEvent::TryShutdownContext { id, force } => { + // If there are async tasks pending then wait for them to complete + if !force && realms.try_get_mut(&id)?.global_refs.count() != 0 { + continue; + } + + // If there are no async tasks then shutdown the context + let Some(realm) = realms.remove(&id) else { + continue; + }; + + let finalizer_registry = realm.finalizer_registry; + finalizer_registry.clear(); + drop(finalizer_registry); + + for shutdown_callback in callback_registry + .take_context_shutdown_callbacks(id.clone()) + .into_iter() + { + shutdown_callback(); + } } } } + for shutdown_callback in callback_registry + .take_worker_shutdown_callbacks() + .into_iter() + { + shutdown_callback(); + } + Ok(()) } #[allow(unused)] +#[rustfmt::skip] impl std::fmt::Debug for JsWorkerEvent { fn fmt( &self, f: &mut std::fmt::Formatter<'_>, ) -> std::fmt::Result { match self { - Self::CreateContext { resolve } => write!(f, "CreateContext"), - Self::BackgroundTaskComplete { id } => write!(f, "BackgroundTaskComplete"), - Self::RequestContextShutdown { id, resolve } => write!(f, "RequestContextShutdown"), - Self::Exec { id, callback, span } => write!(f, "Exec"), - Self::Import { id, specifier } => write!(f, "Import"), - Self::RequestShutdown { resolve } => write!(f, "RequestShutdown"), - Self::RunGarbageCollectionForTesting { resolve } => { - write!(f, "RunGarbageCollectionForTesting") - } + Self::CreateContext { resolve } => write!(f, "CreateContext"), + Self::Exec { id, callback, span } => write!(f, "Exec [id={}]", id), + Self::Import { id, specifier } => write!(f, "Import"), + Self::TryShutdownContext { id, force } => write!(f, "TryShutdownContext [id={} force={}]", id, force), + Self::WorkerHandleDropped => write!(f, "WorkerHandleDropped"), + Self::WorkerHandleDeactivated => write!(f, "WorkerHandleDeactivated"), + Self::ContextHandleDropped { id } => write!(f, "ContextHandleDropped"), + Self::ContextHandleDeactivated { id } => write!(f, "ContextHandleDeactivated [id={}]", id), + Self::RunGarbageCollectionForTesting { resolve } => write!(f, "RunGarbageCollectionForTesting"), } } } From 291c1d7171d2920fcb51b8497186592c3dca0ec4 Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 5 Jan 2026 13:34:38 +1100 Subject: [PATCH 2/6] tests for join --- crates/ion/src/platform/callback_registry.rs | 1 + examples/src/_utils/mod.rs | 1 + examples/src/_utils/thread_id.rs | 12 ++ examples/src/basic_join/basic_join.test.ts | 215 ++++++++++++++++++ examples/src/basic_join/mod.rs | 216 +++++++++++++++++++ examples/src/main.rs | 8 +- examples/test-utils/run_test.ts | 7 +- 7 files changed, 453 insertions(+), 7 deletions(-) create mode 100644 examples/src/_utils/thread_id.rs create mode 100644 examples/src/basic_join/basic_join.test.ts create mode 100644 examples/src/basic_join/mod.rs diff --git a/crates/ion/src/platform/callback_registry.rs b/crates/ion/src/platform/callback_registry.rs index 5e29dd9..aa6994e 100644 --- a/crates/ion/src/platform/callback_registry.rs +++ b/crates/ion/src/platform/callback_registry.rs @@ -35,6 +35,7 @@ impl CallbackRegistry { self.worker_handle_active.swap(false, Ordering::Relaxed); } + #[allow(dead_code)] pub(crate) fn context_handle_active( &self, id: &usize, diff --git a/examples/src/_utils/mod.rs b/examples/src/_utils/mod.rs index b42be1c..ed514d2 100644 --- a/examples/src/_utils/mod.rs +++ b/examples/src/_utils/mod.rs @@ -1,2 +1,3 @@ pub mod memory_blob; pub mod memory_usage; +pub mod thread_id; diff --git a/examples/src/_utils/thread_id.rs b/examples/src/_utils/thread_id.rs new file mode 100644 index 0000000..9aca507 --- /dev/null +++ b/examples/src/_utils/thread_id.rs @@ -0,0 +1,12 @@ +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +static NEXT_THREAD_ID: AtomicUsize = AtomicUsize::new(1); + +thread_local! { + static THREAD_ID: usize = NEXT_THREAD_ID.fetch_add(1, Ordering::Relaxed); +} + +pub fn thread_id() -> usize { + THREAD_ID.with(|&id| id) +} diff --git a/examples/src/basic_join/basic_join.test.ts b/examples/src/basic_join/basic_join.test.ts new file mode 100644 index 0000000..802c653 --- /dev/null +++ b/examples/src/basic_join/basic_join.test.ts @@ -0,0 +1,215 @@ +import { executeExample } from "../../test-utils/run_test.ts"; +import { assert, assertEquals, assertObjectMatch } from "jsr:@std/assert@^1"; + +type Record = { + thread: number; + message: string; + js_context?: number; + event_loop?: boolean; +}; + +async function executeBasicJoin(caseName: string): Promise> { + const result = await executeExample("basic_join", [caseName]); + return result.split("\n").map((record) => JSON.parse(record)); +} + +function assertArraysMatch(arr1: any, arr2: any, msg?: string): void { + return assertObjectMatch({ arr: arr1 }, { arr: arr2 }, msg); +} + +function filterJs(ctx: number, event_loop: boolean = false) { + return (r: Record): boolean => + r.js_context === ctx && !!r.event_loop == event_loop; +} + + +Deno.test("should_cancel_when_dropped", async () => { + const example = "should_cancel_when_dropped"; + const results = await executeBasicJoin(example); + + // The code on the main thread will always run + assertEquals(results.filter((r) => r.thread === 1).length, 2); + + // The code on the JavaScript thread may or may not run + assert( + results.filter((r) => r.js_context === 0 && !r.event_loop).length === + 2 || + results.filter((r) => r.js_context === 0 && !r.event_loop) + .length === 0 + ); + + // The code on the Event Loop should not run + assertEquals(results.filter((r) => r.event_loop).length, 0); +}); + +Deno.test("should_cancel_when_dropped_multiple", async () => { + const example = "should_cancel_when_dropped_multiple"; + const results = await executeBasicJoin(example); + + // The code on the main thread will always run + assertEquals(results.filter((r) => r.thread === 1).length, 2); + + // The code on the JavaScript thread may or may not run + assert( + results.filter((r) => r.js_context === 0 && !r.event_loop).length === + 4 || + results.filter((r) => r.js_context === 0 && !r.event_loop) + .length === 0 + ); + + // The code on the Event Loop should not run + assertEquals(results.filter((r) => r.event_loop).length, 0); +}); + +Deno.test("should_cancel_blocking_when_dropped", async () => { + const example = "should_cancel_blocking_when_dropped"; + const results = await executeBasicJoin(example); + + // The code on the main thread will always run + assertEquals(results.filter((r) => r.thread === 1).length, 2); + + // The code on the JavaScript thread must run + assertEquals( + results.filter((r) => r.js_context === 0 && !r.event_loop).length, + 2 + ); + + // The code on the Event Loop may or may not run, but not progress + assert( + results.filter( + (r) => r.js_context === 0 && r.event_loop && r.message !== "end" + ).length === 1 || + results.filter((r) => r.js_context === 0 && r.event_loop).length === + 0 + ); +}); + +Deno.test("should_cancel_blocking_when_dropped_multiple", async () => { + const example = "should_cancel_blocking_when_dropped_multiple"; + const results = await executeBasicJoin(example); + + // The code on the main thread will always run + assertEquals(results.filter((r) => r.thread === 1).length, 2); + + // The code on the JavaScript thread must run + assertEquals( + results.filter((r) => r.js_context === 0 && !r.event_loop).length, + 4 + ); + + // The code on the Event Loop may or may not run, but not progress + assert( + results.filter( + (r) => r.js_context === 0 && r.event_loop && r.message !== "end" + ).length === 2 || + results.filter((r) => r.js_context === 0 && r.event_loop).length === + 0 + ); +}); + +Deno.test("should_wait_for_code_to_finish", async () => { + const example = "should_wait_for_code_to_finish"; + const results = await executeBasicJoin(example); + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); + +// Hangs +Deno.test.only("should_wait_for_code_to_finish_multiple", async () => { + const example = "should_wait_for_code_to_finish_multiple"; + const results = await executeBasicJoin(example); + + console.log(results); + + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); + +Deno.test("should_wait_for_code_to_finish_blocking", async () => { + const example = "should_wait_for_code_to_finish_blocking"; + const results = await executeBasicJoin(example); + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); + +// Does not complete context +Deno.test.ignore("should_wait_for_code_to_finish_worker", async () => { + const example = "should_wait_for_code_to_finish_worker"; + const results = await executeBasicJoin(example); + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); + +// Does not complete context +Deno.test.ignore("should_wait_for_code_to_finish_worker_blocking", async () => { + const example = "should_wait_for_code_to_finish_worker_blocking"; + const results = await executeBasicJoin(example); + + console.log(results); + + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); + +Deno.test("should_wait_for_code_to_finish_context", async () => { + const example = "should_wait_for_code_to_finish_context"; + const results = await executeBasicJoin(example); + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); + +Deno.test("should_wait_for_code_to_finish_context_blocking", async () => { + const example = "should_wait_for_code_to_finish_context_blocking"; + const results = await executeBasicJoin(example); + assertArraysMatch(results, [ + { thread: 1, message: "start" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 1000, js_context: 0, event_loop: true, message: "start" }, + { thread: 1000, js_context: 0, event_loop: true, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 1, message: "end" }, + ]); +}); diff --git a/examples/src/basic_join/mod.rs b/examples/src/basic_join/mod.rs new file mode 100644 index 0000000..c4c3067 --- /dev/null +++ b/examples/src/basic_join/mod.rs @@ -0,0 +1,216 @@ +use std::sync::Arc; +use std::time::Duration; + +use ion::*; +use serde::Serialize; + +use crate::_utils::thread_id; + +#[derive(Serialize)] +struct Report { + thread: usize, + #[serde(skip_serializing_if = "Option::is_none")] + js_context: Option, + #[serde(skip_serializing_if = "Option::is_none")] + event_loop: Option, + message: String, +} + +impl Report { + fn print( + js_context: Option, + event_loop: Option, + message: &str, + ) { + let thread = if event_loop.is_none() { + thread_id::thread_id() + } else { + 1000 + }; + println!( + "{}", + serde_json::to_string(&Report { + thread, + js_context: js_context, + event_loop: event_loop, + message: message.to_string(), + }) + .unwrap() + ) + } +} + +pub fn main() -> anyhow::Result<()> { + let case = std::env::args() + .collect::>() + .get(2) + .cloned() + .expect("No code provided"); + + let runtime = JsRuntime::initialize_once(JsRuntimeOptions::default())?; + + Report::print(None, None, "start"); + #[rustfmt::skip] + match case.as_str() { + "should_cancel_when_dropped" => should_cancel_when_dropped(runtime), + "should_cancel_when_dropped_multiple" => should_cancel_when_dropped_multiple(runtime), + "should_cancel_blocking_when_dropped" => should_cancel_blocking_when_dropped(runtime), + "should_cancel_blocking_when_dropped_multiple" => should_cancel_blocking_when_dropped_multiple(runtime), + "should_wait_for_code_to_finish" => should_wait_for_code_to_finish(runtime), + "should_wait_for_code_to_finish_multiple" => should_wait_for_code_to_finish_multiple(runtime), + "should_wait_for_code_to_finish_blocking" => should_wait_for_code_to_finish_blocking(runtime), + "should_wait_for_code_to_finish_worker" => should_wait_for_code_to_finish_worker(runtime), + "should_wait_for_code_to_finish_worker_blocking" => should_wait_for_code_to_finish_worker_blocking(runtime), + "should_wait_for_code_to_finish_context" => should_wait_for_code_to_finish_context(runtime), + "should_wait_for_code_to_finish_context_blocking" => should_wait_for_code_to_finish_context_blocking(runtime), + _ => panic!("No Case Selected"), + }?; + + Report::print(None, None, "end"); + Ok(()) +} + +fn non_blocking_exec(context: usize) -> Box ion::Result<()>> { + return Box::new(move |env| { + env.inc_ref(); + Report::print(Some(context), None, "start"); + + env.spawn_background({ + let env = env.as_async(); + + async move { + Report::print(Some(context), Some(true), "start"); + tokio::time::sleep(Duration::from_millis(1000)).await; + Report::print(Some(context), Some(true), "end"); + + env.exec_async(move |env| { + env.dec_ref(); + Report::print(Some(context), None, "resolved"); + Ok(()) + }) + .await + } + })?; + + Report::print(Some(context), None, "end"); + Ok(()) + }); +} + +fn should_cancel_when_dropped(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + Ok(()) +} + +fn should_cancel_when_dropped_multiple(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + c0.exec(non_blocking_exec(0))?; + + Ok(()) +} + +fn should_cancel_blocking_when_dropped(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + + Ok(()) +} + +fn should_cancel_blocking_when_dropped_multiple(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + c0.exec_blocking(non_blocking_exec(0))?; + + Ok(()) +} + +fn should_wait_for_code_to_finish(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + + c0.join_blocking()?; + w0.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_multiple(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + c0.exec(non_blocking_exec(0))?; + + c0.join_blocking()?; + w0.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_blocking(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + + c0.join_blocking()?; + w0.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_worker(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + + w0.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_worker_blocking(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + + w0.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_context(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + + c0.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_context_blocking(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + + c0.join_blocking()?; + + Ok(()) +} diff --git a/examples/src/main.rs b/examples/src/main.rs index 38d894d..45237aa 100644 --- a/examples/src/main.rs +++ b/examples/src/main.rs @@ -1,14 +1,15 @@ -#![deny(unused_crate_dependencies)] +// #![deny(unused_crate_dependencies)] mod _utils; mod background_tasks; mod basic; +mod basic_join; mod context_multiplexing; mod custom_extension; mod custom_resolver; mod deferred; mod eval; mod external_value; -mod http_server; +// mod http_server; mod memory_usage_context; mod memory_usage_external_value; mod memory_usage_module; @@ -34,11 +35,12 @@ fn main() -> anyhow::Result<()> { match example.as_str() { "basic" => basic::main(), + "basic_join" => basic_join::main(), "custom_extension" => custom_extension::main(), "custom_resolver" => custom_resolver::main(), "deferred" => deferred::main(), "eval" => eval::main(), - "http_server" => http_server::main(), + // "http_server" => http_server::main(), "promise" => promise::main(), "run" => run::main(), "set_interval" => set_interval::main(), diff --git a/examples/test-utils/run_test.ts b/examples/test-utils/run_test.ts index a435f5f..d8591c3 100644 --- a/examples/test-utils/run_test.ts +++ b/examples/test-utils/run_test.ts @@ -6,7 +6,7 @@ export async function executeExample(testName: string, args: string[] = [], env: const command = new Deno.Command(Paths["~/"]("target", "debug", binName), { args: [testName, ...args], stdout: "piped", - stderr: "piped", + stderr: "inherit", cwd: Paths["~"], env: { ...Deno.env.toObject(), @@ -14,12 +14,11 @@ export async function executeExample(testName: string, args: string[] = [], env: } }); - const { code, stdout, stderr } = await command.output(); + const { code, stdout } = await command.output(); if (code !== 0) { - const errorText = new TextDecoder().decode(stderr); throw new Error( - `Test '${testName}' failed with exit code ${code}:\n${errorText}` + `Test '${testName}' failed with exit code ${code}:\n` ); } From f6a5d0d40730e9f1bab897ff5acf4efbc9916874 Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 5 Jan 2026 16:54:31 +1100 Subject: [PATCH 3/6] Working --- crates/ion/src/error.rs | 2 + crates/ion/src/js_context.rs | 37 +- crates/ion/src/js_runtime.rs | 17 +- crates/ion/src/js_worker.rs | 38 +- crates/ion/src/platform/callback_registry.rs | 101 --- crates/ion/src/platform/mod.rs | 2 +- crates/ion/src/platform/platform.rs | 12 +- crates/ion/src/platform/realm.rs | 4 + crates/ion/src/platform/worker.rs | 123 ++-- .../ion/src/platform/worker_handle_state.rs | 99 +++ crates/ion/src/utils/complete_signal.rs | 124 ++++ crates/ion/src/utils/mod.rs | 1 + examples/src/basic_join/basic_join.test.ts | 594 ++++++++++++++---- examples/src/basic_join/mod.rs | 50 +- 14 files changed, 857 insertions(+), 347 deletions(-) delete mode 100644 crates/ion/src/platform/callback_registry.rs create mode 100644 crates/ion/src/platform/worker_handle_state.rs create mode 100644 crates/ion/src/utils/complete_signal.rs diff --git a/crates/ion/src/error.rs b/crates/ion/src/error.rs index ecab1cd..db565f7 100644 --- a/crates/ion/src/error.rs +++ b/crates/ion/src/error.rs @@ -13,6 +13,8 @@ pub enum Error { PlatformCommunicationError, PlatformInitializeError, PlatformDisposeError, + WorkerAlreadyShutdown, + ContextAlreadyShutdown, IsolateNotInitializedError, EventLoopNotInitializedError, WorkerInitializeError, diff --git a/crates/ion/src/js_context.rs b/crates/ion/src/js_context.rs index be70bef..e1e1f55 100644 --- a/crates/ion/src/js_context.rs +++ b/crates/ion/src/js_context.rs @@ -6,16 +6,17 @@ use flume::bounded; use crate::Env; use crate::Error; use crate::JsUnknown; -use crate::platform::callback_registry::CallbackRegistry; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::worker::JsWorkerEvent; -use crate::utils::channel::oneshot; +use crate::utils::complete_signal::CompleteSignal; /// This is a handle to a v8::Context #[derive(Debug, Clone)] pub struct JsContext { - pub(crate) callback_registry: Arc, + pub(crate) worker_handle_state: Arc, pub(crate) id: usize, pub(crate) tx: Sender, + pub(crate) context_shutdown_sig: CompleteSignal, } impl JsContext { @@ -90,37 +91,33 @@ impl JsContext { /// Wait for the context to complete all activity pub fn join_blocking(self) -> crate::Result<()> { - if !self.callback_registry.worker_handle_active() { - return Ok(()); + if !self.worker_handle_state.worker_handle_active() { + return Err(crate::Error::WorkerAlreadyShutdown); } - self.callback_registry - .context_handle_set_status(&self.id, false); - - let (tx, rx) = oneshot(); - self.callback_registry - .add_context_shutdown_callback(self.id.clone(), move || tx.try_send(()).unwrap()); - - drop(self.tx.send(JsWorkerEvent::ContextHandleDeactivated { - id: self.id.clone(), - })); - - if rx.recv().is_err() { - panic!("Cannot drop JsContext 2") + if self + .tx + .send(JsWorkerEvent::ContextHandleDeactivated { + id: self.id.clone(), + }) + .is_err() + { + return Err(crate::Error::ContextAlreadyShutdown); } + self.context_shutdown_sig.wait(); + Ok(()) } /// Wait for the context to complete all activity pub async fn join_async(&self) -> crate::Result<()> { - self.callback_registry - .context_handle_set_status(&self.id, false); self.tx .send(JsWorkerEvent::ContextHandleDropped { id: self.id.clone(), }) .unwrap(); + Ok(()) } } diff --git a/crates/ion/src/js_runtime.rs b/crates/ion/src/js_runtime.rs index 628f496..fcefb22 100644 --- a/crates/ion/src/js_runtime.rs +++ b/crates/ion/src/js_runtime.rs @@ -12,9 +12,10 @@ use crate::JsResolver; use crate::JsTransformer; use crate::JsWorker; use crate::JsWorkerOptions; -use crate::platform::callback_registry::CallbackRegistry; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::platform::HAS_INIT; use crate::platform::platform::PLATFORM; +use crate::utils::complete_signal::CompleteSignal; static JS_RUNTIME: OnceLock>> = OnceLock::new(); @@ -175,13 +176,16 @@ impl JsRuntime { &self, options: JsWorkerOptions, ) -> crate::Result { + let worker_handle_state = Arc::new(WorkerHandleState::default()); + let worker_shutdown_sig = CompleteSignal::default(); + let (tx, rx) = bounded(1); - let callback_registry = Arc::new(CallbackRegistry::default()); if self .tx .send(PlatformEvent::SpawnWorker { - callback_registry: Arc::clone(&callback_registry), + worker_shutdown_sig: worker_shutdown_sig.clone(), + worker_handle_state: Arc::clone(&worker_handle_state), extensions: options.extensions, transformers: options.transformers, resolvers: options.resolvers, @@ -196,7 +200,12 @@ impl JsRuntime { return Err(Error::WorkerInitializeError); }; - Ok(JsWorker::new(callback_registry, tx, Arc::new(handle))) + Ok(JsWorker::new( + worker_handle_state, + tx, + Arc::new(handle), + worker_shutdown_sig, + )) } } diff --git a/crates/ion/src/js_worker.rs b/crates/ion/src/js_worker.rs index d162d16..909bf6a 100644 --- a/crates/ion/src/js_worker.rs +++ b/crates/ion/src/js_worker.rs @@ -10,9 +10,9 @@ use crate::Error; use crate::JsExtension; use crate::JsResolver; use crate::JsTransformer; -use crate::platform::callback_registry::CallbackRegistry; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::worker::JsWorkerEvent; -use crate::utils::channel::oneshot; +use crate::utils::complete_signal::CompleteSignal; #[derive(Default)] pub struct JsWorkerOptions { @@ -26,36 +26,42 @@ pub struct JsWorkerOptions { pub extensions: Vec, } + /// This is a handle to a v8::Isolate running on a dedicated thread. /// A worker thread can spawn multiple v8::Contexts within that thread /// to be used to execute JavaScript #[derive(Debug)] pub struct JsWorker { - callback_registry: Arc, + worker_handle_state: Arc, tx: Sender, handle: Arc>>>>, + worker_shutdown_sig: CompleteSignal, } impl JsWorker { pub(crate) fn new( - callback_registry: Arc, + worker_handle_state: Arc, tx: Sender, handle: Arc>>>>, + worker_shutdown_sig: CompleteSignal, ) -> Self { JsWorker { tx, handle, - callback_registry, + worker_handle_state, + worker_shutdown_sig, } } /// Create a handle to a v8::Context associated with this v8::Isolate pub fn create_context(&self) -> crate::Result { + let context_shutdown_sig = CompleteSignal::default(); + let (tx, rx) = bounded(1); if self .tx - .send(JsWorkerEvent::CreateContext { resolve: tx }) + .send(JsWorkerEvent::CreateContext { resolve: tx, context_shutdown_sig: context_shutdown_sig.clone() }) .is_err() { return Err(Error::WorkerInitializeError); @@ -65,12 +71,11 @@ impl JsWorker { return Err(Error::WorkerInitializeError); }; - self.callback_registry.context_handle_set_status(&id, true); - Ok(JsContext { id, tx, - callback_registry: Arc::clone(&self.callback_registry), + worker_handle_state: Arc::clone(&self.worker_handle_state), + context_shutdown_sig, }) } @@ -90,21 +95,12 @@ impl JsWorker { /// Wait for all of the contexts within the worker to complete all activity pub fn join_blocking(self) -> crate::Result<()> { - self.callback_registry.worker_handle_deactivate(); - - let (tx, rx) = oneshot(); - self.callback_registry - .add_worker_shutdown_callback(move || { - tx.send(()).unwrap(); - }); - + self.worker_handle_state.worker_handle_deactivate(); self.tx .send(JsWorkerEvent::WorkerHandleDeactivated) .unwrap(); - if rx.recv().is_err() { - panic!("Cannot drop JsWorker 2"); - } + self.worker_shutdown_sig.wait(); let Ok(mut handle) = self.handle.lock() else { panic!("Cannot drop JsWorker 3"); @@ -125,7 +121,7 @@ impl JsWorker { impl Drop for JsWorker { fn drop(&mut self) { - self.callback_registry.worker_handle_deactivate(); + self.worker_handle_state.worker_handle_deactivate(); drop(self.tx.try_send(JsWorkerEvent::WorkerHandleDropped)); } } diff --git a/crates/ion/src/platform/callback_registry.rs b/crates/ion/src/platform/callback_registry.rs deleted file mode 100644 index aa6994e..0000000 --- a/crates/ion/src/platform/callback_registry.rs +++ /dev/null @@ -1,101 +0,0 @@ -use std::collections::HashMap; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::Ordering; - -use parking_lot::Mutex; -use parking_lot::RwLock; - -pub type WorkerShutdownCallback = Box; -pub type ContextShutdownCallback = Box; - -pub struct CallbackRegistry { - pub worker_handle_active: AtomicBool, - pub context_handle_active: RwLock>, - pub worker_shutdown: Mutex>, - pub context_shutdown: Mutex>>, -} - -impl Default for CallbackRegistry { - fn default() -> Self { - Self { - worker_handle_active: AtomicBool::new(true), - context_handle_active: Default::default(), - worker_shutdown: Default::default(), - context_shutdown: Default::default(), - } - } -} - -impl CallbackRegistry { - pub(crate) fn worker_handle_active(&self) -> bool { - self.worker_handle_active.load(Ordering::Relaxed) - } - - pub(crate) fn worker_handle_deactivate(&self) { - self.worker_handle_active.swap(false, Ordering::Relaxed); - } - - #[allow(dead_code)] - pub(crate) fn context_handle_active( - &self, - id: &usize, - ) -> bool { - self.context_handle_active - .read() - .get(id) - .unwrap_or(&false) - .clone() - } - - pub(crate) fn context_handle_set_status( - &self, - id: &usize, - status: bool, - ) { - self.context_handle_active - .write() - .insert(id.clone(), status); - } - - pub(crate) fn add_worker_shutdown_callback( - &self, - callback: impl 'static + Send + Sync + FnOnce(), - ) { - self.worker_shutdown.lock().push(Box::new(callback)); - } - - pub(crate) fn add_context_shutdown_callback( - &self, - id: usize, - callback: impl 'static + Send + Sync + FnOnce(), - ) { - self.context_shutdown - .lock() - .entry(id) - .or_default() - .push(Box::new(callback)); - } - - pub(crate) fn take_worker_shutdown_callbacks(&self) -> Vec { - std::mem::take(&mut *self.worker_shutdown.lock()) - } - - pub(crate) fn take_context_shutdown_callbacks( - &self, - id: usize, - ) -> Vec { - std::mem::take(&mut *self.context_shutdown.lock().entry(id).or_default()) - } -} - -impl std::fmt::Debug for CallbackRegistry { - fn fmt( - &self, - f: &mut std::fmt::Formatter<'_>, - ) -> std::fmt::Result { - f.debug_struct("CallbackRegistry") - .field("worker_shutdown", &self.worker_shutdown.lock().len()) - .field("context_shutdown", &self.context_shutdown.lock().len()) - .finish() - } -} diff --git a/crates/ion/src/platform/mod.rs b/crates/ion/src/platform/mod.rs index 1833fb8..d127119 100644 --- a/crates/ion/src/platform/mod.rs +++ b/crates/ion/src/platform/mod.rs @@ -1,6 +1,6 @@ #![allow(clippy::module_inception)] pub mod background_worker; -pub(crate) mod callback_registry; +pub(crate) mod worker_handle_state; pub(crate) mod extension; pub(crate) mod finalizer_registry; pub mod module; diff --git a/crates/ion/src/platform/platform.rs b/crates/ion/src/platform/platform.rs index 9b5e5af..fc55128 100644 --- a/crates/ion/src/platform/platform.rs +++ b/crates/ion/src/platform/platform.rs @@ -14,9 +14,10 @@ use crate::JsExtension; use crate::JsResolver; use crate::JsTransformer; use crate::platform::background_worker::BackgroundTaskManager; -use crate::platform::callback_registry::CallbackRegistry; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::worker::JsWorkerEvent; use crate::platform::worker::start_js_worker_thread; +use crate::utils::complete_signal::CompleteSignal; pub(crate) enum PlatformEvent { Init { @@ -27,7 +28,8 @@ pub(crate) enum PlatformEvent { transformers: Vec, }, SpawnWorker { - callback_registry: Arc, + worker_shutdown_sig: CompleteSignal, + worker_handle_state: Arc, extensions: Vec, resolvers: Vec, transformers: Vec, @@ -98,7 +100,8 @@ pub(crate) static PLATFORM: LazyLock> = LazyLock::new(|| { } } PlatformEvent::SpawnWorker { - callback_registry, + worker_shutdown_sig, + worker_handle_state, resolve, extensions: init_extensions, resolvers: init_resolvers, @@ -120,7 +123,8 @@ pub(crate) static PLATFORM: LazyLock> = LazyLock::new(|| { } let (tx, handle) = start_js_worker_thread( - callback_registry, + worker_shutdown_sig, + worker_handle_state, background_task_manager.clone(), worker_extensions, worker_resolvers, diff --git a/crates/ion/src/platform/realm.rs b/crates/ion/src/platform/realm.rs index 6c154c7..6c16bce 100644 --- a/crates/ion/src/platform/realm.rs +++ b/crates/ion/src/platform/realm.rs @@ -16,6 +16,7 @@ use crate::platform::sys; use crate::platform::worker::JsWorkerEvent; use crate::utils::RefCounter; use crate::utils::channel::oneshot; +use crate::utils::complete_signal::CompleteSignal; // Container that constructs a V8 context and preserves the internals until dropped pub struct JsRealm { @@ -31,6 +32,7 @@ pub struct JsRealm { pub(crate) global_refs: RefCounter, pub(crate) modules: ModuleMap, pub(crate) global_this: sys::GlobalThis, + pub(crate) context_shutdown_sig: CompleteSignal, } impl JsRealm { @@ -41,6 +43,7 @@ impl JsRealm { transformers: HashMap>, background_task_manager: Arc, tx: Sender, + context_shutdown_sig: CompleteSignal, ) -> Box { let context = sys::GlobalContext::new(unsafe { &mut *isolate }); let global_this = sys::GlobalThis::new(&context); @@ -72,6 +75,7 @@ impl JsRealm { global_refs, finalizer_registry, global_this, + context_shutdown_sig, }); let realm_ptr = realm.as_mut() as *mut JsRealm; diff --git a/crates/ion/src/platform/worker.rs b/crates/ion/src/platform/worker.rs index 1b04074..6abaefc 100644 --- a/crates/ion/src/platform/worker.rs +++ b/crates/ion/src/platform/worker.rs @@ -18,12 +18,14 @@ use crate::JsResolver; use crate::JsTransformer; use crate::fs::FileSystem; use crate::platform::background_worker::BackgroundTaskManager; -use crate::platform::callback_registry::CallbackRegistry; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::utils::HashMapExt; use crate::utils::PathExt; +use crate::utils::complete_signal::CompleteSignal; pub(crate) enum JsWorkerEvent { CreateContext { + context_shutdown_sig: CompleteSignal, resolve: Sender<(usize, Sender)>, }, Exec { @@ -36,10 +38,6 @@ pub(crate) enum JsWorkerEvent { id: usize, specifier: String, }, - TryShutdownContext { - id: usize, - force: bool, - }, RunGarbageCollectionForTesting { resolve: Sender<()>, }, @@ -56,7 +54,8 @@ pub(crate) enum JsWorkerEvent { // Create a dedicated thread to host the isolate #[allow(clippy::type_complexity)] pub(crate) fn start_js_worker_thread( - callback_registry: Arc, + worker_shutdown_sig: CompleteSignal, + worker_handle_state: Arc, background_task_manager: Arc, extensions: Vec>, resolvers: Vec, @@ -72,7 +71,8 @@ pub(crate) fn start_js_worker_thread( let tx: Sender = tx.clone(); move || { worker_thread( - callback_registry, + worker_shutdown_sig.clone(), + worker_handle_state, tx, rx, background_task_manager, @@ -87,7 +87,8 @@ pub(crate) fn start_js_worker_thread( } fn worker_thread( - callback_registry: Arc, + worker_shutdown_sig: CompleteSignal, + worker_handle_state: Arc, tx: Sender, rx: Receiver, background_task_manager: Arc, @@ -105,14 +106,10 @@ fn worker_thread( let mut realms = HashMap::>::new(); while let Ok(event) = rx.recv() { - // eprintln!("{:?}", event); - - if realms.len() == 0 && !callback_registry.worker_handle_active() { - break; - } + // println!(" {:?}", event); match event { - JsWorkerEvent::CreateContext { resolve } => { + JsWorkerEvent::CreateContext { resolve, context_shutdown_sig } => { let realm = JsRealm::new( isolate_ptr, fs.clone(), @@ -120,6 +117,7 @@ fn worker_thread( transformers.clone(), background_task_manager.clone(), tx.clone(), + context_shutdown_sig, ); let realm_id = realm.id(); @@ -137,8 +135,19 @@ fn worker_thread( panic!("Callback errored {:?}", err) }; - tx.try_send(JsWorkerEvent::TryShutdownContext { id, force: false }) - .unwrap(); + if realm.global_refs.count() != 0 { + continue; + } + + let Some(realm) = realms.remove(&id) else { + continue; + }; + + let finalizer_registry = realm.finalizer_registry; + finalizer_registry.clear(); + drop(finalizer_registry); + + realm.context_shutdown_sig.done(); } JsWorkerEvent::Import { id, specifier } => { Module::v8_initialize( @@ -162,46 +171,38 @@ fn worker_thread( finalizer_registry.clear(); drop(finalizer_registry); - for shutdown_callback in callback_registry - .take_context_shutdown_callbacks(id.clone()) - .into_iter() - { - shutdown_callback(); - } + realm.context_shutdown_sig.done(); } break; } JsWorkerEvent::WorkerHandleDeactivated => { - for id in realms.keys() { - tx.try_send(JsWorkerEvent::TryShutdownContext { - id: id.clone(), - force: true, - }) - .unwrap(); + let mut to_drop = vec![]; + + for (id, realm) in realms.iter() { + if realm.global_refs.count() != 0 { + continue; + } + to_drop.push(id.clone()); + } + + for id in to_drop { + let Some(realm) = realms.remove(&id) else { + continue; + }; + + let finalizer_registry = realm.finalizer_registry; + finalizer_registry.clear(); + drop(finalizer_registry); + + realm.context_shutdown_sig.done(); } } JsWorkerEvent::ContextHandleDeactivated { id } => { - tx.try_send(JsWorkerEvent::TryShutdownContext { - id: id.clone(), - force: false, - }) - .unwrap(); - } - JsWorkerEvent::ContextHandleDropped { id } => { - tx.try_send(JsWorkerEvent::TryShutdownContext { - id: id.clone(), - force: true, - }) - .unwrap(); - } - JsWorkerEvent::TryShutdownContext { id, force } => { - // If there are async tasks pending then wait for them to complete - if !force && realms.try_get_mut(&id)?.global_refs.count() != 0 { + if realms.try_get_mut(&id)?.global_refs.count() != 0 { continue; } - // If there are no async tasks then shutdown the context let Some(realm) = realms.remove(&id) else { continue; }; @@ -210,23 +211,28 @@ fn worker_thread( finalizer_registry.clear(); drop(finalizer_registry); - for shutdown_callback in callback_registry - .take_context_shutdown_callbacks(id.clone()) - .into_iter() - { - shutdown_callback(); - } + realm.context_shutdown_sig.done(); + } + JsWorkerEvent::ContextHandleDropped { id } => { + let Some(realm) = realms.remove(&id) else { + continue; + }; + + let finalizer_registry = realm.finalizer_registry; + finalizer_registry.clear(); + drop(finalizer_registry); + + realm.context_shutdown_sig.done(); } } - } - for shutdown_callback in callback_registry - .take_worker_shutdown_callbacks() - .into_iter() - { - shutdown_callback(); + if realms.len() == 0 && !worker_handle_state.worker_handle_active() { + break; + } } + worker_shutdown_sig.done(); + Ok(()) } @@ -238,10 +244,9 @@ impl std::fmt::Debug for JsWorkerEvent { f: &mut std::fmt::Formatter<'_>, ) -> std::fmt::Result { match self { - Self::CreateContext { resolve } => write!(f, "CreateContext"), + Self::CreateContext { resolve, context_shutdown_sig } => write!(f, "CreateContext"), Self::Exec { id, callback, span } => write!(f, "Exec [id={}]", id), Self::Import { id, specifier } => write!(f, "Import"), - Self::TryShutdownContext { id, force } => write!(f, "TryShutdownContext [id={} force={}]", id, force), Self::WorkerHandleDropped => write!(f, "WorkerHandleDropped"), Self::WorkerHandleDeactivated => write!(f, "WorkerHandleDeactivated"), Self::ContextHandleDropped { id } => write!(f, "ContextHandleDropped"), diff --git a/crates/ion/src/platform/worker_handle_state.rs b/crates/ion/src/platform/worker_handle_state.rs new file mode 100644 index 0000000..deeada3 --- /dev/null +++ b/crates/ion/src/platform/worker_handle_state.rs @@ -0,0 +1,99 @@ +use std::collections::HashMap; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; + +use parking_lot::Mutex; + +pub type WorkerShutdownCallback = Box; +pub type ContextShutdownCallback = Box; + +pub struct WorkerHandleState { + pub (crate) worker_handle_active: AtomicBool, + // pub (crate) context_handle_active: RwLock>, + pub (crate) worker_shutdown: Mutex>, + pub (crate) context_shutdown: Mutex>>, +} + +impl Default for WorkerHandleState { + fn default() -> Self { + Self { + worker_handle_active: AtomicBool::new(true), + // context_handle_active: Default::default(), + worker_shutdown: Default::default(), + context_shutdown: Default::default(), + } + } +} + +impl WorkerHandleState { + pub(crate) fn worker_handle_active(&self) -> bool { + self.worker_handle_active.load(Ordering::Relaxed) + } + + pub(crate) fn worker_handle_deactivate(&self) { + self.worker_handle_active.swap(false, Ordering::Relaxed); + } + + // pub(crate) fn context_handle_active( + // &self, + // id: &usize, + // ) -> bool { + // self.context_handle_active + // .read() + // .get(id) + // .unwrap_or(&false) + // .clone() + // } + + // pub(crate) fn context_handle_set_status( + // &self, + // id: &usize, + // status: bool, + // ) { + // self.context_handle_active + // .write() + // .insert(id.clone(), status); + // } + + // pub(crate) fn add_worker_shutdown_callback( + // &self, + // callback: impl 'static + Send + Sync + FnOnce(), + // ) { + // self.worker_shutdown.lock().push(Box::new(callback)); + // } + + // pub(crate) fn add_context_shutdown_callback( + // &self, + // id: usize, + // callback: impl 'static + Send + Sync + FnOnce(), + // ) { + // self.context_shutdown + // .lock() + // .entry(id) + // .or_default() + // .push(Box::new(callback)); + // } + + // pub(crate) fn take_worker_shutdown_callbacks(&self) -> Vec { + // std::mem::take(&mut *self.worker_shutdown.lock()) + // } + + // pub(crate) fn take_context_shutdown_callbacks( + // &self, + // id: usize, + // ) -> Vec { + // std::mem::take(&mut *self.context_shutdown.lock().entry(id).or_default()) + // } +} + +impl std::fmt::Debug for WorkerHandleState { + fn fmt( + &self, + f: &mut std::fmt::Formatter<'_>, + ) -> std::fmt::Result { + f.debug_struct("CallbackRegistry") + .field("worker_shutdown", &self.worker_shutdown.lock().len()) + .field("context_shutdown", &self.context_shutdown.lock().len()) + .finish() + } +} diff --git a/crates/ion/src/utils/complete_signal.rs b/crates/ion/src/utils/complete_signal.rs new file mode 100644 index 0000000..133768e --- /dev/null +++ b/crates/ion/src/utils/complete_signal.rs @@ -0,0 +1,124 @@ +use std::sync::{Arc, Condvar, Mutex}; +use tokio::sync::Notify; + +#[derive(Clone, Default)] +pub struct CompleteSignal { + inner: Arc, +} + +impl std::fmt::Debug for CompleteSignal { + fn fmt( + &self, + f: &mut std::fmt::Formatter<'_>, + ) -> std::fmt::Result { + f.debug_struct("CompleteSignal") + .finish() + } +} + +struct Inner { + completed: Mutex, + condvar: Condvar, + notify: Notify, +} + +impl Default for Inner { + fn default() -> Self { + Self { + completed: Mutex::new(false), + condvar: Condvar::new(), + notify: Notify::new(), + } + } +} + +impl CompleteSignal { + pub fn new() -> Self { + Self::default() + } + + pub fn done(&self) { + let mut completed = self.inner.completed.lock().unwrap(); + if !*completed { + *completed = true; + self.inner.condvar.notify_all(); + self.inner.notify.notify_waiters(); + } + } + + pub fn wait(&self) { + let mut completed = self.inner.completed.lock().unwrap(); + while !*completed { + completed = self.inner.condvar.wait(completed).unwrap(); + } + } + + pub async fn wait_async(&self) { + { + let completed = self.inner.completed.lock().unwrap(); + if *completed { + return; + } + } + + let notified = self.inner.notify.notified(); + + { + let completed = self.inner.completed.lock().unwrap(); + if *completed { + return; + } + } + + notified.await; + } + + pub fn is_done(&self) -> bool { + *self.inner.completed.lock().unwrap() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::thread; + use std::time::Duration; + + #[tokio::test] + async fn test_complete_signal() { + let sig = CompleteSignal::default(); + + thread::spawn({ + let sig = sig.clone(); + move || { + println!("** Controller WAITING"); + thread::sleep(Duration::from_millis(100)); + println!("** Controller DONE"); + sig.done(); + } + }); + + thread::spawn({ + let sig = sig.clone(); + move || { + println!("Sync WAITING"); + sig.wait(); + println!("Sync DONE 1"); + sig.wait(); + sig.wait(); + println!("Sync DONE 2"); + } + }); + + println!("Async WAITING"); + sig.wait_async().await; + println!("Async DONE 1"); + + // Subsequent calls after the signal is complete will complete immediately + sig.wait_async().await; + sig.wait_async().await; + println!("Async DONE 2"); + + assert!(sig.is_done()); + } +} diff --git a/crates/ion/src/utils/mod.rs b/crates/ion/src/utils/mod.rs index 54b4dcd..122bbaa 100644 --- a/crates/ion/src/utils/mod.rs +++ b/crates/ion/src/utils/mod.rs @@ -8,6 +8,7 @@ pub mod random_string; pub mod ref_counter; pub mod ref_counter_atomic; pub mod tokio_ext; +pub mod complete_signal; pub use debug::*; pub use hash::*; diff --git a/examples/src/basic_join/basic_join.test.ts b/examples/src/basic_join/basic_join.test.ts index 802c653..bde7120 100644 --- a/examples/src/basic_join/basic_join.test.ts +++ b/examples/src/basic_join/basic_join.test.ts @@ -1,215 +1,537 @@ import { executeExample } from "../../test-utils/run_test.ts"; import { assert, assertEquals, assertObjectMatch } from "jsr:@std/assert@^1"; -type Record = { +type Result = { thread: number; message: string; js_context?: number; event_loop?: boolean; }; -async function executeBasicJoin(caseName: string): Promise> { +async function executeBasicJoin(caseName: string): Promise> { const result = await executeExample("basic_join", [caseName]); return result.split("\n").map((record) => JSON.parse(record)); } -function assertArraysMatch(arr1: any, arr2: any, msg?: string): void { - return assertObjectMatch({ arr: arr1 }, { arr: arr2 }, msg); +type ProcessedRecords = { + main: Array; + jsContexts: Record>; + eventLoop: Record>; +}; + +function processResults(input: Array): ProcessedRecords { + const processed: ProcessedRecords = { + main: [], + jsContexts: {}, + eventLoop: {}, + }; + + for (const result of input) { + // No js_context -> goes to main + if (result.js_context === undefined) { + processed.main.push(result); + continue; + } + + // Has event_loop flag -> goes to eventLoop + if (result.event_loop) { + if (!processed.eventLoop[result.js_context]) { + processed.eventLoop[result.js_context] = []; + } + processed.eventLoop[result.js_context].push(result); + continue; + } + + // Has js_context but no event_loop -> goes to jsContexts + if (!processed.jsContexts[result.js_context]) { + processed.jsContexts[result.js_context] = []; + } + processed.jsContexts[result.js_context].push(result); + } + + return processed; } -function filterJs(ctx: number, event_loop: boolean = false) { - return (r: Record): boolean => - r.js_context === ctx && !!r.event_loop == event_loop; +async function run(caseName: string): Promise { + return processResults(await executeBasicJoin(caseName)); } +function assertArraysMatch, Y extends Array>( + arr1: T, + arr2: Y, + msg?: string +): void { + return assertObjectMatch({ arr: arr1 }, { arr: arr2 }, msg); +} Deno.test("should_cancel_when_dropped", async () => { const example = "should_cancel_when_dropped"; - const results = await executeBasicJoin(example); + const results = await run(example); // The code on the main thread will always run - assertEquals(results.filter((r) => r.thread === 1).length, 2); + assertArraysMatch(results.main, [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ]); // The code on the JavaScript thread may or may not run assert( - results.filter((r) => r.js_context === 0 && !r.event_loop).length === - 2 || - results.filter((r) => r.js_context === 0 && !r.event_loop) - .length === 0 + (results.jsContexts[0] || []).length === 0 || + (results.jsContexts[0] || []).length === 2 ); // The code on the Event Loop should not run - assertEquals(results.filter((r) => r.event_loop).length, 0); + assertObjectMatch(results.eventLoop, {}); }); Deno.test("should_cancel_when_dropped_multiple", async () => { const example = "should_cancel_when_dropped_multiple"; - const results = await executeBasicJoin(example); + const results = await run(example); // The code on the main thread will always run - assertEquals(results.filter((r) => r.thread === 1).length, 2); + assertArraysMatch(results.main, [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ]); // The code on the JavaScript thread may or may not run assert( - results.filter((r) => r.js_context === 0 && !r.event_loop).length === - 4 || - results.filter((r) => r.js_context === 0 && !r.event_loop) - .length === 0 + (results.jsContexts[0] || []).length === 0 || + (results.jsContexts[0] || []).length === 4 ); // The code on the Event Loop should not run - assertEquals(results.filter((r) => r.event_loop).length, 0); + assertObjectMatch(results.eventLoop, {}); }); Deno.test("should_cancel_blocking_when_dropped", async () => { const example = "should_cancel_blocking_when_dropped"; - const results = await executeBasicJoin(example); + const results = await run(example); - // The code on the main thread will always run - assertEquals(results.filter((r) => r.thread === 1).length, 2); + assertArraysMatch(results.main, [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ]); - // The code on the JavaScript thread must run - assertEquals( - results.filter((r) => r.js_context === 0 && !r.event_loop).length, - 2 - ); + assertObjectMatch(results.jsContexts, { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + ], + }); - // The code on the Event Loop may or may not run, but not progress - assert( - results.filter( - (r) => r.js_context === 0 && r.event_loop && r.message !== "end" - ).length === 1 || - results.filter((r) => r.js_context === 0 && r.event_loop).length === - 0 - ); + // The code on the Event Loop may or may not run, but not progress when the task sleeps + assertObjectMatch(results.eventLoop, {}); }); Deno.test("should_cancel_blocking_when_dropped_multiple", async () => { const example = "should_cancel_blocking_when_dropped_multiple"; - const results = await executeBasicJoin(example); + const results = await run(example); - // The code on the main thread will always run - assertEquals(results.filter((r) => r.thread === 1).length, 2); - - // The code on the JavaScript thread must run - assertEquals( - results.filter((r) => r.js_context === 0 && !r.event_loop).length, - 4 - ); + assertArraysMatch(results.main, [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ]); - // The code on the Event Loop may or may not run, but not progress - assert( - results.filter( - (r) => r.js_context === 0 && r.event_loop && r.message !== "end" - ).length === 2 || - results.filter((r) => r.js_context === 0 && r.event_loop).length === - 0 - ); + assertObjectMatch(results.jsContexts, { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + ], + }); + + // The code on the Event Loop may or may not run, but not progress when the task sleeps + assertObjectMatch(results.eventLoop, {}); }); Deno.test("should_wait_for_code_to_finish", async () => { const example = "should_wait_for_code_to_finish"; - const results = await executeBasicJoin(example); - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); -// Hangs -Deno.test.only("should_wait_for_code_to_finish_multiple", async () => { +Deno.test("should_wait_for_code_to_finish_multiple", async () => { const example = "should_wait_for_code_to_finish_multiple"; - const results = await executeBasicJoin(example); - - console.log(results); - - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); Deno.test("should_wait_for_code_to_finish_blocking", async () => { const example = "should_wait_for_code_to_finish_blocking"; - const results = await executeBasicJoin(example); - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); -// Does not complete context -Deno.test.ignore("should_wait_for_code_to_finish_worker", async () => { +Deno.test("should_wait_for_code_to_finish_worker", async () => { const example = "should_wait_for_code_to_finish_worker"; - const results = await executeBasicJoin(example); - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); -// Does not complete context -Deno.test.ignore("should_wait_for_code_to_finish_worker_blocking", async () => { +Deno.test("should_wait_for_code_to_finish_worker_blocking", async () => { const example = "should_wait_for_code_to_finish_worker_blocking"; - const results = await executeBasicJoin(example); - - console.log(results); - - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); Deno.test("should_wait_for_code_to_finish_context", async () => { const example = "should_wait_for_code_to_finish_context"; - const results = await executeBasicJoin(example); - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); Deno.test("should_wait_for_code_to_finish_context_blocking", async () => { const example = "should_wait_for_code_to_finish_context_blocking"; - const results = await executeBasicJoin(example); - assertArraysMatch(results, [ - { thread: 1, message: "start" }, - { thread: 2, js_context: 0, message: "start" }, - { thread: 2, js_context: 0, message: "end" }, - { thread: 1000, js_context: 0, event_loop: true, message: "start" }, - { thread: 1000, js_context: 0, event_loop: true, message: "end" }, - { thread: 2, js_context: 0, message: "resolved" }, - { thread: 1, message: "end" }, - ]); + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); +}); + +Deno.test("should_wait_for_code_to_finish_contexts_blocking", async () => { + const example = "should_wait_for_code_to_finish_contexts_blocking"; + const results = processResults(await executeBasicJoin(example)); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + "1": [ + { thread: 2, js_context: 1, message: "start" }, + { thread: 2, js_context: 1, message: "end" }, + { thread: 2, js_context: 1, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + "1": [ + { + thread: 1000, + js_context: 1, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 1, + event_loop: true, + message: "end", + }, + ], + }, + }); +}); + +Deno.test("should_wait_for_code_to_finish_contexts", async () => { + const example = "should_wait_for_code_to_finish_contexts"; + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); +}); + +Deno.test("should_not_run_code_after_joining", async () => { + const example = "should_not_run_code_after_joining"; + const results = await run(example); + assertObjectMatch(results, { + main: [ + { thread: 1, message: "start" }, + { thread: 1, message: "did_not_run" }, + { thread: 1, message: "end" }, + ], + jsContexts: { + "0": [ + { thread: 2, js_context: 0, message: "start" }, + { thread: 2, js_context: 0, message: "end" }, + { thread: 2, js_context: 0, message: "resolved" }, + ], + }, + eventLoop: { + "0": [ + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "start", + }, + { + thread: 1000, + js_context: 0, + event_loop: true, + message: "end", + }, + ], + }, + }); }); diff --git a/examples/src/basic_join/mod.rs b/examples/src/basic_join/mod.rs index c4c3067..19c77dd 100644 --- a/examples/src/basic_join/mod.rs +++ b/examples/src/basic_join/mod.rs @@ -63,6 +63,9 @@ pub fn main() -> anyhow::Result<()> { "should_wait_for_code_to_finish_worker_blocking" => should_wait_for_code_to_finish_worker_blocking(runtime), "should_wait_for_code_to_finish_context" => should_wait_for_code_to_finish_context(runtime), "should_wait_for_code_to_finish_context_blocking" => should_wait_for_code_to_finish_context_blocking(runtime), + "should_wait_for_code_to_finish_contexts" => should_wait_for_code_to_finish_contexts(runtime), + "should_wait_for_code_to_finish_contexts_blocking" => should_wait_for_code_to_finish_contexts_blocking(runtime), + "should_not_run_code_after_joining" => should_not_run_code_after_joining(runtime), _ => panic!("No Case Selected"), }?; @@ -80,7 +83,7 @@ fn non_blocking_exec(context: usize) -> Box ion::Result<( async move { Report::print(Some(context), Some(true), "start"); - tokio::time::sleep(Duration::from_millis(1000)).await; + tokio::time::sleep(Duration::from_millis(100)).await; Report::print(Some(context), Some(true), "end"); env.exec_async(move |env| { @@ -214,3 +217,48 @@ fn should_wait_for_code_to_finish_context_blocking(runtime: Arc) -> a Ok(()) } + +fn should_wait_for_code_to_finish_contexts(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + + let c0 = w0.create_context()?; + let c1 = w0.create_context()?; + + c0.exec(non_blocking_exec(0))?; + c1.exec(non_blocking_exec(0))?; + + c0.join_blocking()?; + c1.join_blocking()?; + + Ok(()) +} + +fn should_wait_for_code_to_finish_contexts_blocking(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + + let c0 = w0.create_context()?; + let c1 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + c1.exec_blocking(non_blocking_exec(1))?; + + c0.join_blocking()?; + c1.join_blocking()?; + + Ok(()) +} + +fn should_not_run_code_after_joining(runtime: Arc) -> anyhow::Result<()> { + let w0 = runtime.spawn_worker(JsWorkerOptions::default())?; + let c0 = w0.create_context()?; + + c0.exec_blocking(non_blocking_exec(0))?; + + w0.join_blocking()?; + + if c0.exec_blocking(non_blocking_exec(0)).is_err() { + Report::print(None, None, "did_not_run"); + }; + + Ok(()) +} From 706fab63f8a7199e66a09d52dbed8e395b54f77b Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 5 Jan 2026 16:59:00 +1100 Subject: [PATCH 4/6] formatting and async --- crates/ion/src/js_context.rs | 31 ++++++++++++++---- crates/ion/src/js_runtime.rs | 2 +- crates/ion/src/js_worker.rs | 27 +++++++++++++--- crates/ion/src/platform/mod.rs | 2 +- crates/ion/src/platform/platform.rs | 2 +- crates/ion/src/platform/worker.rs | 5 ++- .../ion/src/platform/worker_handle_state.rs | 6 ++-- crates/ion/src/utils/complete_signal.rs | 11 ++++--- crates/ion/src/utils/mod.rs | 2 +- examples/src/basic/mod.rs | 2 ++ examples/src/basic_join/mod.rs | 32 +++++++++---------- 11 files changed, 83 insertions(+), 39 deletions(-) diff --git a/crates/ion/src/js_context.rs b/crates/ion/src/js_context.rs index e1e1f55..9c35277 100644 --- a/crates/ion/src/js_context.rs +++ b/crates/ion/src/js_context.rs @@ -6,8 +6,8 @@ use flume::bounded; use crate::Env; use crate::Error; use crate::JsUnknown; -use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::worker::JsWorkerEvent; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::utils::complete_signal::CompleteSignal; /// This is a handle to a v8::Context @@ -90,7 +90,7 @@ impl JsContext { } /// Wait for the context to complete all activity - pub fn join_blocking(self) -> crate::Result<()> { + pub fn join(self) -> crate::Result<()> { if !self.worker_handle_state.worker_handle_active() { return Err(crate::Error::WorkerAlreadyShutdown); } @@ -112,12 +112,31 @@ impl JsContext { /// Wait for the context to complete all activity pub async fn join_async(&self) -> crate::Result<()> { - self.tx - .send(JsWorkerEvent::ContextHandleDropped { + if !self.worker_handle_state.worker_handle_active() { + return Err(crate::Error::WorkerAlreadyShutdown); + } + + if self + .tx + .send(JsWorkerEvent::ContextHandleDeactivated { id: self.id.clone(), }) - .unwrap(); - + .is_err() + { + return Err(crate::Error::ContextAlreadyShutdown); + } + + self.context_shutdown_sig.wait_async().await; + Ok(()) } } + +impl Drop for JsContext { + fn drop(&mut self) { + drop( + self.tx + .try_send(JsWorkerEvent::ContextHandleDropped { id: self.id }), + ); + } +} diff --git a/crates/ion/src/js_runtime.rs b/crates/ion/src/js_runtime.rs index fcefb22..503e600 100644 --- a/crates/ion/src/js_runtime.rs +++ b/crates/ion/src/js_runtime.rs @@ -12,9 +12,9 @@ use crate::JsResolver; use crate::JsTransformer; use crate::JsWorker; use crate::JsWorkerOptions; -use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::platform::HAS_INIT; use crate::platform::platform::PLATFORM; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::utils::complete_signal::CompleteSignal; static JS_RUNTIME: OnceLock>> = OnceLock::new(); diff --git a/crates/ion/src/js_worker.rs b/crates/ion/src/js_worker.rs index 909bf6a..d2799f8 100644 --- a/crates/ion/src/js_worker.rs +++ b/crates/ion/src/js_worker.rs @@ -10,8 +10,8 @@ use crate::Error; use crate::JsExtension; use crate::JsResolver; use crate::JsTransformer; -use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::worker::JsWorkerEvent; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::utils::complete_signal::CompleteSignal; #[derive(Default)] @@ -26,7 +26,6 @@ pub struct JsWorkerOptions { pub extensions: Vec, } - /// This is a handle to a v8::Isolate running on a dedicated thread. /// A worker thread can spawn multiple v8::Contexts within that thread /// to be used to execute JavaScript @@ -56,12 +55,15 @@ impl JsWorker { /// Create a handle to a v8::Context associated with this v8::Isolate pub fn create_context(&self) -> crate::Result { let context_shutdown_sig = CompleteSignal::default(); - + let (tx, rx) = bounded(1); if self .tx - .send(JsWorkerEvent::CreateContext { resolve: tx, context_shutdown_sig: context_shutdown_sig.clone() }) + .send(JsWorkerEvent::CreateContext { + resolve: tx, + context_shutdown_sig: context_shutdown_sig.clone(), + }) .is_err() { return Err(Error::WorkerInitializeError); @@ -94,7 +96,7 @@ impl JsWorker { } /// Wait for all of the contexts within the worker to complete all activity - pub fn join_blocking(self) -> crate::Result<()> { + pub fn join(self) -> crate::Result<()> { self.worker_handle_state.worker_handle_deactivate(); self.tx .send(JsWorkerEvent::WorkerHandleDeactivated) @@ -115,6 +117,21 @@ impl JsWorker { /// Wait for all of the contexts within the worker to complete all activity pub async fn join_async(self) -> crate::Result<()> { + self.worker_handle_state.worker_handle_deactivate(); + self.tx + .send(JsWorkerEvent::WorkerHandleDeactivated) + .unwrap(); + + self.worker_shutdown_sig.wait_async().await; + + let Ok(mut handle) = self.handle.lock() else { + panic!("Cannot drop JsWorker 3"); + }; + + if let Some(handle) = handle.take() { + drop(handle.join().unwrap()); + } + Ok(()) } } diff --git a/crates/ion/src/platform/mod.rs b/crates/ion/src/platform/mod.rs index d127119..8d1dc62 100644 --- a/crates/ion/src/platform/mod.rs +++ b/crates/ion/src/platform/mod.rs @@ -1,6 +1,5 @@ #![allow(clippy::module_inception)] pub mod background_worker; -pub(crate) mod worker_handle_state; pub(crate) mod extension; pub(crate) mod finalizer_registry; pub mod module; @@ -10,5 +9,6 @@ mod realm; pub mod resolve; pub(crate) mod sys; pub(crate) mod worker; +pub(crate) mod worker_handle_state; pub(crate) use realm::*; diff --git a/crates/ion/src/platform/platform.rs b/crates/ion/src/platform/platform.rs index fc55128..cdacb63 100644 --- a/crates/ion/src/platform/platform.rs +++ b/crates/ion/src/platform/platform.rs @@ -14,9 +14,9 @@ use crate::JsExtension; use crate::JsResolver; use crate::JsTransformer; use crate::platform::background_worker::BackgroundTaskManager; -use crate::platform::worker_handle_state::WorkerHandleState; use crate::platform::worker::JsWorkerEvent; use crate::platform::worker::start_js_worker_thread; +use crate::platform::worker_handle_state::WorkerHandleState; use crate::utils::complete_signal::CompleteSignal; pub(crate) enum PlatformEvent { diff --git a/crates/ion/src/platform/worker.rs b/crates/ion/src/platform/worker.rs index 6abaefc..92812c3 100644 --- a/crates/ion/src/platform/worker.rs +++ b/crates/ion/src/platform/worker.rs @@ -109,7 +109,10 @@ fn worker_thread( // println!(" {:?}", event); match event { - JsWorkerEvent::CreateContext { resolve, context_shutdown_sig } => { + JsWorkerEvent::CreateContext { + resolve, + context_shutdown_sig, + } => { let realm = JsRealm::new( isolate_ptr, fs.clone(), diff --git a/crates/ion/src/platform/worker_handle_state.rs b/crates/ion/src/platform/worker_handle_state.rs index deeada3..1cc0d57 100644 --- a/crates/ion/src/platform/worker_handle_state.rs +++ b/crates/ion/src/platform/worker_handle_state.rs @@ -8,10 +8,10 @@ pub type WorkerShutdownCallback = Box; pub type ContextShutdownCallback = Box; pub struct WorkerHandleState { - pub (crate) worker_handle_active: AtomicBool, + pub(crate) worker_handle_active: AtomicBool, // pub (crate) context_handle_active: RwLock>, - pub (crate) worker_shutdown: Mutex>, - pub (crate) context_shutdown: Mutex>>, + pub(crate) worker_shutdown: Mutex>, + pub(crate) context_shutdown: Mutex>>, } impl Default for WorkerHandleState { diff --git a/crates/ion/src/utils/complete_signal.rs b/crates/ion/src/utils/complete_signal.rs index 133768e..d172743 100644 --- a/crates/ion/src/utils/complete_signal.rs +++ b/crates/ion/src/utils/complete_signal.rs @@ -1,4 +1,7 @@ -use std::sync::{Arc, Condvar, Mutex}; +use std::sync::Arc; +use std::sync::Condvar; +use std::sync::Mutex; + use tokio::sync::Notify; #[derive(Clone, Default)] @@ -11,8 +14,7 @@ impl std::fmt::Debug for CompleteSignal { &self, f: &mut std::fmt::Formatter<'_>, ) -> std::fmt::Result { - f.debug_struct("CompleteSignal") - .finish() + f.debug_struct("CompleteSignal").finish() } } @@ -80,10 +82,11 @@ impl CompleteSignal { #[cfg(test)] mod tests { - use super::*; use std::thread; use std::time::Duration; + use super::*; + #[tokio::test] async fn test_complete_signal() { let sig = CompleteSignal::default(); diff --git a/crates/ion/src/utils/mod.rs b/crates/ion/src/utils/mod.rs index 122bbaa..6e4bd30 100644 --- a/crates/ion/src/utils/mod.rs +++ b/crates/ion/src/utils/mod.rs @@ -1,4 +1,5 @@ pub mod channel; +pub mod complete_signal; pub mod debug; pub mod hash; pub mod hash_map_ext; @@ -8,7 +9,6 @@ pub mod random_string; pub mod ref_counter; pub mod ref_counter_atomic; pub mod tokio_ext; -pub mod complete_signal; pub use debug::*; pub use hash::*; diff --git a/examples/src/basic/mod.rs b/examples/src/basic/mod.rs index 1a07431..4547388 100644 --- a/examples/src/basic/mod.rs +++ b/examples/src/basic/mod.rs @@ -22,5 +22,7 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; + Ok(()) } diff --git a/examples/src/basic_join/mod.rs b/examples/src/basic_join/mod.rs index 19c77dd..32bd888 100644 --- a/examples/src/basic_join/mod.rs +++ b/examples/src/basic_join/mod.rs @@ -143,8 +143,8 @@ fn should_wait_for_code_to_finish(runtime: Arc) -> anyhow::Result<()> c0.exec(non_blocking_exec(0))?; - c0.join_blocking()?; - w0.join_blocking()?; + c0.join()?; + w0.join()?; Ok(()) } @@ -156,8 +156,8 @@ fn should_wait_for_code_to_finish_multiple(runtime: Arc) -> anyhow::R c0.exec(non_blocking_exec(0))?; c0.exec(non_blocking_exec(0))?; - c0.join_blocking()?; - w0.join_blocking()?; + c0.join()?; + w0.join()?; Ok(()) } @@ -168,8 +168,8 @@ fn should_wait_for_code_to_finish_blocking(runtime: Arc) -> anyhow::R c0.exec_blocking(non_blocking_exec(0))?; - c0.join_blocking()?; - w0.join_blocking()?; + c0.join()?; + w0.join()?; Ok(()) } @@ -180,7 +180,7 @@ fn should_wait_for_code_to_finish_worker(runtime: Arc) -> anyhow::Res c0.exec(non_blocking_exec(0))?; - w0.join_blocking()?; + w0.join()?; Ok(()) } @@ -191,7 +191,7 @@ fn should_wait_for_code_to_finish_worker_blocking(runtime: Arc) -> an c0.exec_blocking(non_blocking_exec(0))?; - w0.join_blocking()?; + w0.join()?; Ok(()) } @@ -202,7 +202,7 @@ fn should_wait_for_code_to_finish_context(runtime: Arc) -> anyhow::Re c0.exec(non_blocking_exec(0))?; - c0.join_blocking()?; + c0.join()?; Ok(()) } @@ -213,7 +213,7 @@ fn should_wait_for_code_to_finish_context_blocking(runtime: Arc) -> a c0.exec_blocking(non_blocking_exec(0))?; - c0.join_blocking()?; + c0.join()?; Ok(()) } @@ -227,8 +227,8 @@ fn should_wait_for_code_to_finish_contexts(runtime: Arc) -> anyhow::R c0.exec(non_blocking_exec(0))?; c1.exec(non_blocking_exec(0))?; - c0.join_blocking()?; - c1.join_blocking()?; + c0.join()?; + c1.join()?; Ok(()) } @@ -242,8 +242,8 @@ fn should_wait_for_code_to_finish_contexts_blocking(runtime: Arc) -> c0.exec_blocking(non_blocking_exec(0))?; c1.exec_blocking(non_blocking_exec(1))?; - c0.join_blocking()?; - c1.join_blocking()?; + c0.join()?; + c1.join()?; Ok(()) } @@ -254,11 +254,11 @@ fn should_not_run_code_after_joining(runtime: Arc) -> anyhow::Result< c0.exec_blocking(non_blocking_exec(0))?; - w0.join_blocking()?; + w0.join()?; if c0.exec_blocking(non_blocking_exec(0)).is_err() { Report::print(None, None, "did_not_run"); }; - + Ok(()) } From 8b25374c6a0968786c60b3ac8625af66c33abe98 Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 5 Jan 2026 17:18:58 +1100 Subject: [PATCH 5/6] tests passing --- crates/ion/src/js_context.rs | 11 ++++- crates/ion/src/js_worker.rs | 3 ++ crates/ion/src/platform/worker.rs | 8 +++- .../ion/src/platform/worker_handle_state.rs | 43 ++++++++++--------- examples/src/background_tasks/mod.rs | 2 + examples/src/custom_extension/mod.rs | 2 + examples/src/custom_resolver/mod.rs | 1 + examples/src/deferred/mod.rs | 1 + examples/src/eval/mod.rs | 2 + examples/src/external_value/mod.rs | 1 + examples/src/memory_usage_context/mod.rs | 8 ++-- examples/src/multiple_workers/mod.rs | 4 ++ examples/src/promise/mod.rs | 1 + examples/src/run/mod.rs | 1 + examples/src/set_interval/mod.rs | 2 + examples/src/set_timeout/mod.rs | 2 + examples/src/thread_safe_function/mod.rs | 1 + examples/src/thread_safe_promise/mod.rs | 1 + examples/src/transformers/mod.rs | 1 + examples/src/typescript/mod.rs | 1 + 20 files changed, 69 insertions(+), 27 deletions(-) diff --git a/crates/ion/src/js_context.rs b/crates/ion/src/js_context.rs index 9c35277..d5f6c8d 100644 --- a/crates/ion/src/js_context.rs +++ b/crates/ion/src/js_context.rs @@ -95,6 +95,9 @@ impl JsContext { return Err(crate::Error::WorkerAlreadyShutdown); } + self.worker_handle_state + .context_handle_set_status(&self.id, false); + if self .tx .send(JsWorkerEvent::ContextHandleDeactivated { @@ -111,11 +114,14 @@ impl JsContext { } /// Wait for the context to complete all activity - pub async fn join_async(&self) -> crate::Result<()> { + pub async fn join_async(self) -> crate::Result<()> { if !self.worker_handle_state.worker_handle_active() { return Err(crate::Error::WorkerAlreadyShutdown); } + self.worker_handle_state + .context_handle_set_status(&self.id, false); + if self .tx .send(JsWorkerEvent::ContextHandleDeactivated { @@ -134,6 +140,9 @@ impl JsContext { impl Drop for JsContext { fn drop(&mut self) { + self.worker_handle_state + .context_handle_set_status(&self.id, false); + drop( self.tx .try_send(JsWorkerEvent::ContextHandleDropped { id: self.id }), diff --git a/crates/ion/src/js_worker.rs b/crates/ion/src/js_worker.rs index d2799f8..26fde70 100644 --- a/crates/ion/src/js_worker.rs +++ b/crates/ion/src/js_worker.rs @@ -73,6 +73,9 @@ impl JsWorker { return Err(Error::WorkerInitializeError); }; + self.worker_handle_state + .context_handle_set_status(&id, true); + Ok(JsContext { id, tx, diff --git a/crates/ion/src/platform/worker.rs b/crates/ion/src/platform/worker.rs index 92812c3..a3294db 100644 --- a/crates/ion/src/platform/worker.rs +++ b/crates/ion/src/platform/worker.rs @@ -138,6 +138,12 @@ fn worker_thread( panic!("Callback errored {:?}", err) }; + if worker_handle_state.worker_handle_active() + && worker_handle_state.context_handle_active(&id) + { + continue; + } + if realm.global_refs.count() != 0 { continue; } @@ -229,7 +235,7 @@ fn worker_thread( } } - if realms.len() == 0 && !worker_handle_state.worker_handle_active() { + if !worker_handle_state.worker_handle_active() && realms.len() == 0 { break; } } diff --git a/crates/ion/src/platform/worker_handle_state.rs b/crates/ion/src/platform/worker_handle_state.rs index 1cc0d57..caa7eca 100644 --- a/crates/ion/src/platform/worker_handle_state.rs +++ b/crates/ion/src/platform/worker_handle_state.rs @@ -3,13 +3,14 @@ use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use parking_lot::Mutex; +use parking_lot::RwLock; pub type WorkerShutdownCallback = Box; pub type ContextShutdownCallback = Box; pub struct WorkerHandleState { pub(crate) worker_handle_active: AtomicBool, - // pub (crate) context_handle_active: RwLock>, + pub(crate) context_handle_active: RwLock>, pub(crate) worker_shutdown: Mutex>, pub(crate) context_shutdown: Mutex>>, } @@ -18,7 +19,7 @@ impl Default for WorkerHandleState { fn default() -> Self { Self { worker_handle_active: AtomicBool::new(true), - // context_handle_active: Default::default(), + context_handle_active: Default::default(), worker_shutdown: Default::default(), context_shutdown: Default::default(), } @@ -34,26 +35,26 @@ impl WorkerHandleState { self.worker_handle_active.swap(false, Ordering::Relaxed); } - // pub(crate) fn context_handle_active( - // &self, - // id: &usize, - // ) -> bool { - // self.context_handle_active - // .read() - // .get(id) - // .unwrap_or(&false) - // .clone() - // } + pub(crate) fn context_handle_active( + &self, + id: &usize, + ) -> bool { + self.context_handle_active + .read() + .get(id) + .unwrap_or(&false) + .clone() + } - // pub(crate) fn context_handle_set_status( - // &self, - // id: &usize, - // status: bool, - // ) { - // self.context_handle_active - // .write() - // .insert(id.clone(), status); - // } + pub(crate) fn context_handle_set_status( + &self, + id: &usize, + status: bool, + ) { + self.context_handle_active + .write() + .insert(id.clone(), status); + } // pub(crate) fn add_worker_shutdown_callback( // &self, diff --git a/examples/src/background_tasks/mod.rs b/examples/src/background_tasks/mod.rs index a83c75d..b40b7e3 100644 --- a/examples/src/background_tasks/mod.rs +++ b/examples/src/background_tasks/mod.rs @@ -54,5 +54,7 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; + Ok(()) } diff --git a/examples/src/custom_extension/mod.rs b/examples/src/custom_extension/mod.rs index 1c4a9d7..edbaaae 100644 --- a/examples/src/custom_extension/mod.rs +++ b/examples/src/custom_extension/mod.rs @@ -24,6 +24,8 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; + Ok(()) } diff --git a/examples/src/custom_resolver/mod.rs b/examples/src/custom_resolver/mod.rs index a21cc5d..c4c9e8c 100644 --- a/examples/src/custom_resolver/mod.rs +++ b/examples/src/custom_resolver/mod.rs @@ -23,6 +23,7 @@ pub fn main() -> anyhow::Result<()> { ctx.import(&entry_point)?; + ctx.join()?; Ok(()) } diff --git a/examples/src/deferred/mod.rs b/examples/src/deferred/mod.rs index 26b2f3f..4c69977 100644 --- a/examples/src/deferred/mod.rs +++ b/examples/src/deferred/mod.rs @@ -63,5 +63,6 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; Ok(()) } diff --git a/examples/src/eval/mod.rs b/examples/src/eval/mod.rs index 323e226..19f31b9 100644 --- a/examples/src/eval/mod.rs +++ b/examples/src/eval/mod.rs @@ -28,5 +28,7 @@ pub fn main() -> anyhow::Result<()> { let ctx = worker.create_context()?; ctx.eval(code)?; + ctx.join()?; + Ok(()) } diff --git a/examples/src/external_value/mod.rs b/examples/src/external_value/mod.rs index bc25342..d0ebc57 100644 --- a/examples/src/external_value/mod.rs +++ b/examples/src/external_value/mod.rs @@ -40,5 +40,6 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; Ok(()) } diff --git a/examples/src/memory_usage_context/mod.rs b/examples/src/memory_usage_context/mod.rs index e02c187..316e253 100644 --- a/examples/src/memory_usage_context/mod.rs +++ b/examples/src/memory_usage_context/mod.rs @@ -34,7 +34,7 @@ pub fn main() -> anyhow::Result<()> { let ctx1 = worker.create_context()?; ctx0.eval("globalThis.value = []")?; - for i in 0..100 { + for i in 0..1 { ctx0.eval(format!("globalThis.value.push({})", i))?; } @@ -43,12 +43,12 @@ pub fn main() -> anyhow::Result<()> { ctx1.eval(format!("globalThis.value.push({})", i))?; } - drop(ctx0); - drop(ctx1); + ctx0.join()?; + ctx1.join()?; }; worker.run_garbage_collection_for_testing()?; - drop(worker); + worker.join()?; println!("{}", memu.megabytes().json()); } diff --git a/examples/src/multiple_workers/mod.rs b/examples/src/multiple_workers/mod.rs index 9ae7b7a..65b396b 100644 --- a/examples/src/multiple_workers/mod.rs +++ b/examples/src/multiple_workers/mod.rs @@ -30,5 +30,9 @@ pub fn main() -> anyhow::Result<()> { wrk2ctx1.eval("console.log('wrk2ctx1')")?; wrk3ctx1.eval("console.log('wrk3ctx1')")?; + wrk1ctx1.join()?; + wrk2ctx1.join()?; + wrk3ctx1.join()?; + Ok(()) } diff --git a/examples/src/promise/mod.rs b/examples/src/promise/mod.rs index 5ab3b8a..b51a9c0 100644 --- a/examples/src/promise/mod.rs +++ b/examples/src/promise/mod.rs @@ -55,5 +55,6 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; Ok(()) } diff --git a/examples/src/run/mod.rs b/examples/src/run/mod.rs index dc5e6d1..de58f77 100644 --- a/examples/src/run/mod.rs +++ b/examples/src/run/mod.rs @@ -43,6 +43,7 @@ pub fn main() -> anyhow::Result<()> { let ctx = worker.create_context()?; ctx.import(file_path.try_to_string()?)?; + ctx.join()?; Ok(()) } diff --git a/examples/src/set_interval/mod.rs b/examples/src/set_interval/mod.rs index 4eb79a9..2359f52 100644 --- a/examples/src/set_interval/mod.rs +++ b/examples/src/set_interval/mod.rs @@ -40,5 +40,7 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + + ctx.join()?; Ok(()) } diff --git a/examples/src/set_timeout/mod.rs b/examples/src/set_timeout/mod.rs index 2275acf..562b7e5 100644 --- a/examples/src/set_timeout/mod.rs +++ b/examples/src/set_timeout/mod.rs @@ -42,5 +42,7 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + + ctx.join()?; Ok(()) } diff --git a/examples/src/thread_safe_function/mod.rs b/examples/src/thread_safe_function/mod.rs index b8c594e..86e6c17 100644 --- a/examples/src/thread_safe_function/mod.rs +++ b/examples/src/thread_safe_function/mod.rs @@ -78,5 +78,6 @@ pub fn main() -> anyhow::Result<()> { Ok(()) })?; + ctx.join()?; Ok(()) } diff --git a/examples/src/thread_safe_promise/mod.rs b/examples/src/thread_safe_promise/mod.rs index f5eeba1..2550f0a 100644 --- a/examples/src/thread_safe_promise/mod.rs +++ b/examples/src/thread_safe_promise/mod.rs @@ -47,5 +47,6 @@ pub fn main() -> anyhow::Result<()> { println!("[Rust] Got {}", result); + ctx.join()?; Ok(()) } diff --git a/examples/src/transformers/mod.rs b/examples/src/transformers/mod.rs index cc9fdbb..83e1f2e 100644 --- a/examples/src/transformers/mod.rs +++ b/examples/src/transformers/mod.rs @@ -34,5 +34,6 @@ pub fn main() -> anyhow::Result<()> { ctx.exec_blocking(move |env| env.import(entry_point.try_to_string()?))?; + ctx.join()?; Ok(()) } diff --git a/examples/src/typescript/mod.rs b/examples/src/typescript/mod.rs index 3e8ae40..3e68663 100644 --- a/examples/src/typescript/mod.rs +++ b/examples/src/typescript/mod.rs @@ -34,5 +34,6 @@ pub fn main() -> anyhow::Result<()> { ctx.exec_blocking(move |env| env.import(entry_point.try_to_string()?))?; + ctx.join()?; Ok(()) } From 4a85979f7549d1b606c4424a1bfee743a5340abf Mon Sep 17 00:00:00 2001 From: David Alsh Date: Mon, 5 Jan 2026 17:33:33 +1100 Subject: [PATCH 6/6] background tasks need to trigger context/worker purge --- crates/ion/src/platform/realm.rs | 8 ++++-- crates/ion/src/platform/worker.rs | 43 +++++++++++++++++++++++++------ 2 files changed, 41 insertions(+), 10 deletions(-) diff --git a/crates/ion/src/platform/realm.rs b/crates/ion/src/platform/realm.rs index 6c16bce..f638985 100644 --- a/crates/ion/src/platform/realm.rs +++ b/crates/ion/src/platform/realm.rs @@ -33,6 +33,7 @@ pub struct JsRealm { pub(crate) modules: ModuleMap, pub(crate) global_this: sys::GlobalThis, pub(crate) context_shutdown_sig: CompleteSignal, + pub(crate) tx: Sender, } impl JsRealm { @@ -76,6 +77,7 @@ impl JsRealm { finalizer_registry, global_this, context_shutdown_sig, + tx, }); let realm_ptr = realm.as_mut() as *mut JsRealm; @@ -113,11 +115,13 @@ impl JsRealm { &self, fut: impl 'static + Send + Sync + Future>, ) -> crate::Result<()> { + let tx = self.tx.clone(); + let id = self.id; self.background_task_manager.spawn(async move { if let Err(_error) = fut.await { todo!("Missing global error handler") }; - Ok(()) + Ok(tx.try_send(JsWorkerEvent::BackgroundTaskComplete { id })?) }) } @@ -127,7 +131,7 @@ impl JsRealm { ) -> crate::Result { let (tx, rx) = oneshot(); self.background_task_manager.spawn(async move { - tx.try_send(fut.await).unwrap(); + tx.try_send(fut.await).expect("Unable to resolve"); Ok(()) })?; rx.recv()? diff --git a/crates/ion/src/platform/worker.rs b/crates/ion/src/platform/worker.rs index a3294db..d79b586 100644 --- a/crates/ion/src/platform/worker.rs +++ b/crates/ion/src/platform/worker.rs @@ -49,6 +49,9 @@ pub(crate) enum JsWorkerEvent { ContextHandleDeactivated { id: usize, }, + BackgroundTaskComplete { + id: usize, + }, } // Create a dedicated thread to host the isolate @@ -158,6 +161,29 @@ fn worker_thread( realm.context_shutdown_sig.done(); } + JsWorkerEvent::BackgroundTaskComplete { id } => { + let realm = realms.try_get(&id)?; + + if worker_handle_state.worker_handle_active() + && worker_handle_state.context_handle_active(&id) + { + continue; + } + + if realm.global_refs.count() != 0 { + continue; + } + + let Some(realm) = realms.remove(&id) else { + continue; + }; + + let finalizer_registry = realm.finalizer_registry; + finalizer_registry.clear(); + drop(finalizer_registry); + + realm.context_shutdown_sig.done(); + } JsWorkerEvent::Import { id, specifier } => { Module::v8_initialize( true, @@ -253,14 +279,15 @@ impl std::fmt::Debug for JsWorkerEvent { f: &mut std::fmt::Formatter<'_>, ) -> std::fmt::Result { match self { - Self::CreateContext { resolve, context_shutdown_sig } => write!(f, "CreateContext"), - Self::Exec { id, callback, span } => write!(f, "Exec [id={}]", id), - Self::Import { id, specifier } => write!(f, "Import"), - Self::WorkerHandleDropped => write!(f, "WorkerHandleDropped"), - Self::WorkerHandleDeactivated => write!(f, "WorkerHandleDeactivated"), - Self::ContextHandleDropped { id } => write!(f, "ContextHandleDropped"), - Self::ContextHandleDeactivated { id } => write!(f, "ContextHandleDeactivated [id={}]", id), - Self::RunGarbageCollectionForTesting { resolve } => write!(f, "RunGarbageCollectionForTesting"), + Self::CreateContext { resolve, context_shutdown_sig } => write!(f, "CreateContext"), + Self::Exec { id, callback, span } => write!(f, "Exec [id={}]", id), + Self::BackgroundTaskComplete { id } => write!(f, "BackgroundTaskComplete [id={}]", id), + Self::Import { id, specifier } => write!(f, "Import"), + Self::WorkerHandleDropped => write!(f, "WorkerHandleDropped"), + Self::WorkerHandleDeactivated => write!(f, "WorkerHandleDeactivated"), + Self::ContextHandleDropped { id } => write!(f, "ContextHandleDropped"), + Self::ContextHandleDeactivated { id } => write!(f, "ContextHandleDeactivated [id={}]", id), + Self::RunGarbageCollectionForTesting { resolve } => write!(f, "RunGarbageCollectionForTesting"), } } }