Merge pull request #23 from NativeScript/feat/worker-transfer

Transfer lists in worker postMessage, native addons in workers

Author: triniwiz

napi-v8-shim/csrc/env_ext.cpp

@@ -14,6 +14,9 @@ static_assert(sizeof(v8::Local<v8::Context>) == sizeof(void*), "v8::Local must b
 // From node_api_types.h, which the vendored headers don't carry.
 typedef napi_value (*napi_addon_register_func)(napi_env env, napi_value exports);
 
+// v8-api.cpp.
+void ns_napi_finalize_external_arraybuffers(napi_env env);
+
 extern "C" {
 
 // `context` is a `v8::Local<v8::Context>` (as its underlying pointer) of the isolate current on
@@ -53,14 +56,16 @@ bool ns_napi_has_pending_finalizers(napi_env env) {
     return !env->pending_finalizers.empty();
 }
 
-// Tears the env down: finalizes remaining references (running their finalizers) and frees it.
+// Tears the env down: finalizes remaining references and external ArrayBuffers (running their
+// finalizers) and frees it.
 void ns_napi_env_teardown(napi_env env) {
     v8::Isolate* isolate = env->isolate;
     v8::HandleScope handle_scope(isolate);
     // A real handle, not env->context(): that aliases the env's persistent slot, which DeleteMe
     // frees before this scope exits.
     v8::Local<v8::Context> context = v8::Local<v8::Context>::New(isolate, env->context());
     v8::Context::Scope context_scope(context);
+    ns_napi_finalize_external_arraybuffers(env);
     env->DeleteMe();
 }
 

packages/windows-v8/vendor/shim/v8-api.cpp

@@ -3,6 +3,10 @@
 #include <cmath>
 #include <string_view> // string_view, u16string_view
 #include <sstream>
+#include <mutex>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
 
 #define NAPI_EXPERIMENTAL
 
@@ -2995,6 +2999,41 @@ const v8::ArrayBuffer *v8__ArrayBuffer__New__with_backing_store(
 void std__shared_ptr__v8__BackingStore__reset(void *shared_ptr_ref);
 }
 
+namespace {
+// An external ArrayBuffer's finalizer. Its backing store can outlive the env (it goes with the
+// isolate's heap), so the env's teardown runs it instead and clears `cb`, as Node does.
+struct AbFinalize {
+    napi_env env;
+    napi_finalize cb;
+    void *data;
+    void *hint;
+};
+
+std::mutex ab_finalizers_mutex;
+std::unordered_map<napi_env, std::unordered_set<AbFinalize *>> ab_finalizers;
+}
+
+// [windows port] Called by the env's teardown (napi-v8-shim/csrc/env_ext.cpp).
+void ns_napi_finalize_external_arraybuffers(napi_env env) {
+    // Copies: once `cb` is cleared, a deleter on another thread may free the record.
+    std::vector<AbFinalize> pending;
+    {
+        std::lock_guard<std::mutex> lock(ab_finalizers_mutex);
+        auto found = ab_finalizers.find(env);
+        if (found == ab_finalizers.end()) {
+            return;
+        }
+        for (AbFinalize *fd : found->second) {
+            pending.push_back(*fd);
+            fd->cb = nullptr;
+        }
+        ab_finalizers.erase(found);
+    }
+    for (const AbFinalize &fd : pending) {
+        fd.cb(env, fd.data, fd.hint);
+    }
+}
+
 napi_status NAPI_CDECL
 napi_create_external_arraybuffer(napi_env env,
                                  void *external_data,
@@ -3010,19 +3049,30 @@ napi_create_external_arraybuffer(napi_env env,
     // [windows port] TRUE zero-copy: build a BackingStore that aliases external_data with a deleter
     // that invokes the napi finalizer when the ArrayBuffer is GC'd. Goes through rusty_v8's C
     // bindings (see above) to avoid the libc++ std::unique_ptr ABI boundary.
-    struct AbFinalize {
-        napi_env env;
-        napi_finalize cb;
-        void *hint;
-    };
     void *deleter_data = nullptr;
-    void (*deleter)(void *, size_t, void *) = nullptr;
+    // V8 calls the deleter unconditionally when the backing store goes (at the latest when the
+    // isolate is disposed, as a terminated worker's is), so it can't be null.
+    void (*deleter)(void *, size_t, void *) = [](void *, size_t, void *) {};
     if (finalize_cb != nullptr) {
-        deleter_data = new AbFinalize{env, finalize_cb, finalize_hint};
+        auto *fd = new AbFinalize{env, finalize_cb, external_data, finalize_hint};
+        {
+            std::lock_guard<std::mutex> lock(ab_finalizers_mutex);
+            ab_finalizers[env].insert(fd);
+        }
+        deleter_data = fd;
         deleter = [](void *data, size_t, void *dd) {
             auto *fd = static_cast<AbFinalize *>(dd);
-            if (fd->cb) {
-                fd->cb(fd->env, data, fd->hint);
+            napi_finalize cb;
+            {
+                std::lock_guard<std::mutex> lock(ab_finalizers_mutex);
+                cb = fd->cb;
+                auto found = ab_finalizers.find(fd->env);
+                if (cb && found != ab_finalizers.end()) {
+                    found->second.erase(fd);
+                }
+            }
+            if (cb) {
+                cb(fd->env, data, fd->hint);
             }
             delete fd;
         };

runtime/src/animation_frames.rs

@@ -5,23 +5,66 @@
 //! once per compositor frame from `CompositionTarget.Rendering`, outside the render walk) then
 //! runs the queued callbacks once and drains microtasks, which is where rendering work such as
 //! canvas presents happens. Nothing waits for vsync on the UI thread, and a continuous rAF loop
-//! gives the dispatcher back between frames.
+//! gives the dispatcher back between frames. A worker's frames follow the UI thread's: each pump
+//! there hands a frame to every worker that asked for one. The compositor stops raising frames
+//! while nothing on screen changes, so a worker not handed one in time takes its own.
 
 use std::cell::Cell;
+use std::time::{Duration, Instant};
 
 use crate::DELEGATE_ISOLATE_PTR;
 
 thread_local! {
     static REQUESTED: Cell<bool> = const { Cell::new(false) };
+    static LAST_FRAME: Cell<Option<Instant>> = const { Cell::new(None) };
 }
 
+#[cfg(feature = "classic")]
+const WORKER_FRAME_INTERVAL: Duration = Duration::from_micros(16_667);
+
+#[cfg(feature = "classic")]
+pub(crate) fn worker_frame_wait() -> Option<Duration> {
+    if !REQUESTED.with(|r| r.get()) {
+        return None;
+    }
+    let last = LAST_FRAME.with(|l| l.get());
+    Some(last.map_or(Duration::ZERO, |last| WORKER_FRAME_INTERVAL.saturating_sub(last.elapsed())))
+}
+
+#[cfg(feature = "classic")]
+static WORKERS_REQUESTED: std::sync::Mutex<Vec<usize>> = std::sync::Mutex::new(Vec::new());
+
 /// `__nsRequestFrame()`: run animation callbacks at the next pump.
 pub(crate) fn handle_request_frame(
     _scope: &mut v8::PinScope<'_, '_>,
     _args: v8::FunctionCallbackArguments,
     _retval: v8::ReturnValue,
 ) {
     REQUESTED.with(|r| r.set(true));
+    #[cfg(feature = "classic")]
+    {
+        let isolate = DELEGATE_ISOLATE_PTR.with(|c| c.get()) as usize;
+        if crate::worker_threads::is_worker_isolate(isolate) {
+            let mut workers = WORKERS_REQUESTED.lock().unwrap_or_else(|e| e.into_inner());
+            if !workers.contains(&isolate) {
+                workers.push(isolate);
+            }
+        }
+    }
+}
+
+#[cfg(feature = "classic")]
+fn hand_frames_to_workers() {
+    let isolate = DELEGATE_ISOLATE_PTR.with(|c| c.get()) as usize;
+    if crate::worker_threads::is_worker_isolate(isolate) {
+        return;
+    }
+    let workers = std::mem::take(&mut *WORKERS_REQUESTED.lock().unwrap_or_else(|e| e.into_inner()));
+    for worker in workers {
+        crate::worker_threads::post_to_worker(worker, || {
+            pump();
+        });
+    }
 }
 
 /// Drops a pending request (the runtime on this thread is going away).
@@ -40,9 +83,12 @@ fn now_ms() -> f64 {
 /// Runs this thread's pending animation-frame callbacks, if a frame was requested, then drains
 /// microtasks. Returns whether callbacks ran.
 pub fn pump() -> bool {
+    #[cfg(feature = "classic")]
+    hand_frames_to_workers();
     if !REQUESTED.with(|r| r.replace(false)) {
         return false;
     }
+    LAST_FRAME.with(|l| l.set(Some(Instant::now())));
     let isolate_ptr = DELEGATE_ISOLATE_PTR.with(|c| c.get());
     if isolate_ptr.is_null() {
         return false;

runtime/src/global_fns.rs

@@ -14,7 +14,7 @@ use windows::Win32::UI::WindowsAndMessaging::{
 
 use crate::type_description::build_runtime_type_descriptor;
 use crate::dotnet::{bin_write_str16, bin_write_str32};
-use crate::{normalize_js_path, proxy_manifests, throw_js_error, try_resolve_with_known_extensions, Runtime, ASYNC_PUMP_HOOK};
+use crate::{normalize_js_path, proxy_manifests, throw_js_error, try_resolve_with_known_extensions, ASYNC_PUMP_HOOK};
 use std::cell::RefCell;
 use std::ffi::c_void;
 
@@ -209,23 +209,29 @@ fn default_auto_capture_path() -> PathBuf {
     PathBuf::from("sbg_output").join("sbg_metadata.json")
 }
 
+/// Delivered as a `messageerror`.
+fn worker_error<'s>(scope: &mut v8::PinScope<'s, '_>, error: &str) -> Option<v8::Local<'s, v8::Value>> {
+    let obj = v8::Object::new(scope);
+    if let Some(key) = v8::String::new(scope, "__workerError") {
+        if let Some(val) = v8::String::new(scope, error) {
+            obj.set(scope, key.into(), val.into());
+        }
+    }
+    Some(obj.into())
+}
+
 fn polled_event_to_v8<'s>(
     scope: &mut v8::PinScope<'s, '_>,
     event: crate::worker_threads::PolledWorkerEvent,
 ) -> Option<v8::Local<'s, v8::Value>> {
     match event {
-        crate::worker_threads::PolledWorkerEvent::Message(bytes) => {
-            Runtime::deserialize_value(scope, &bytes)
-        }
-        crate::worker_threads::PolledWorkerEvent::Error(error) => {
-            let obj = v8::Object::new(scope);
-            if let Some(key) = v8::String::new(scope, "__workerError") {
-                if let Some(val) = v8::String::new(scope, error.as_str()) {
-                    obj.set(scope, key.into(), val.into());
-                }
+        crate::worker_threads::PolledWorkerEvent::Message(message) => {
+            match crate::transfer::deserialize(scope, message) {
+                Ok(value) => Some(value),
+                Err(error) => worker_error(scope, &error),
             }
-            Some(obj.into())
         }
+        crate::worker_threads::PolledWorkerEvent::Error(error) => worker_error(scope, &error),
         crate::worker_threads::PolledWorkerEvent::Exited => {
             let obj = v8::Object::new(scope);
             if let Some(key) = v8::String::new(scope, "__workerExit") {
@@ -919,6 +925,11 @@ pub(crate) fn handle_resolve_module_path(
         throw_js_error(scope, "__nsResolveModulePath: module specifier is empty");
         return;
     }
+    // The app directory, as NativeScript's webpack worker loader names worker chunks.
+    let specifier = match specifier.strip_prefix("~/") {
+        Some(rest) => rest.to_string(),
+        None => specifier,
+    };
     let parent_path = if args.length() >= 2 {
         value_to_string(scope, args.get(1))
     } else {
@@ -1061,7 +1072,7 @@ pub(crate) fn handle_worker_post_message(
     if args.length() < 2 {
         throw_js_error(
             scope,
-            "__nsWorkerPostMessage(workerId, value) expects 2 arguments",
+            "__nsWorkerPostMessage(workerId, value, transfer?) expects 2 arguments",
         );
         return;
     }
@@ -1070,16 +1081,38 @@ pub(crate) fn handle_worker_post_message(
         throw_js_error(scope, "Invalid worker id");
         return;
     }
-    let value = args.get(1);
-    let Some(bytes) = Runtime::serialize_value(scope, value) else {
-        throw_js_error(scope, "DataCloneError: value could not be cloned.");
+    // On failure the exception is pending.
+    let Some(message) = crate::transfer::serialize(scope, args.get(1), args.get(2)) else {
         return;
     };
-    if let Err(err) = crate::worker_threads::post_message(worker_id as u64, bytes) {
+    if let Err(err) = crate::worker_threads::post_message(worker_id as u64, message) {
         throw_js_error(scope, err.as_str());
     }
 }
 
+/// A dispatcher the runtime made itself only runs when its host pumps messages: there a worker's
+/// replies only arrive by polling.
+pub(crate) fn handle_worker_pushes_messages(
+    _scope: &mut v8::PinScope<'_, '_>,
+    _args: v8::FunctionCallbackArguments,
+    mut retval: v8::ReturnValue,
+) {
+    let isolate = crate::DELEGATE_ISOLATE_PTR.with(|c| c.get()) as usize;
+    let xaml_host = crate::ui_dispatcher::is_initialized() && !crate::ui_dispatcher::needs_win32_pump();
+    retval.set_bool(crate::worker_threads::is_worker_isolate(isolate) || xaml_host);
+}
+
+/// Cloned and transferred now, as on the web; sent once the current job is done.
+pub(crate) fn handle_worker_queue_message(
+    scope: &mut v8::PinScope<'_, '_>,
+    args: v8::FunctionCallbackArguments,
+    _retval: v8::ReturnValue,
+) {
+    if let Some(message) = crate::transfer::serialize(scope, args.get(0), args.get(1)) {
+        crate::worker_threads::queue_outgoing(message);
+    }
+}
+
 /// Runs on a worker's creating thread when the worker has queued messages: hands them to that
 /// `Worker` object (`__nsWorkerDeliver`, installed by the Worker shim).
 pub(crate) fn deliver_worker_events(worker_id: u64) {
@@ -5788,6 +5821,8 @@ pub(crate) fn init_async_helpers(
     register!("__nsDescribeWinRTType", handle_describe_winrt_type);
     register!("__nsWorkerCreateThreaded", handle_worker_create_threaded);
     register!("__nsWorkerPostMessage", handle_worker_post_message);
+    register!("__nsWorkerQueueMessage", handle_worker_queue_message);
+    register!("__nsWorkerPushesMessages", handle_worker_pushes_messages);
     register!("__nsWorkerPollMessages", handle_worker_poll_messages);
     register!("__nsWorkerTerminate", handle_worker_terminate);
     register!(
@@ -5879,6 +5914,7 @@ pub(crate) fn init_async_helpers(
     }
 
     crate::message_port::install_message_port_runtime(scope);
+    crate::transfer::install_transfer_runtime(scope);
     crate::worker_support::install_worker_runtime(scope);
     crate::hmr_support::install_hmr_support(scope);
     crate::livesync::install_livesync_support(scope);