From 27a67945d1b6e352c4985be15919c6dbef226dd2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 08:15:36 +0000 Subject: [PATCH 1/4] Implement hermes_napi_host and pass it to hermes_napi_create_env Provide the Phase 3 host integration for Hermes' first-party Node-API: - New HermesNapiHost.{hpp,cpp}: a mirror of the hermes_napi_host struct (pinned to HERMES_GIT_SHA) and a HostContext per React Native runtime, backed by a process-global 4-thread worker pool (post_work / cancel_work) and the runtime's CallInvoker behind a type-erased JS dispatcher (post_task and work completions). fatal_exception stringifies the error, logs and aborts; uv_loop and ref_loop/unref_loop stay null by design. Contexts are retained for the process lifetime because the env reads the struct during Runtime teardown after env cleanup hooks have run. - CxxNodeApiHostModule passes the host at env creation - before the addon's init runs, fixing init-time async work - and drops setCallInvoker. - Delete the RuntimeNodeApiAsync overrides: async work falls through to Hermes' implementation, so execute now runs on a worker thread instead of the JS thread, and thread-safe functions work for the first time. - tests/async: execute/complete thread-identity assertions, a gated blocking execute (deadlock-proof that execute is off the JS thread) and a deterministic cancel-of-running-work case. - tests/threadsafe-function: port of Node's test_threadsafe_function (pthread shim for uv threads, upstream assertions restored) plus JS-thread and never-inline supplements; re-enable the async_work_thread_safe_function example (its SIGABRT was the null host). - packages/host/tests: Catch2 suite exercising the worker pool, cancellation atomicity, post_task ordering/reentrancy and the teardown drop path on plain Linux, with a host-cpp-tests CI job mirroring weak-node-api-tests. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6xHHeMFF853z5R6th5CkF --- .changeset/hermes-napi-host-integration.md | 5 + .github/workflows/check.yml | 36 ++ apps/test-app/App.tsx | 5 +- docs/HOW-IT-WORKS.md | 2 +- eslint.config.js | 3 + packages/host/.gitignore | 3 + packages/host/android/CMakeLists.txt | 4 +- packages/host/cpp/CxxNodeApiHostModule.cpp | 37 +- packages/host/cpp/CxxNodeApiHostModule.hpp | 4 + packages/host/cpp/HermesNapiHost.cpp | 241 +++++++++++ packages/host/cpp/HermesNapiHost.hpp | 114 +++++ packages/host/cpp/RuntimeNodeApiAsync.cpp | 200 --------- packages/host/cpp/RuntimeNodeApiAsync.hpp | 24 -- packages/host/package.json | 3 + packages/host/scripts/generate-injector.mts | 1 - packages/host/src/node/cli/hermes.ts | 5 + packages/host/tests/CMakeLists.txt | 42 ++ packages/host/tests/test_hermes_napi_host.cpp | 345 +++++++++++++++ packages/node-addon-examples/src/index.ts | 7 +- .../node-addon-examples/tests/async/addon.c | 143 ++++++ .../node-addon-examples/tests/async/addon.js | 76 +++- .../tests/threadsafe-function/CMakeLists.txt | 26 ++ .../tests/threadsafe-function/addon.c | 407 ++++++++++++++++++ .../tests/threadsafe-function/addon.js | 287 ++++++++++++ .../tests/threadsafe-function/binding.gyp | 8 + .../tests/threadsafe-function/package.json | 14 + 26 files changed, 1790 insertions(+), 252 deletions(-) create mode 100644 .changeset/hermes-napi-host-integration.md create mode 100644 packages/host/cpp/HermesNapiHost.cpp create mode 100644 packages/host/cpp/HermesNapiHost.hpp delete mode 100644 packages/host/cpp/RuntimeNodeApiAsync.cpp delete mode 100644 packages/host/cpp/RuntimeNodeApiAsync.hpp create mode 100644 packages/host/tests/CMakeLists.txt create mode 100644 packages/host/tests/test_hermes_napi_host.cpp create mode 100644 packages/node-addon-examples/tests/threadsafe-function/CMakeLists.txt create mode 100644 packages/node-addon-examples/tests/threadsafe-function/addon.c create mode 100644 packages/node-addon-examples/tests/threadsafe-function/addon.js create mode 100644 packages/node-addon-examples/tests/threadsafe-function/binding.gyp create mode 100644 packages/node-addon-examples/tests/threadsafe-function/package.json diff --git a/.changeset/hermes-napi-host-integration.md b/.changeset/hermes-napi-host-integration.md new file mode 100644 index 00000000..5a3f632c --- /dev/null +++ b/.changeset/hermes-napi-host-integration.md @@ -0,0 +1,5 @@ +--- +"react-native-node-api": minor +--- + +Provide a `hermes_napi_host` implementation to the Hermes Node-API environments. This enables thread-safe functions (`napi_create_threadsafe_function` and friends) and moves `napi_async_work` execution onto a worker pool — previously the `execute` callback ran on the JavaScript thread, blocking it for the duration of the work. The host is also in place before an addon's module init runs, so async work and thread-safe functions can now be created during initialization. diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index cb845c5c..ec9d1fe8 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -143,6 +143,42 @@ jobs: cmake --build build ctest --test-dir build --output-on-failure working-directory: packages/weak-node-api + host-cpp-tests: + if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/next' || contains(github.event.pull_request.labels.*.name, 'host') + strategy: + fail-fast: false + matrix: + runner: + - ubuntu-latest + - windows-latest + - macos-latest + runs-on: ${{ matrix.runner }} + name: Host C++ tests (${{ matrix.runner }}) + steps: + - uses: actions/checkout@v4 + - uses: pnpm/action-setup@v4 + - uses: actions/setup-node@v6 + with: + node-version: lts/krypton + cache: pnpm + - name: Setup cpp tools + uses: aminya/setup-cpp@v1 + with: + clang-format: true + - name: ccache + uses: hendrikmuhs/ccache-action@v1.2 + with: + key: ${{ github.job }}-${{ runner.os }} + - run: pnpm install + - run: pnpm run build + - name: Prepare weak-node-api + run: pnpm --filter weak-node-api run prebuild:prepare + - name: Build and run react-native-node-api host C++ tests + run: | + cmake -S tests -B tests/build + cmake --build tests/build + ctest --test-dir tests/build --output-on-failure + working-directory: packages/host test-ios: if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/next' || contains(github.event.pull_request.labels.*.name, 'Apple 🍎') name: Test app (iOS) diff --git a/apps/test-app/App.tsx b/apps/test-app/App.tsx index 397409bb..3006b1fa 100644 --- a/apps/test-app/App.tsx +++ b/apps/test-app/App.tsx @@ -38,7 +38,10 @@ function loadTests({ )) { describe(suiteName, () => { for (const [exampleName, requireExample] of Object.entries(examples)) { - it(exampleName, async () => { + it(exampleName, async function () { + // Some examples (the threadsafe-function suite in particular) + // marshal thousands of values across threads. + this.timeout(30_000); const test = requireExample(); if (test instanceof Function) { const result = test(); diff --git a/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index b78f4686..3a5167b3 100644 --- a/docs/HOW-IT-WORKS.md +++ b/docs/HOW-IT-WORKS.md @@ -54,7 +54,7 @@ Hermes implements both halves of Node-API: the engine-specific functions (see [j - `ref_loop` / `unref_loop` — keep the event loop alive while a thread-safe function is referenced, modelling libuv's "ref" semantics. - `fatal_exception` and, for embedders that have one, a libuv loop pointer for `napi_get_uv_event_loop`. -`react-native-node-api` provides that struct, backed by React Native's `CallInvoker` for anything that has to land on the JavaScript thread and a worker pool for the rest. +`react-native-node-api` provides that struct (see `packages/host/cpp/HermesNapiHost.cpp`), backed by React Native's `CallInvoker` for anything that has to land on the JavaScript thread and a process-global worker pool (four threads, like libuv's default) for the rest. `ref_loop` / `unref_loop` and the libuv loop pointer are deliberately left null: React Native's JavaScript thread has no ref-counted event-loop lifetime to model, so thread-safe function ref/unref are tracked but inert, and `napi_get_uv_event_loop` returns `napi_generic_failure` as upstream documents for hosts without libuv. ## `my-app` regain control and call `add` diff --git a/eslint.config.js b/eslint.config.js index bbe15a20..7e4d196d 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -62,6 +62,9 @@ export default tseslint.config( }, globals: { ...globals.commonjs, + // Timers provided by React Native's runtime, where these files run. + setTimeout: "readonly", + setImmediate: "readonly", }, }, rules: { diff --git a/packages/host/.gitignore b/packages/host/.gitignore index 5ba3e2fe..42920839 100644 --- a/packages/host/.gitignore +++ b/packages/host/.gitignore @@ -18,3 +18,6 @@ android/build/ # Generated via `npm run generate-weak-node-api-injector` /cpp/WeakNodeApiInjector.cpp + +# C++ test build artifacts (see `npm run test:configure`) +/tests/build/ diff --git a/packages/host/android/CMakeLists.txt b/packages/host/android/CMakeLists.txt index 19ba1d03..2a45e96d 100644 --- a/packages/host/android/CMakeLists.txt +++ b/packages/host/android/CMakeLists.txt @@ -14,8 +14,8 @@ add_library(node-api-host SHARED ../cpp/WeakNodeApiInjector.cpp ../cpp/RuntimeNodeApi.cpp ../cpp/RuntimeNodeApi.hpp - ../cpp/RuntimeNodeApiAsync.cpp - ../cpp/RuntimeNodeApiAsync.hpp + ../cpp/HermesNapiHost.cpp + ../cpp/HermesNapiHost.hpp ) target_include_directories(node-api-host PRIVATE diff --git a/packages/host/cpp/CxxNodeApiHostModule.cpp b/packages/host/cpp/CxxNodeApiHostModule.cpp index 05745892..9a3ef8c7 100644 --- a/packages/host/cpp/CxxNodeApiHostModule.cpp +++ b/packages/host/cpp/CxxNodeApiHostModule.cpp @@ -1,27 +1,10 @@ #include "CxxNodeApiHostModule.hpp" #include "Logger.hpp" -#include "RuntimeNodeApiAsync.hpp" #include using namespace facebook; -// Declared by the vendored Hermes in API/napi/hermes_napi.h. We forward declare -// it here (rather than including that header) to avoid pulling in Hermes' own -// node_api.h alongside the weak-node-api copy already included transitively. -// -// The declaration must be `extern "C"`: since facebook/hermes#2106 (included in -// the pinned Hermes commit) the public hermes_napi.h wraps these entry points -// in `extern "C"`, so Hermes exports the unmangled C symbol. Without matching C -// linkage here the reference would be to the C++-mangled name and the app fails -// to link ("Undefined symbol: hermes_napi_create_env"). Passing host as nullptr -// is enough — async work / thread-safe functions will return failure until a -// host integration is wired up (Phase 3). -extern "C" { -struct hermes_napi_host; -napi_env hermes_napi_create_env(void *hermes_runtime, hermes_napi_host *host); -} - namespace callstack::react_native_node_api { CxxNodeApiHostModule::CxxNodeApiHostModule( @@ -31,6 +14,22 @@ CxxNodeApiHostModule::CxxNodeApiHostModule( MethodMetadata{1, &CxxNodeApiHostModule::requireNodeAddon}; callInvoker_ = std::move(jsInvoker); + + // The JS-thread dispatcher behind the hermes_napi_host integration: + // CallInvoker::invokeAsync is callable from any thread, never runs the + // function inline and delivers in order on the JS thread. The CallInvoker + // is captured weakly so tasks in flight during a runtime teardown are + // dropped instead of dispatched into a dead runtime. + hostContext_ = HostContext::create( + [weakInvoker = std::weak_ptr(callInvoker_)](std::function &&fn) { + if (auto invoker = weakInvoker.lock()) { + invoker->invokeAsync(std::move(fn)); + } else { + log_warning( + "NapiHost: dropping a task posted after runtime teardown"); + } + }); + HostContext::retainForProcessLifetime(hostContext_); } jsi::Value @@ -141,7 +140,8 @@ bool CxxNodeApiHostModule::initializeNodeModule(jsi::Runtime &rt, "create a Node-API environment"); abort(); } - addon.env = hermes_napi_create_env(hermes->getVMRuntimeUnsafe(), nullptr); + addon.env = + hermes_napi_create_env(hermes->getVMRuntimeUnsafe(), hostContext_->host()); assert(addon.env != nullptr); } napi_env env = addon.env; @@ -163,7 +163,6 @@ bool CxxNodeApiHostModule::initializeNodeModule(jsi::Runtime &rt, napi_set_named_property(env, global, addon.generatedName.data(), exports); assert(status == napi_ok); - callstack::react_native_node_api::setCallInvoker(env, callInvoker_); return true; } diff --git a/packages/host/cpp/CxxNodeApiHostModule.hpp b/packages/host/cpp/CxxNodeApiHostModule.hpp index 7be3e598..e71df553 100644 --- a/packages/host/cpp/CxxNodeApiHostModule.hpp +++ b/packages/host/cpp/CxxNodeApiHostModule.hpp @@ -5,6 +5,7 @@ #include #include "AddonLoaders.hpp" +#include "HermesNapiHost.hpp" namespace callstack::react_native_node_api { @@ -37,6 +38,9 @@ class JSI_EXPORT CxxNodeApiHostModule : public facebook::react::TurboModule { }; std::unordered_map nodeAddons_; std::shared_ptr callInvoker_; + // The hermes_napi_host integration passed to every env this module creates. + // Also retained process-wide, as the envs outlive this module on teardown. + std::shared_ptr hostContext_; using LoaderPolicy = PosixLoader; // FIXME: HACK: This is temporary workaround // for my lazyness (work on iOS and Android) diff --git a/packages/host/cpp/HermesNapiHost.cpp b/packages/host/cpp/HermesNapiHost.cpp new file mode 100644 index 00000000..8fb17250 --- /dev/null +++ b/packages/host/cpp/HermesNapiHost.cpp @@ -0,0 +1,241 @@ +#include "HermesNapiHost.hpp" +#include "Logger.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace callstack::react_native_node_api { +namespace { + +struct WorkItem { + // Identifies the HostContext that posted the item; matched together with + // workData on cancellation, since the pool is shared by all runtimes and a + // freed napi_async_work address could be reused by another env. + void *loopData = nullptr; + // The dispatcher may expire while an item is in flight (React Native + // reload); the completion is then dropped, which is safe because the env it + // targets is torn down with its runtime. + std::weak_ptr context; + void *workData = nullptr; + void (*execute)(void *work_data) = nullptr; + void (*complete)(void *work_data, napi_status status) = nullptr; +}; + +class WorkerPool { +public: + static WorkerPool &instance() { + // Deliberately leaked, with detached threads, like libuv's process-global + // thread pool: the pool must be able to outlive any single React Native + // runtime and there is no shutdown point at which joining would be safe. + static WorkerPool *pool = new WorkerPool(); + return *pool; + } + + void enqueue(WorkItem &&item) { + { + std::lock_guard lock(mutex_); + for (const WorkItem &queued : queue_) { + if (queued.loopData == item.loopData && + queued.workData == item.workData) { + // Queueing the same napi_async_work twice is undefined behavior in + // Node (libuv asserts); warn instead of crashing. + log_warning( + "NapiHost: napi_async_work %p was queued while already queued", + item.workData); + break; + } + } + queue_.push_back(std::move(item)); + } + cv_.notify_one(); + } + + bool tryRemove(void *loopData, void *workData, WorkItem &result) { + std::lock_guard lock(mutex_); + for (auto it = queue_.begin(); it != queue_.end(); ++it) { + if (it->loopData == loopData && it->workData == workData) { + result = std::move(*it); + queue_.erase(it); + return true; + } + } + return false; + } + +private: + // libuv's default thread pool size. Keep this below 5: the cancellation + // tests make cancel-while-queued deterministic by saturating the pool with + // 5 blocking jobs before queueing the item they cancel. + static constexpr size_t kThreadCount = 4; + + WorkerPool() { + for (size_t i = 0; i < kThreadCount; i++) { + std::thread([this] { workerMain(); }).detach(); + } + } + + void workerMain() { + for (;;) { + WorkItem item; + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return !queue_.empty(); }); + item = std::move(queue_.front()); + queue_.pop_front(); + } + // An item is either popped here (execute runs, complete gets napi_ok) + // or removed by tryRemove (complete gets napi_cancelled) — never both, + // as both happen under the queue mutex. + item.execute(item.workData); + if (auto context = item.context.lock()) { + context->dispatchToJs( + [workData = item.workData, complete = item.complete] { + // No pool state refers to workData at this point, so the + // complete callback is free to napi_delete_async_work it. + complete(workData, napi_ok); + }); + } else { + log_warning("NapiHost: dropping an async work completion posted after " + "runtime teardown"); + } + } + } + + std::mutex mutex_; + std::condition_variable cv_; + std::deque queue_; +}; + +std::optional stringValue(napi_env env, napi_value value) { + size_t length = 0; + if (napi_get_value_string_utf8(env, value, nullptr, 0, &length) != napi_ok) { + return std::nullopt; + } + std::string result(length, '\0'); + if (napi_get_value_string_utf8(env, value, result.data(), length + 1, + nullptr) != napi_ok) { + return std::nullopt; + } + return result; +} + +std::string describeError(napi_env env, napi_value err) { + // Prefer the error's stack (which includes its message), fall back to + // coercing the value to a string. Every call is status-checked: this runs + // right before an abort and must not assume anything about the value. + napi_value stack = nullptr; + napi_valuetype type = napi_undefined; + if (napi_get_named_property(env, err, "stack", &stack) == napi_ok && + napi_typeof(env, stack, &type) == napi_ok && type == napi_string) { + if (auto text = stringValue(env, stack)) { + return *text; + } + } + napi_value coerced = nullptr; + if (napi_coerce_to_string(env, err, &coerced) == napi_ok) { + if (auto text = stringValue(env, coerced)) { + return *text; + } + } + return "(unable to stringify the error value)"; +} + +} // namespace + +HostContext::HostContext(JsDispatcher dispatchToJs) + : dispatchToJs_(std::move(dispatchToJs)), + host_{ + .post_work = &HostContext::postWork, + // Hermes null-checks only the host pointer itself before invoking + // post_work and cancel_work, so neither may individually be null. + .cancel_work = &HostContext::cancelWork, + .post_task = &HostContext::postTask, + .data = this, + // React Native has no libuv loop: napi_get_uv_event_loop() returns + // napi_generic_failure, as upstream documents for non-Node hosts. + .uv_loop = nullptr, + .fatal_exception = &HostContext::fatalException, + // The JS thread outlives every producer thread, so there is no loop + // lifetime to model: tsfn ref/unref are tracked by Hermes but inert. + .ref_loop = nullptr, + .unref_loop = nullptr, + } {} + +std::shared_ptr HostContext::create(JsDispatcher dispatchToJs) { + return std::shared_ptr(new HostContext(std::move(dispatchToJs))); +} + +void HostContext::retainForProcessLifetime( + std::shared_ptr context) { + // Leaked for the same reason as the WorkerPool: no safe destruction point. + static std::mutex *mutex = new std::mutex(); + static auto *retained = new std::vector>(); + std::lock_guard lock(*mutex); + retained->push_back(std::move(context)); +} + +void HostContext::postWork(void *loop_data, void *work_data, + void (*execute)(void *work_data), + void (*complete)(void *work_data, + napi_status status)) noexcept { + auto *self = static_cast(loop_data); + // Called on the JS thread (napi_queue_async_work) while the env — and + // therefore this context — is alive, so weak_from_this() is populated. + WorkerPool::instance().enqueue(WorkItem{ + .loopData = loop_data, + .context = self->weak_from_this(), + .workData = work_data, + .execute = execute, + .complete = complete, + }); +} + +bool HostContext::cancelWork(void *loop_data, void *work_data) noexcept { + WorkItem item; + if (!WorkerPool::instance().tryRemove(loop_data, work_data, item)) { + // Already picked up by a worker (or never queued): cancellation failed + // and Hermes surfaces napi_generic_failure, like Node. + return false; + } + if (auto context = item.context.lock()) { + // Deliver the cancelled completion asynchronously, matching Node, where a + // cancelled complete callback still runs on a later loop tick. + context->dispatchToJs([workData = item.workData, complete = item.complete] { + complete(workData, napi_cancelled); + }); + return true; + } + // The runtime is being torn down; the complete callback can never run, so + // report the cancellation as failed. + return false; +} + +void HostContext::postTask(void *loop_data, void *task_data, + void (*callback)(void *task_data)) noexcept { + auto *self = static_cast(loop_data); + // Thread-safe functions call this from arbitrary producer threads, and + // Hermes' tsfnDispatch re-posts itself from inside the callback. The + // dispatcher never runs the callback inline (JS would run off-thread) and + // never drops it while the runtime is alive — a dropped dispatch would + // permanently wedge the tsfn, as its dispatch_pending flag stays set. + self->dispatchToJs_([task_data, callback] { callback(task_data); }); +} + +void HostContext::fatalException(void *, napi_env env, + napi_value err) noexcept { + // Called on the JS thread by napi_fatal_exception(). Node routes this to + // process.emit('uncaughtException'); with no process object we log the + // error and abort — the same observable outcome as Hermes' null-host + // default, but surfaced through the host logger. `err` is only valid for + // the duration of this call, so it is stringified before returning. + log_error("napi_fatal_exception: %s", describeError(env, err).c_str()); + abort(); +} + +} // namespace callstack::react_native_node_api diff --git a/packages/host/cpp/HermesNapiHost.hpp b/packages/host/cpp/HermesNapiHost.hpp new file mode 100644 index 00000000..1013d595 --- /dev/null +++ b/packages/host/cpp/HermesNapiHost.hpp @@ -0,0 +1,114 @@ +#pragma once + +#include + +#include +#include + +// Mirror of the host-integration interface declared by the vendored Hermes in +// API/napi/hermes_napi.h. We mirror it here (rather than including that +// header) to avoid pulling in Hermes' own node_api.h alongside the +// weak-node-api copy already included transitively. +// +// IMPORTANT: member order and types must match API/napi/hermes_napi.h at the +// commit pinned as HERMES_GIT_SHA in src/node/cli/hermes.ts — re-diff this +// struct against that header whenever the pin is bumped. +// +// The declarations must be `extern "C"`: since facebook/hermes#2106 (included +// in the pinned Hermes commit) the public hermes_napi.h wraps its entry points +// in `extern "C"`, so Hermes exports the unmangled C symbol. Without matching +// C linkage here the reference would be to the C++-mangled name and the app +// fails to link ("Undefined symbol: hermes_napi_create_env"). +extern "C" { +struct uv_loop_s; + +struct hermes_napi_host { + /// Schedule `execute` to run on a worker thread. When execute completes, + /// schedule `complete` to run on the main (JS) thread with napi_ok, or with + /// napi_cancelled if the work was cancelled before it started. + void (*post_work)(void *loop_data, void *work_data, + void (*execute)(void *work_data), + void (*complete)(void *work_data, napi_status status)); + + /// Attempt to cancel a previously posted work item. Returns true if the + /// work was still queued (its `complete` will run with napi_cancelled), + /// false if it already started or completed. + bool (*cancel_work)(void *loop_data, void *work_data); + + /// Schedule `callback` to run on the main (JS) thread. Used by thread-safe + /// functions to dispatch queued calls; may be invoked from any thread. + void (*post_task)(void *loop_data, void *task_data, + void (*callback)(void *task_data)); + + /// Opaque pointer passed as `loop_data` to the callbacks above. + void *data; + + /// If non-null, napi_get_uv_event_loop() returns this pointer. + struct uv_loop_s *uv_loop; + + /// If non-null, called by napi_fatal_exception() instead of aborting. + void (*fatal_exception)(void *data, napi_env env, napi_value err); + + /// Optional libuv-style loop refs used by thread-safe functions; may both + /// be null, in which case tsfn ref/unref are tracked but inert. + void (*ref_loop)(void *loop_data); + void (*unref_loop)(void *loop_data); +}; + +napi_env hermes_napi_create_env(void *hermes_runtime, hermes_napi_host *host); +} + +namespace callstack::react_native_node_api { + +/// Provides the `hermes_napi_host` integration for the Hermes Node-API +/// environments created by the host: a worker pool backing +/// napi_queue_async_work / napi_cancel_async_work and a JS-thread dispatcher +/// backing thread-safe functions. +/// +/// One instance exists per React Native runtime. The JS-thread hop is +/// type-erased as `JsDispatcher` (backed by CallInvoker::invokeAsync in the +/// app) so this class has no React Native dependencies and its threading +/// machinery can be exercised by plain C++ tests. +class HostContext : public std::enable_shared_from_this { +public: + /// Dispatches a function onto the JS thread. Implementations must be safe + /// to call from arbitrary threads, must never run the function inline and + /// must deliver functions one at a time, in order, on the single JS thread. + /// Dropping a function is only acceptable once the JS runtime is gone. + using JsDispatcher = std::function &&)>; + + static std::shared_ptr create(JsDispatcher dispatchToJs); + + /// Keep `context` alive for the remaining lifetime of the process. The + /// Hermes env reads the host struct during Runtime teardown *after* running + /// env cleanup hooks (napi_env__::shutdown() runs cleanup hooks first, then + /// hermes_napi_cleanup_tsfns, which reaches host_->unref_loop through + /// releaseTsfnLoopRef — verified at the pinned Hermes commit), so no + /// cleanup hook can tell us when the last env is truly done with the + /// struct. Retaining the context forever guarantees the documented + /// contract that the struct outlives every env it was passed to, at the + /// cost of a small allocation per React Native runtime (i.e. per reload). + static void retainForProcessLifetime(std::shared_ptr context); + + /// The struct to pass to hermes_napi_create_env. Owned by this context. + hermes_napi_host *host() { return &host_; } + + void dispatchToJs(std::function &&fn) { dispatchToJs_(std::move(fn)); } + +private: + explicit HostContext(JsDispatcher dispatchToJs); + + static void postWork(void *loop_data, void *work_data, + void (*execute)(void *work_data), + void (*complete)(void *work_data, + napi_status status)) noexcept; + static bool cancelWork(void *loop_data, void *work_data) noexcept; + static void postTask(void *loop_data, void *task_data, + void (*callback)(void *task_data)) noexcept; + static void fatalException(void *data, napi_env env, napi_value err) noexcept; + + JsDispatcher dispatchToJs_; + hermes_napi_host host_; +}; + +} // namespace callstack::react_native_node_api diff --git a/packages/host/cpp/RuntimeNodeApiAsync.cpp b/packages/host/cpp/RuntimeNodeApiAsync.cpp deleted file mode 100644 index 647aa71b..00000000 --- a/packages/host/cpp/RuntimeNodeApiAsync.cpp +++ /dev/null @@ -1,200 +0,0 @@ -#include "RuntimeNodeApiAsync.hpp" -#include "Logger.hpp" -#include - -struct AsyncJob { - using IdType = uint64_t; - enum State { Created, Queued, Completed, Cancelled, Deleted }; - - IdType id{}; - State state{}; - napi_env env; - napi_value async_resource; - napi_value async_resource_name; - napi_async_execute_callback execute; - napi_async_complete_callback complete; - void *data{nullptr}; - - static AsyncJob *fromWork(napi_async_work work) { - return reinterpret_cast(work); - } - static napi_async_work toWork(AsyncJob *job) { - return reinterpret_cast(job); - } -}; - -class AsyncWorkRegistry { -public: - using IdType = AsyncJob::IdType; - - std::shared_ptr create(napi_env env, napi_value async_resource, - napi_value async_resource_name, - napi_async_execute_callback execute, - napi_async_complete_callback complete, - void *data) { - const auto job = std::shared_ptr(new AsyncJob{ - .id = next_id(), - .state = AsyncJob::State::Created, - .env = env, - .async_resource = async_resource, - .async_resource_name = async_resource_name, - .execute = execute, - .complete = complete, - .data = data, - }); - - jobs_[job->id] = job; - return job; - } - - std::shared_ptr get(napi_async_work work) const { - const auto job = AsyncJob::fromWork(work); - if (!job) { - return {}; - } - if (const auto it = jobs_.find(job->id); it != jobs_.end()) { - return it->second; - } - return {}; - } - - bool release(IdType id) { - if (const auto it = jobs_.find(id); it != jobs_.end()) { - it->second->state = AsyncJob::State::Deleted; - jobs_.erase(it); - return true; - } - return false; - } - -private: - IdType next_id() { - if (current_id_ == std::numeric_limits::max()) [[unlikely]] { - current_id_ = 0; - } - return ++current_id_; - } - - IdType current_id_{0}; - std::unordered_map> jobs_; -}; - -static std::unordered_map> - callInvokers; -static AsyncWorkRegistry asyncWorkRegistry; - -namespace callstack::react_native_node_api { - -// Drop an env's entry when the env is torn down with its runtime (on a reload, -// for example). There is one env per addon, so without this the map keeps a -// stale entry per addon per runtime for the lifetime of the process. -static void NAPI_CDECL removeCallInvoker(void *env) { - callInvokers.erase(static_cast(env)); -} - -void setCallInvoker( - napi_env env, - const std::shared_ptr &invoker) { - const bool isFirstForEnv = !callInvokers.contains(env); - callInvokers[env] = invoker; - if (isFirstForEnv) { - ::napi_add_env_cleanup_hook(env, removeCallInvoker, env); - } -} - -std::weak_ptr getCallInvoker(napi_env env) { - return callInvokers.contains(env) - ? callInvokers[env] - : std::weak_ptr{}; -} - -napi_status napi_create_async_work(napi_env env, napi_value async_resource, - napi_value async_resource_name, - napi_async_execute_callback execute, - napi_async_complete_callback complete, - void *data, napi_async_work *result) { - const auto job = asyncWorkRegistry.create( - env, async_resource, async_resource_name, execute, complete, data); - if (!job) { - log_debug("Error: Failed to create async work job"); - return napi_generic_failure; - } - - *result = AsyncJob::toWork(job.get()); - return napi_ok; -} - -napi_status napi_queue_async_work(node_api_basic_env env, - napi_async_work work) { - const auto job = asyncWorkRegistry.get(work); - if (!job) { - log_debug("Error: Received null job in napi_queue_async_work"); - return napi_invalid_arg; - } - - const auto invoker = getCallInvoker(env).lock(); - if (!invoker) { - log_debug("Error: No CallInvoker available for async work"); - return napi_invalid_arg; - } - - invoker->invokeAsync([env, weakJob = std::weak_ptr{job}]() { - const auto job = weakJob.lock(); - if (!job) { - log_debug("Error: Async job has been deleted before execution"); - return; - } - if (job->state == AsyncJob::State::Queued) { - job->execute(job->env, job->data); - } - - job->complete(env, - job->state == AsyncJob::State::Cancelled ? napi_cancelled - : napi_ok, - job->data); - job->state = AsyncJob::State::Completed; - }); - - job->state = AsyncJob::State::Queued; - return napi_ok; -} - -napi_status napi_delete_async_work(node_api_basic_env env, - napi_async_work work) { - const auto job = asyncWorkRegistry.get(work); - if (!job) { - log_debug("Error: Received non-existent job in napi_delete_async_work"); - return napi_invalid_arg; - } - - if (!asyncWorkRegistry.release(job->id)) { - log_debug("Error: Failed to release async work job"); - return napi_generic_failure; - } - - return napi_ok; -} - -napi_status napi_cancel_async_work(node_api_basic_env env, - napi_async_work work) { - const auto job = asyncWorkRegistry.get(work); - if (!job) { - log_debug("Error: Received null job in napi_cancel_async_work"); - return napi_invalid_arg; - } - switch (job->state) { - case AsyncJob::State::Completed: - log_debug("Error: Cannot cancel async work that is already completed"); - return napi_generic_failure; - case AsyncJob::State::Deleted: - log_debug("Warning: Async work job is already deleted"); - return napi_generic_failure; - case AsyncJob::State::Cancelled: - log_debug("Warning: Async work job is already cancelled"); - return napi_ok; - } - - job->state = AsyncJob::State::Cancelled; - return napi_ok; -} -} // namespace callstack::react_native_node_api diff --git a/packages/host/cpp/RuntimeNodeApiAsync.hpp b/packages/host/cpp/RuntimeNodeApiAsync.hpp deleted file mode 100644 index be20128c..00000000 --- a/packages/host/cpp/RuntimeNodeApiAsync.hpp +++ /dev/null @@ -1,24 +0,0 @@ -#pragma once - -#include "node_api.h" -#include -#include - -namespace callstack::react_native_node_api { -void setCallInvoker( - napi_env env, const std::shared_ptr &invoker); - -napi_status napi_create_async_work(napi_env env, napi_value async_resource, - napi_value async_resource_name, - napi_async_execute_callback execute, - napi_async_complete_callback complete, - void *data, napi_async_work *result); - -napi_status napi_queue_async_work(node_api_basic_env env, napi_async_work work); - -napi_status napi_delete_async_work(node_api_basic_env env, - napi_async_work work); - -napi_status napi_cancel_async_work(node_api_basic_env env, - napi_async_work work); -} // namespace callstack::react_native_node_api diff --git a/packages/host/package.json b/packages/host/package.json index 6633b534..fb22e4f0 100644 --- a/packages/host/package.json +++ b/packages/host/package.json @@ -46,6 +46,9 @@ "injector:generate": "node scripts/generate-injector.mts", "test": "tsx --test --test-reporter=@reporters/github --test-reporter-destination=stdout --test-reporter=spec --test-reporter-destination=stdout src/node/**/*.test.ts src/node/*.test.ts", "test:gradle": "ENABLE_GRADLE_TESTS=true node --run test", + "test:configure": "cmake -S tests -B tests/build", + "test:build": "cmake --build tests/build", + "test:run": "ctest --test-dir tests/build --output-on-failure", "bootstrap": "node --run injector:generate", "prerelease": "node --run injector:generate" }, diff --git a/packages/host/scripts/generate-injector.mts b/packages/host/scripts/generate-injector.mts index d5c6cfd3..c58bb2f5 100644 --- a/packages/host/scripts/generate-injector.mts +++ b/packages/host/scripts/generate-injector.mts @@ -20,7 +20,6 @@ export function generateSource(functions: FunctionDecl[]) { #include #include - #include #if defined(__APPLE__) #define WEAK_NODE_API_LIBRARY_NAME "@rpath/weak-node-api.framework/weak-node-api" diff --git a/packages/host/src/node/cli/hermes.ts b/packages/host/src/node/cli/hermes.ts index e9893d5c..b1f74a3c 100644 --- a/packages/host/src/node/cli/hermes.ts +++ b/packages/host/src/node/cli/hermes.ts @@ -39,6 +39,11 @@ const HERMES_GIT_URL = "https://github.com/facebook/hermes.git"; // libraries, so `libhermesvm.so` ended up with undefined references to // `facebook::jsi::Serialized` that nothing in the APK defined, and the app died // on startup with "cannot locate symbol _ZTIN8facebook3jsi10SerializedE". +// +// When bumping this pin, re-diff the `hermes_napi_host` mirror in +// cpp/HermesNapiHost.hpp against `API/napi/hermes_napi.h` at the new commit: +// the struct is mirrored there (not included) and any change to its member +// order or signatures is an ABI break the compiler cannot catch. const HERMES_GIT_SHA = "5a795c9f880002c862c9254a26b57199819c97f7"; const platformOption = new Option( diff --git a/packages/host/tests/CMakeLists.txt b/packages/host/tests/CMakeLists.txt new file mode 100644 index 00000000..43723bbb --- /dev/null +++ b/packages/host/tests/CMakeLists.txt @@ -0,0 +1,42 @@ +cmake_minimum_required(VERSION 3.19) +project(react-native-node-api-host-tests) + +find_package(Threads REQUIRED) + +# Build weak-node-api from source for the host platform: it provides the +# node_api.h headers HermesNapiHost.hpp needs and the napi_* symbols the +# fatal-exception path references at link time. Requires the generated sources +# from `pnpm --filter weak-node-api run prebuild:prepare`. +add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/../../weak-node-api weak-node-api) + +Include(FetchContent) + +FetchContent_Declare( + Catch2 + GIT_REPOSITORY https://github.com/catchorg/Catch2.git + GIT_TAG v3.11.0 +) + +FetchContent_MakeAvailable(Catch2) + +add_executable(node-api-host-tests + test_hermes_napi_host.cpp + ../cpp/HermesNapiHost.cpp + ../cpp/Logger.cpp +) +target_include_directories(node-api-host-tests PRIVATE ../cpp) +target_link_libraries(node-api-host-tests + PRIVATE + weak-node-api + Catch2::Catch2WithMain + Threads::Threads +) + +target_compile_features(node-api-host-tests PRIVATE cxx_std_20) +target_compile_definitions(node-api-host-tests PRIVATE NAPI_VERSION=10) + +# As per https://github.com/catchorg/Catch2/blob/devel/docs/cmake-integration.md#catchcmake-and-catchaddtestscmake +list(APPEND CMAKE_MODULE_PATH ${catch2_SOURCE_DIR}/extras) +include(CTest) +include(Catch) +catch_discover_tests(node-api-host-tests) diff --git a/packages/host/tests/test_hermes_napi_host.cpp b/packages/host/tests/test_hermes_napi_host.cpp new file mode 100644 index 00000000..06a2bd87 --- /dev/null +++ b/packages/host/tests/test_hermes_napi_host.cpp @@ -0,0 +1,345 @@ +// Exercises the hermes_napi_host implementation (HermesNapiHost.cpp) from the +// Hermes side of the contract: the tests stand in for the calls Hermes' NAPI +// makes through the struct (napi_queue_async_work -> post_work, +// napi_cancel_async_work -> cancel_work, tsfn dispatch -> post_task), with a +// manually drained queue standing in for the CallInvoker-backed JS thread. +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace callstack::react_native_node_api; +using namespace std::chrono_literals; + +namespace { + +// Stands in for the JS thread: functions are queued by the dispatcher (from +// any thread) and only run when the test drains the queue. +struct FakeJsQueue { + HostContext::JsDispatcher dispatcher() { + return [this](std::function &&fn) { + { + std::lock_guard lock(mutex_); + queue_.push_back(std::move(fn)); + } + cv_.notify_all(); + }; + } + + // Runs queued functions one at a time until the queue is empty, including + // functions queued reentrantly while draining. Returns how many ran. + size_t drain() { + size_t count = 0; + for (;;) { + std::function fn; + { + std::lock_guard lock(mutex_); + if (queue_.empty()) { + return count; + } + fn = std::move(queue_.front()); + queue_.pop_front(); + } + fn(); + count++; + } + } + + bool waitForItems(size_t count, std::chrono::milliseconds timeout = 5s) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, + [&] { return queue_.size() >= count; }); + } + + size_t size() { + std::lock_guard lock(mutex_); + return queue_.size(); + } + +private: + std::mutex mutex_; + std::condition_variable cv_; + std::deque> queue_; +}; + +// A work payload whose execute blocks until the gate opens, for holding +// worker threads busy or keeping an item observably "running". +struct GatedWork { + std::mutex mutex; + std::condition_variable cv; + bool open = false; + std::atomic started{0}; + std::atomic completions{0}; + std::atomic executions{0}; + napi_status lastStatus = napi_ok; + + static void execute(void *data) { + auto *self = static_cast(data); + self->executions++; + self->started++; + std::unique_lock lock(self->mutex); + self->cv.wait(lock, [self] { return self->open; }); + } + + static void complete(void *data, napi_status status) { + auto *self = static_cast(data); + self->lastStatus = status; + self->completions++; + } + + void openGate() { + { + std::lock_guard lock(mutex); + open = true; + } + cv.notify_all(); + } + + void waitForStarted(int count) { + while (started.load() < count) { + std::this_thread::sleep_for(1ms); + } + } +}; + +// Matches WorkerPool::kThreadCount in HermesNapiHost.cpp; saturating all +// workers keeps a subsequently posted item deterministically queued. +constexpr int kWorkerCount = 4; + +} // namespace + +TEST_CASE("post_work runs execute off the posting thread and delivers " + "complete(napi_ok) through the dispatcher") { + FakeJsQueue js; + auto context = HostContext::create(js.dispatcher()); + hermes_napi_host *host = context->host(); + + struct Work { + std::thread::id executeThread{}; + std::atomic executed{false}; + std::atomic completions{0}; + napi_status status = napi_cancelled; + } work; + + host->post_work( + host->data, &work, + [](void *data) { + auto *w = static_cast(data); + w->executeThread = std::this_thread::get_id(); + w->executed = true; + }, + [](void *data, napi_status status) { + auto *w = static_cast(data); + w->status = status; + w->completions++; + }); + + // The completion is posted to the JS queue once execute finished on a + // worker thread — and must not have run inline. + REQUIRE(js.waitForItems(1)); + REQUIRE(work.executed.load()); + REQUIRE(work.executeThread != std::this_thread::get_id()); + REQUIRE(work.completions.load() == 0); + REQUIRE(js.drain() == 1); + REQUIRE(work.completions.load() == 1); + REQUIRE(work.status == napi_ok); +} + +TEST_CASE("cancel_work cancels queued items and rejects started items") { + FakeJsQueue js; + auto context = HostContext::create(js.dispatcher()); + hermes_napi_host *host = context->host(); + + SECTION("a queued item is cancelled: execute skipped, complete gets " + "napi_cancelled, a second cancel fails") { + auto *busy = new GatedWork(); + for (int i = 0; i < kWorkerCount; i++) { + host->post_work(host->data, busy, GatedWork::execute, + GatedWork::complete); + } + busy->waitForStarted(kWorkerCount); + + // Every worker is blocked on the gate, so this item stays queued. + auto *target = new GatedWork(); + host->post_work(host->data, target, GatedWork::execute, + GatedWork::complete); + REQUIRE(host->cancel_work(host->data, target)); + // Cancelling the same item again fails: it is no longer queued. + REQUIRE(!host->cancel_work(host->data, target)); + + REQUIRE(js.waitForItems(1)); + REQUIRE(js.drain() == 1); + REQUIRE(target->completions.load() == 1); + REQUIRE(target->lastStatus == napi_cancelled); + REQUIRE(target->executions.load() == 0); + + busy->openGate(); + REQUIRE(js.waitForItems(kWorkerCount)); + REQUIRE(js.drain() == kWorkerCount); + REQUIRE(busy->completions.load() == kWorkerCount); + delete busy; + delete target; + } + + SECTION("an item that started executing cannot be cancelled") { + auto *target = new GatedWork(); + host->post_work(host->data, target, GatedWork::execute, + GatedWork::complete); + target->waitForStarted(1); + REQUIRE(!host->cancel_work(host->data, target)); + target->openGate(); + REQUIRE(js.waitForItems(1)); + REQUIRE(js.drain() == 1); + REQUIRE(target->completions.load() == 1); + REQUIRE(target->lastStatus == napi_ok); + REQUIRE(target->executions.load() == 1); + delete target; + } +} + +TEST_CASE("cancel_work racing worker pickup yields exactly one outcome") { + FakeJsQueue js; + auto context = HostContext::create(js.dispatcher()); + hermes_napi_host *host = context->host(); + + struct Work { + std::atomic executions{0}; + std::atomic completions{0}; + std::atomic status{napi_generic_failure}; + }; + + for (int i = 0; i < 200; i++) { + Work work; + host->post_work( + host->data, &work, + [](void *data) { static_cast(data)->executions++; }, + [](void *data, napi_status status) { + auto *w = static_cast(data); + w->status = status; + w->completions++; + }); + bool cancelled = host->cancel_work(host->data, &work); + + // Exactly one completion arrives either way... + REQUIRE(js.waitForItems(1)); + REQUIRE(js.drain() == 1); + REQUIRE(work.completions.load() == 1); + // ...and it matches whether execute ran: cancelled XOR executed. + if (cancelled) { + REQUIRE(work.executions.load() == 0); + REQUIRE(work.status.load() == napi_cancelled); + } else { + REQUIRE(work.executions.load() == 1); + REQUIRE(work.status.load() == napi_ok); + } + } +} + +TEST_CASE("post_task delivers exactly once, in order and never inline") { + FakeJsQueue js; + auto context = HostContext::create(js.dispatcher()); + hermes_napi_host *host = context->host(); + + SECTION("a task posted from the current thread does not run inline") { + std::atomic runs{0}; + auto callback = [](void *data) { + static_cast *>(data)->fetch_add(1); + }; + host->post_task(host->data, &runs, callback); + REQUIRE(runs.load() == 0); + REQUIRE(js.drain() == 1); + REQUIRE(runs.load() == 1); + } + + SECTION("tasks are delivered in posting order") { + std::vector order; + struct Task { + std::vector *order; + int value; + }; + std::vector tasks; + for (int i = 0; i < 10; i++) { + tasks.push_back(Task{&order, i}); + } + for (auto &task : tasks) { + host->post_task(host->data, &task, [](void *data) { + auto *t = static_cast(data); + t->order->push_back(t->value); + }); + } + REQUIRE(js.drain() == 10); + REQUIRE(order == std::vector{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}); + } + + SECTION("tasks posted concurrently from many threads are all delivered") { + constexpr int kThreads = 8; + constexpr int kPostsPerThread = 100; + std::atomic runs{0}; + std::vector producers; + for (int i = 0; i < kThreads; i++) { + producers.emplace_back([&] { + for (int j = 0; j < kPostsPerThread; j++) { + host->post_task(host->data, &runs, [](void *data) { + static_cast *>(data)->fetch_add(1); + }); + } + }); + } + for (auto &producer : producers) { + producer.join(); + } + REQUIRE(js.drain() == kThreads * kPostsPerThread); + REQUIRE(runs.load() == kThreads * kPostsPerThread); + } + + SECTION("a task can repost itself from inside its own callback, as Hermes' " + "tsfn dispatch does") { + struct Repost { + hermes_napi_host *host; + std::atomic runs{0}; + + static void callback(void *data) { + auto *self = static_cast(data); + if (self->runs.fetch_add(1) + 1 < 5) { + self->host->post_task(self->host->data, self, &Repost::callback); + } + } + } repost{host, {}}; + host->post_task(host->data, &repost, &Repost::callback); + // The drain loop keeps going until reposted tasks stop arriving. + REQUIRE(js.drain() == 5); + REQUIRE(repost.runs.load() == 5); + } +} + +TEST_CASE("work completing after its context died is dropped, not crashed") { + FakeJsQueue js; + auto context = HostContext::create(js.dispatcher()); + hermes_napi_host *host = context->host(); + + // Leaked deliberately: complete never runs, so nothing would free it, and + // the worker may still be inside execute when the assertions run. + auto *work = new GatedWork(); + host->post_work(host->data, work, GatedWork::execute, GatedWork::complete); + work->waitForStarted(1); + + // Simulates a React Native runtime teardown: in production the context is + // retained for the process lifetime, but the dispatcher's CallInvoker — + // modelled here by the context itself — can die while work is in flight. + context.reset(); + work->openGate(); + + // The completion cannot be delivered anywhere; give the worker a moment to + // hit the drop path and assert nothing was queued and nothing crashed. + std::this_thread::sleep_for(100ms); + REQUIRE(js.size() == 0); + REQUIRE(work->completions.load() == 0); +} diff --git a/packages/node-addon-examples/src/index.ts b/packages/node-addon-examples/src/index.ts index 869d4825..68bac88a 100644 --- a/packages/node-addon-examples/src/index.ts +++ b/packages/node-addon-examples/src/index.ts @@ -77,13 +77,16 @@ export const suites: Record< }, ["hello world"]), }, "5-async-work": { - // TODO: This crashes (SIGABRT) - // "async_work_thread_safe_function": () => require("../examples/5-async-work/async_work_thread_safe_function/napi/index.js"), + async_work_thread_safe_function: () => { + require("../examples/5-async-work/async_work_thread_safe_function/napi/index.js"); + }, }, tests: { buffers: () => { require("../tests/buffers/addon.js"); }, async: () => require("../tests/async/addon.js") as () => Promise, + "threadsafe-function": () => + require("../tests/threadsafe-function/addon.js") as () => Promise, }, }; diff --git a/packages/node-addon-examples/tests/async/addon.c b/packages/node-addon-examples/tests/async/addon.c index 9444aacf..b385c50f 100644 --- a/packages/node-addon-examples/tests/async/addon.c +++ b/packages/node-addon-examples/tests/async/addon.c @@ -1,5 +1,7 @@ #include #include +#include +#include #include #include #include @@ -242,11 +244,152 @@ static napi_value DoRepeatedWork(napi_env env, napi_callback_info info) { return NULL; } +// The thread the addon was initialized on, i.e. the JS thread. +static pthread_t js_thread; + +typedef struct { + pthread_t execute_thread; + napi_ref callback; + napi_async_work work; +} thread_check_carrier; + +static thread_check_carrier thread_check; + +static void ThreadCheckExecute(napi_env env, void* data) { + thread_check_carrier* c = (thread_check_carrier*)data; + c->execute_thread = pthread_self(); +} + +static void ThreadCheckComplete(napi_env env, napi_status status, void* data) { + thread_check_carrier* c = (thread_check_carrier*)data; + napi_value argv[2]; + NODE_API_CALL_RETURN_VOID(env, + napi_get_boolean( + env, !pthread_equal(c->execute_thread, js_thread), &argv[0])); + NODE_API_CALL_RETURN_VOID(env, + napi_get_boolean(env, pthread_equal(pthread_self(), js_thread), &argv[1])); + napi_value callback; + NODE_API_CALL_RETURN_VOID( + env, napi_get_reference_value(env, c->callback, &callback)); + napi_value global; + NODE_API_CALL_RETURN_VOID(env, napi_get_global(env, &global)); + NODE_API_CALL_RETURN_VOID( + env, napi_call_function(env, global, callback, 2, argv, NULL)); + NODE_API_CALL_RETURN_VOID(env, napi_delete_reference(env, c->callback)); + NODE_API_CALL_RETURN_VOID(env, napi_delete_async_work(env, c->work)); +} + +// Queues work whose execute records its thread; the callback receives +// (executeRanOffJsThread, completeRanOnJsThread) booleans. +static napi_value TestExecuteThread(napi_env env, napi_callback_info info) { + size_t argc = 1; + napi_value cb, resource_name; + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, &cb, NULL, NULL)); + NODE_API_CALL(env, napi_create_reference(env, cb, 1, &thread_check.callback)); + NODE_API_CALL(env, + napi_create_string_utf8( + env, "TestExecuteThread", NAPI_AUTO_LENGTH, &resource_name)); + NODE_API_CALL(env, + napi_create_async_work(env, + NULL, + resource_name, + ThreadCheckExecute, + ThreadCheckComplete, + &thread_check, + &thread_check.work)); + NODE_API_CALL(env, napi_queue_async_work(env, thread_check.work)); + return NULL; +} + +static atomic_bool gate_open; +static atomic_bool gate_started; + +typedef struct { + napi_ref callback; + napi_async_work work; +} gated_carrier; + +static gated_carrier gated; + +static void GatedExecute(napi_env env, void* data) { + atomic_store(&gate_started, true); + while (!atomic_load(&gate_open)) { + sleep_ms(1); + } +} + +static void GatedComplete(napi_env env, napi_status status, void* data) { + gated_carrier* c = (gated_carrier*)data; + napi_value argv[1]; + NODE_API_CALL_RETURN_VOID( + env, napi_create_uint32(env, (uint32_t)status, &argv[0])); + napi_value callback; + NODE_API_CALL_RETURN_VOID( + env, napi_get_reference_value(env, c->callback, &callback)); + napi_value global; + NODE_API_CALL_RETURN_VOID(env, napi_get_global(env, &global)); + NODE_API_CALL_RETURN_VOID( + env, napi_call_function(env, global, callback, 1, argv, NULL)); + NODE_API_CALL_RETURN_VOID(env, napi_delete_reference(env, c->callback)); + NODE_API_CALL_RETURN_VOID(env, napi_delete_async_work(env, c->work)); +} + +// Queues work whose execute blocks until ReleaseGate() is called from JS. If +// execute ran on the JS thread (as the pre-hermes_napi_host implementation +// did), the JS thread could never call ReleaseGate and the test would hang. +static napi_value TestBlockingExecute(napi_env env, napi_callback_info info) { + size_t argc = 1; + napi_value cb, resource_name; + atomic_store(&gate_open, false); + atomic_store(&gate_started, false); + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, &cb, NULL, NULL)); + NODE_API_CALL(env, napi_create_reference(env, cb, 1, &gated.callback)); + NODE_API_CALL(env, + napi_create_string_utf8( + env, "TestBlockingExecute", NAPI_AUTO_LENGTH, &resource_name)); + NODE_API_CALL(env, + napi_create_async_work(env, + NULL, + resource_name, + GatedExecute, + GatedComplete, + &gated, + &gated.work)); + NODE_API_CALL(env, napi_queue_async_work(env, gated.work)); + return NULL; +} + +static napi_value HasStarted(napi_env env, napi_callback_info info) { + napi_value result; + NODE_API_CALL(env, napi_get_boolean(env, atomic_load(&gate_started), &result)); + return result; +} + +static napi_value ReleaseGate(napi_env env, napi_callback_info info) { + atomic_store(&gate_open, true); + return NULL; +} + +// Attempts to cancel the gated work and returns napi_cancel_async_work's +// status as a number, so JS can assert cancelling running work fails. +static napi_value CancelGated(napi_env env, napi_callback_info info) { + napi_status status = napi_cancel_async_work(env, gated.work); + napi_value result; + NODE_API_CALL(env, napi_create_uint32(env, (uint32_t)status, &result)); + return result; +} + static napi_value Init(napi_env env, napi_value exports) { + js_thread = pthread_self(); napi_property_descriptor properties[] = { DECLARE_NODE_API_PROPERTY("Test", Test), DECLARE_NODE_API_PROPERTY("TestCancel", TestCancel), DECLARE_NODE_API_PROPERTY("DoRepeatedWork", DoRepeatedWork), + DECLARE_NODE_API_PROPERTY("TestExecuteThread", TestExecuteThread), + DECLARE_NODE_API_PROPERTY("TestBlockingExecute", TestBlockingExecute), + DECLARE_NODE_API_PROPERTY("HasStarted", HasStarted), + DECLARE_NODE_API_PROPERTY("ReleaseGate", ReleaseGate), + DECLARE_NODE_API_PROPERTY("CancelGated", CancelGated), }; NODE_API_CALL(env, diff --git a/packages/node-addon-examples/tests/async/addon.js b/packages/node-addon-examples/tests/async/addon.js index a4a4bc6e..92821bfc 100644 --- a/packages/node-addon-examples/tests/async/addon.js +++ b/packages/node-addon-examples/tests/async/addon.js @@ -41,6 +41,78 @@ const doRepeatedWork = (count = 0) => test_async.DoRepeatedWork(workDone); }); -module.exports = () => { - return Promise.all([test(), testCancel(), doRepeatedWork()]); +const testExecuteThread = () => + new Promise((resolve, reject) => { + test_async.TestExecuteThread((executeOffJsThread, completeOnJsThread) => { + try { + assert.strictEqual( + executeOffJsThread, + true, + "expected execute to run off the JS thread", + ); + assert.strictEqual( + completeOnJsThread, + true, + "expected complete to run on the JS thread", + ); + resolve(); + } catch (e) { + reject(e); + } + }); + }); + +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +const waitForExecuteStart = async () => { + while (!test_async.HasStarted()) { + await delay(1); + } +}; + +const testBlockingExecute = async () => { + let completed = false; + const completion = new Promise((resolve, reject) => { + test_async.TestBlockingExecute((status) => { + completed = true; + try { + assert.strictEqual(status, 0 /* napi_ok */); + resolve(); + } catch (e) { + reject(e); + } + }); + }); + await waitForExecuteStart(); + assert.strictEqual(completed, false); + test_async.ReleaseGate(); + await completion; +}; + +const testCancelRunning = async () => { + const completion = new Promise((resolve, reject) => { + test_async.TestBlockingExecute((status) => { + try { + assert.strictEqual(status, 0 /* napi_ok */); + resolve(); + } catch (e) { + reject(e); + } + }); + }); + await waitForExecuteStart(); + // The work is executing, so cancellation must fail (unlike TestCancel, + // which cancels work that is still queued). + const status = test_async.CancelGated(); + assert.strictEqual(status, 9 /* napi_generic_failure */); + test_async.ReleaseGate(); + await completion; +}; + +module.exports = async () => { + await Promise.all([test(), testCancel(), doRepeatedWork()]); + // The gated tests share state in the addon, so they run sequentially. + await testExecuteThread(); + await testBlockingExecute(); + await testCancelRunning(); }; diff --git a/packages/node-addon-examples/tests/threadsafe-function/CMakeLists.txt b/packages/node-addon-examples/tests/threadsafe-function/CMakeLists.txt new file mode 100644 index 00000000..1be47aff --- /dev/null +++ b/packages/node-addon-examples/tests/threadsafe-function/CMakeLists.txt @@ -0,0 +1,26 @@ +cmake_minimum_required(VERSION 3.15...3.31) +project(threadsafe-function-test) + +find_package(weak-node-api REQUIRED CONFIG) + +add_library(addon SHARED addon.c) + +option(BUILD_APPLE_FRAMEWORK "Wrap addon in an Apple framework" ON) + +if(APPLE AND BUILD_APPLE_FRAMEWORK) + set_target_properties(addon PROPERTIES + FRAMEWORK TRUE + MACOSX_FRAMEWORK_IDENTIFIER threadsafe-function-test.addon + MACOSX_FRAMEWORK_SHORT_VERSION_STRING 1.0 + MACOSX_FRAMEWORK_BUNDLE_VERSION 1.0 + XCODE_ATTRIBUTE_SKIP_INSTALL NO + ) +else() + set_target_properties(addon PROPERTIES + PREFIX "" + SUFFIX .node + ) +endif() + +target_link_libraries(addon PRIVATE weak-node-api) +target_compile_features(addon PRIVATE cxx_std_17) diff --git a/packages/node-addon-examples/tests/threadsafe-function/addon.c b/packages/node-addon-examples/tests/threadsafe-function/addon.c new file mode 100644 index 00000000..ea694aa2 --- /dev/null +++ b/packages/node-addon-examples/tests/threadsafe-function/addon.c @@ -0,0 +1,407 @@ +// Ported from Node.js' test/node-api/test_threadsafe_function/binding.c. +// Upstream uses libuv's threading library; React Native has no libuv, so a +// small pthread-based shim stands in for the uv_* calls. The test logic and +// its assertions are kept as close to upstream as practical, with supplements +// marked as such: the call-into-JS callbacks assert they run on the JS thread, +// and Ref is exported alongside Unref (upstream exercises unref through a +// child-process teardown test, which does not port to React Native). +#include +#include +#include +#include +#include +#include +#include "../RuntimeNodeApiTestsCommon.h" + +// Upstream uses ARRAY_LENGTH 10000 and pauses every 1000 items; scaled down +// to keep the on-device runtime within the test timeout while preserving the +// ratios that matter: ARRAY_LENGTH / 2 must exceed Hermes' 1000-item tsfn +// dispatch budget (kMaxDispatchCount in API/napi/hermes_napi_tsfn.cpp) so the +// final run exercises the budget-exhausted re-post path, and the abort runs +// still get multiple pause windows. +#define ARRAY_LENGTH 2500 +#define MAX_QUEUE_SIZE 2 +#define PAUSE_EVERY 250 + +// pthread-based stand-ins for the libuv threading APIs used upstream. +typedef pthread_t uv_thread_t; +typedef void (*uv_thread_cb)(void* arg); + +typedef struct { + uv_thread_cb entry; + void* arg; +} uv_thread_shim; + +static void* uv_thread_shim_main(void* arg) { + uv_thread_shim shim = *(uv_thread_shim*)arg; + free(arg); + shim.entry(shim.arg); + return NULL; +} + +static int uv_thread_create(uv_thread_t* tid, uv_thread_cb entry, void* arg) { + uv_thread_shim* shim = malloc(sizeof(uv_thread_shim)); + if (shim == NULL) return -1; + shim->entry = entry; + shim->arg = arg; + int result = pthread_create(tid, NULL, uv_thread_shim_main, shim); + if (result != 0) free(shim); + return result; +} + +static int uv_thread_join(uv_thread_t* tid) { return pthread_join(*tid, NULL); } + +static uint64_t uv_hrtime(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000000000u + (uint64_t)ts.tv_nsec; +} + +// Supplement: the thread the addon was initialized on, i.e. the JS thread. +static pthread_t js_thread; + +static void assert_on_js_thread(const char* who) { + if (!pthread_equal(pthread_self(), js_thread)) { + napi_fatal_error(who, NAPI_AUTO_LENGTH, + "expected to be called on the JS thread", NAPI_AUTO_LENGTH); + } +} + +static uv_thread_t uv_threads[2]; +static napi_threadsafe_function ts_fn; + +typedef struct { + napi_threadsafe_function_call_mode block_on_full; + napi_threadsafe_function_release_mode abort; + bool start_secondary; + napi_ref js_finalize_cb; + uint32_t max_queue_size; +} ts_fn_hint; + +static ts_fn_hint ts_info; + +// Thread data to transmit to JS +static int ints[ARRAY_LENGTH]; + +static void secondary_thread(void* data) { + napi_threadsafe_function ts_fn = data; + + if (napi_release_threadsafe_function(ts_fn, napi_tsfn_release) != napi_ok) { + napi_fatal_error("secondary_thread", NAPI_AUTO_LENGTH, + "napi_release_threadsafe_function failed", NAPI_AUTO_LENGTH); + } +} + +// Source thread producing the data +static void data_source_thread(void* data) { + napi_threadsafe_function ts_fn = data; + int index; + void* hint; + ts_fn_hint* ts_fn_info; + napi_status status; + bool queue_was_full = false; + bool queue_was_closing = false; + + if (napi_get_threadsafe_function_context(ts_fn, &hint) != napi_ok) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "napi_get_threadsafe_function_context failed", NAPI_AUTO_LENGTH); + } + + ts_fn_info = (ts_fn_hint*)hint; + + if (ts_fn_info != &ts_info) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "thread-safe function hint is not as expected", NAPI_AUTO_LENGTH); + } + + if (ts_fn_info->start_secondary) { + if (napi_acquire_threadsafe_function(ts_fn) != napi_ok) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "napi_acquire_threadsafe_function failed", NAPI_AUTO_LENGTH); + } + + if (uv_thread_create(&uv_threads[1], secondary_thread, ts_fn) != 0) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "failed to start secondary thread", NAPI_AUTO_LENGTH); + } + } + + for (index = ARRAY_LENGTH - 1; index > -1 && !queue_was_closing; index--) { + status = napi_call_threadsafe_function(ts_fn, &ints[index], + ts_fn_info->block_on_full); + if (ts_fn_info->max_queue_size == 0 && (index % PAUSE_EVERY == 0)) { + // Let's make this thread really busy for 200 ms to give the main thread + // a chance to abort. + uint64_t start = uv_hrtime(); + for (; uv_hrtime() - start < 200000000;); + } + switch (status) { + case napi_queue_full: + queue_was_full = true; + index++; + // fall through + + case napi_ok: + continue; + + case napi_closing: + queue_was_closing = true; + break; + + default: + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "napi_call_threadsafe_function failed", NAPI_AUTO_LENGTH); + } + } + + // Assert that the enqueuing of a value was refused at least once, if this is + // a non-blocking test run. + if (!ts_fn_info->block_on_full && !queue_was_full) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "queue was never full", NAPI_AUTO_LENGTH); + } + + // Assert that the queue was marked as closing at least once, if this is an + // aborting test run. + if (ts_fn_info->abort == napi_tsfn_abort && !queue_was_closing) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "queue was never closing", NAPI_AUTO_LENGTH); + } + + if (!queue_was_closing && + napi_release_threadsafe_function(ts_fn, napi_tsfn_release) != napi_ok) { + napi_fatal_error("data_source_thread", NAPI_AUTO_LENGTH, + "napi_release_threadsafe_function failed", NAPI_AUTO_LENGTH); + } +} + +// Getting the data into JS +static void call_js(napi_env env, napi_value cb, void* hint, void* data) { + if (!(env == NULL || cb == NULL)) { + assert_on_js_thread("call_js"); + napi_value argv, undefined; + NODE_API_CALL_RETURN_VOID(env, napi_create_int32(env, *(int*)data, &argv)); + NODE_API_CALL_RETURN_VOID(env, napi_get_undefined(env, &undefined)); + NODE_API_CALL_RETURN_VOID( + env, napi_call_function(env, undefined, cb, 1, &argv, NULL)); + } +} + +static napi_ref alt_ref; +// Getting the data into JS with the alternative reference +static void call_ref(napi_env env, napi_value _, void* hint, void* data) { + if (!(env == NULL || alt_ref == NULL)) { + assert_on_js_thread("call_ref"); + napi_value fn, argv, undefined; + NODE_API_CALL_RETURN_VOID(env, napi_get_reference_value(env, alt_ref, &fn)); + NODE_API_CALL_RETURN_VOID(env, napi_create_int32(env, *(int*)data, &argv)); + NODE_API_CALL_RETURN_VOID(env, napi_get_undefined(env, &undefined)); + NODE_API_CALL_RETURN_VOID( + env, napi_call_function(env, undefined, fn, 1, &argv, NULL)); + } +} + +// Cleanup +static napi_value StopThread(napi_env env, napi_callback_info info) { + size_t argc = 2; + napi_value argv[2]; + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, argv, NULL, NULL)); + napi_valuetype value_type; + NODE_API_CALL(env, napi_typeof(env, argv[0], &value_type)); + NODE_API_ASSERT(env, value_type == napi_function, + "StopThread argument is a function"); + NODE_API_ASSERT(env, (ts_fn != NULL), "Existing threadsafe function"); + NODE_API_CALL(env, + napi_create_reference(env, argv[0], 1, &(ts_info.js_finalize_cb))); + bool abort; + NODE_API_CALL(env, napi_get_value_bool(env, argv[1], &abort)); + NODE_API_CALL(env, + napi_release_threadsafe_function( + ts_fn, abort ? napi_tsfn_abort : napi_tsfn_release)); + ts_fn = NULL; + return NULL; +} + +// Join the thread and inform JS that we're done. +static void join_the_threads(napi_env env, void* data, void* hint) { + assert_on_js_thread("join_the_threads"); + uv_thread_t* the_threads = data; + ts_fn_hint* the_hint = hint; + napi_value js_cb, undefined; + + uv_thread_join(&the_threads[0]); + if (the_hint->start_secondary) { + uv_thread_join(&the_threads[1]); + } + + NODE_API_CALL_RETURN_VOID( + env, napi_get_reference_value(env, the_hint->js_finalize_cb, &js_cb)); + NODE_API_CALL_RETURN_VOID(env, napi_get_undefined(env, &undefined)); + NODE_API_CALL_RETURN_VOID( + env, napi_call_function(env, undefined, js_cb, 0, NULL, NULL)); + NODE_API_CALL_RETURN_VOID( + env, napi_delete_reference(env, the_hint->js_finalize_cb)); + if (alt_ref != NULL) { + NODE_API_CALL_RETURN_VOID(env, napi_delete_reference(env, alt_ref)); + alt_ref = NULL; + } +} + +static napi_value StartThreadInternal(napi_env env, napi_callback_info info, + napi_threadsafe_function_call_js cb, bool block_on_full, + bool alt_ref_js_cb) { + size_t argc = 4; + napi_value argv[4]; + + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, argv, NULL, NULL)); + if (alt_ref_js_cb) { + NODE_API_CALL(env, napi_create_reference(env, argv[0], 1, &alt_ref)); + argv[0] = NULL; + } + + ts_info.block_on_full = + (block_on_full ? napi_tsfn_blocking : napi_tsfn_nonblocking); + + NODE_API_ASSERT(env, (ts_fn == NULL), "Existing thread-safe function"); + napi_value async_name; + NODE_API_CALL(env, + napi_create_string_utf8(env, "Node-API Thread-safe Function Test", + NAPI_AUTO_LENGTH, &async_name)); + NODE_API_CALL(env, + napi_get_value_uint32(env, argv[3], &ts_info.max_queue_size)); + NODE_API_CALL(env, + napi_create_threadsafe_function(env, + argv[0], + NULL, + async_name, + ts_info.max_queue_size, + 2, + uv_threads, + join_the_threads, + &ts_info, + cb, + &ts_fn)); + bool abort; + NODE_API_CALL(env, napi_get_value_bool(env, argv[1], &abort)); + ts_info.abort = abort ? napi_tsfn_abort : napi_tsfn_release; + NODE_API_CALL(env, + napi_get_value_bool(env, argv[2], &(ts_info.start_secondary))); + + NODE_API_ASSERT(env, + (uv_thread_create(&uv_threads[0], data_source_thread, ts_fn) == 0), + "Thread creation"); + + return NULL; +} + +static napi_value Ref(napi_env env, napi_callback_info info) { + NODE_API_ASSERT(env, ts_fn != NULL, "No existing thread-safe function"); + NODE_API_CALL(env, napi_ref_threadsafe_function(env, ts_fn)); + return NULL; +} + +static napi_value Unref(napi_env env, napi_callback_info info) { + NODE_API_ASSERT(env, ts_fn != NULL, "No existing thread-safe function"); + NODE_API_CALL(env, napi_unref_threadsafe_function(env, ts_fn)); + return NULL; +} + +static napi_value Release(napi_env env, napi_callback_info info) { + NODE_API_ASSERT(env, ts_fn != NULL, "No existing thread-safe function"); + NODE_API_CALL( + env, napi_release_threadsafe_function(ts_fn, napi_tsfn_release)); + return NULL; +} + +// Startup +static napi_value StartThread(napi_env env, napi_callback_info info) { + return StartThreadInternal(env, info, call_js, + /** block_on_full */ true, /** alt_ref_js_cb */ false); +} + +static napi_value StartThreadNonblocking(napi_env env, + napi_callback_info info) { + return StartThreadInternal(env, info, call_js, + /** block_on_full */ false, /** alt_ref_js_cb */ false); +} + +static napi_value StartThreadNoNative(napi_env env, napi_callback_info info) { + return StartThreadInternal(env, info, NULL, + /** block_on_full */ true, /** alt_ref_js_cb */ false); +} + +static napi_value StartThreadNoJsFunc(napi_env env, napi_callback_info info) { + return StartThreadInternal(env, info, call_ref, + /** block_on_full */ true, /** alt_ref_js_cb */ true); +} + +// Testing calling into JavaScript +static void ThreadSafeFunctionFinalize(napi_env env, void* finalize_data, + void* finalize_hint) { + assert_on_js_thread("ThreadSafeFunctionFinalize"); + napi_ref js_func_ref = (napi_ref)finalize_data; + napi_value js_func; + napi_value recv; + NODE_API_CALL_RETURN_VOID( + env, napi_get_reference_value(env, js_func_ref, &js_func)); + NODE_API_CALL_RETURN_VOID(env, napi_get_global(env, &recv)); + NODE_API_CALL_RETURN_VOID( + env, napi_call_function(env, recv, js_func, 0, NULL, NULL)); + NODE_API_CALL_RETURN_VOID(env, napi_delete_reference(env, js_func_ref)); +} + +// Testing calling into JavaScript +static napi_value CallIntoModule(napi_env env, napi_callback_info info) { + size_t argc = 4; + napi_value argv[4]; + NODE_API_CALL(env, napi_get_cb_info(env, info, &argc, argv, NULL, NULL)); + + napi_ref finalize_func; + NODE_API_CALL(env, napi_create_reference(env, argv[3], 1, &finalize_func)); + + napi_threadsafe_function tsfn; + NODE_API_CALL(env, + napi_create_threadsafe_function(env, argv[0], argv[1], argv[2], 0, 1, + finalize_func, ThreadSafeFunctionFinalize, NULL, NULL, &tsfn)); + NODE_API_CALL( + env, napi_call_threadsafe_function(tsfn, NULL, napi_tsfn_blocking)); + NODE_API_CALL(env, napi_release_threadsafe_function(tsfn, napi_tsfn_release)); + return NULL; +} + +// Module init +static napi_value Init(napi_env env, napi_value exports) { + js_thread = pthread_self(); + size_t index; + for (index = 0; index < ARRAY_LENGTH; index++) { + ints[index] = index; + } + napi_value js_array_length, js_max_queue_size; + napi_create_uint32(env, ARRAY_LENGTH, &js_array_length); + napi_create_uint32(env, MAX_QUEUE_SIZE, &js_max_queue_size); + + napi_property_descriptor properties[] = { + {"ARRAY_LENGTH", NULL, NULL, NULL, NULL, js_array_length, napi_enumerable, + NULL}, + {"MAX_QUEUE_SIZE", NULL, NULL, NULL, NULL, js_max_queue_size, + napi_enumerable, NULL}, + DECLARE_NODE_API_PROPERTY("StartThread", StartThread), + DECLARE_NODE_API_PROPERTY("StartThreadNoNative", StartThreadNoNative), + DECLARE_NODE_API_PROPERTY("StartThreadNonblocking", + StartThreadNonblocking), + DECLARE_NODE_API_PROPERTY("StartThreadNoJsFunc", StartThreadNoJsFunc), + DECLARE_NODE_API_PROPERTY("StopThread", StopThread), + DECLARE_NODE_API_PROPERTY("Ref", Ref), + DECLARE_NODE_API_PROPERTY("Unref", Unref), + DECLARE_NODE_API_PROPERTY("Release", Release), + DECLARE_NODE_API_PROPERTY("CallIntoModule", CallIntoModule), + }; + + NODE_API_CALL(env, + napi_define_properties( + env, exports, sizeof(properties) / sizeof(properties[0]), + properties)); + + return exports; +} +NAPI_MODULE(NODE_GYP_MODULE_NAME, Init) diff --git a/packages/node-addon-examples/tests/threadsafe-function/addon.js b/packages/node-addon-examples/tests/threadsafe-function/addon.js new file mode 100644 index 00000000..a1b8e385 --- /dev/null +++ b/packages/node-addon-examples/tests/threadsafe-function/addon.js @@ -0,0 +1,287 @@ +// Ported from Node.js' test/node-api/test_threadsafe_function/test.js. The +// upstream child-process teardown tests (testUnref) do not port to React +// Native; ref/unref are instead exercised in-process by testRefUnref, and +// testCallIntoModule supplements the suite by asserting that delivery is +// never synchronous, even when calling from the JS thread itself. +const assert = require("assert"); +const binding = require("bindings")("addon.node"); +const expectedArray = (function (arrayLength) { + const result = []; + for (let index = 0; index < arrayLength; index++) { + result.push(arrayLength - 1 - index); + } + return result; +})(binding.ARRAY_LENGTH); + +function testWithJSMarshaller({ + threadStarter, + quitAfter, + abort, + maxQueueSize, + launchSecondary, +}) { + return new Promise((resolve) => { + const array = []; + binding[threadStarter]( + function testCallback(value) { + array.push(value); + if (array.length === quitAfter) { + setImmediate(() => { + binding.StopThread(() => { + resolve(array); + }, !!abort); + }); + } + }, + !!abort, + !!launchSecondary, + maxQueueSize, + ); + if (threadStarter === "StartThreadNonblocking") { + // Let's make this thread really busy for a short while to ensure that + // the queue fills and the thread receives a napi_queue_full. + const start = Date.now(); + while (Date.now() - start < 200); + } + }); +} + +function testWithoutJSMarshaller() { + return new Promise((resolve) => { + let callCount = 0; + binding.StartThreadNoNative( + function testCallback() { + callCount++; + + // The default call-into-JS implementation passes no arguments. + assert.strictEqual(arguments.length, 0); + if (callCount === binding.ARRAY_LENGTH) { + setImmediate(() => { + binding.StopThread(() => { + resolve(); + }, false); + }); + } + }, + false /* abort */, + false /* launchSecondary */, + binding.MAX_QUEUE_SIZE, + ); + }); +} + +// With no libuv loop to act on in React Native (ref_loop/unref_loop are left +// null in the hermes_napi_host), napi_ref/unref_threadsafe_function must +// still succeed and leave delivery unaffected. +function testRefUnref() { + return new Promise((resolve) => { + const array = []; + let refCycled = false; + binding.StartThread( + function testCallback(value) { + array.push(value); + if (!refCycled) { + refCycled = true; + binding.Unref(); + binding.Ref(); + binding.Unref(); + } + if (array.length === binding.ARRAY_LENGTH) { + setImmediate(() => { + binding.StopThread(() => { + resolve(array); + }, false); + }); + } + }, + false /* abort */, + false /* launchSecondary */, + binding.MAX_QUEUE_SIZE, + ); + }).then((result) => assert.deepStrictEqual(result, expectedArray)); +} + +// Create a threadsafe function and call it from the JS thread itself: the +// delivery and the finalize callback must both still happen asynchronously. +function testCallIntoModule() { + return new Promise((resolve, reject) => { + let delivered = false; + let finalized = false; + binding.CallIntoModule( + () => { + delivered = true; + }, + {}, + "test_tsfn_resource", + () => { + finalized = true; + try { + assert.strictEqual(delivered, true); + resolve(); + } catch (e) { + reject(e); + } + }, + ); + assert.strictEqual(delivered, false); + assert.strictEqual(finalized, false); + }); +} + +module.exports = () => + testWithoutJSMarshaller() + // Start the thread in blocking mode, and assert that all values are + // passed. Quit after it's done. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + maxQueueSize: binding.MAX_QUEUE_SIZE, + quitAfter: binding.ARRAY_LENGTH, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in blocking mode, and assert that all values are + // passed. Quit after it's done. + // Doesn't pass the callback js function to napi_create_threadsafe_function. + // Instead, use an alternative reference to get js function called. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThreadNoJsFunc", + maxQueueSize: binding.MAX_QUEUE_SIZE, + quitAfter: binding.ARRAY_LENGTH, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in blocking mode with an infinite queue, and assert + // that all values are passed. Quit after it's done. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + maxQueueSize: 0, + quitAfter: binding.ARRAY_LENGTH, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in non-blocking mode, and assert that all values are + // passed. Quit after it's done. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThreadNonblocking", + maxQueueSize: binding.MAX_QUEUE_SIZE, + quitAfter: binding.ARRAY_LENGTH, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in blocking mode, and assert that all values are + // passed. Quit early, but let the thread finish. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + maxQueueSize: binding.MAX_QUEUE_SIZE, + quitAfter: 1, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in blocking mode with an infinite queue, and assert + // that all values are passed. Quit early, but let the thread finish. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + maxQueueSize: 0, + quitAfter: 1, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in non-blocking mode, and assert that all values are + // passed. Quit early, but let the thread finish. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThreadNonblocking", + maxQueueSize: binding.MAX_QUEUE_SIZE, + quitAfter: 1, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in blocking mode, and assert that all values are + // passed. Quit early, but let the thread finish. Launch a secondary thread + // to test the reference counter incrementing functionality. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + quitAfter: 1, + maxQueueSize: binding.MAX_QUEUE_SIZE, + launchSecondary: true, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in non-blocking mode, and assert that all values are + // passed. Quit early, but let the thread finish. Launch a secondary thread + // to test the reference counter incrementing functionality. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThreadNonblocking", + quitAfter: 1, + maxQueueSize: binding.MAX_QUEUE_SIZE, + launchSecondary: true, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + // Start the thread in blocking mode, and assert that it could not finish. + // Quit early by aborting. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + quitAfter: 1, + maxQueueSize: binding.MAX_QUEUE_SIZE, + abort: true, + }), + ) + .then((result) => assert.strictEqual(result.indexOf(0), -1)) + + // Start the thread in blocking mode with an infinite queue, and assert + // that it could not finish. Quit early by aborting. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThread", + quitAfter: 1, + maxQueueSize: 0, + abort: true, + }), + ) + .then((result) => assert.strictEqual(result.indexOf(0), -1)) + + // Start the thread in non-blocking mode, and assert that it could not + // finish. Quit early and aborting. + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThreadNonblocking", + quitAfter: 1, + maxQueueSize: binding.MAX_QUEUE_SIZE, + abort: true, + }), + ) + .then((result) => assert.strictEqual(result.indexOf(0), -1)) + + // Make sure that the threadsafe function isn't stalled when the queue + // outgrows what a single dispatch may drain (kMaxDispatchCount in + // Hermes' API/napi/hermes_napi_tsfn.cpp). + .then(() => + testWithJSMarshaller({ + threadStarter: "StartThreadNonblocking", + maxQueueSize: binding.ARRAY_LENGTH >>> 1, + quitAfter: binding.ARRAY_LENGTH, + }), + ) + .then((result) => assert.deepStrictEqual(result, expectedArray)) + + .then(() => testRefUnref()) + .then(() => testCallIntoModule()); diff --git a/packages/node-addon-examples/tests/threadsafe-function/binding.gyp b/packages/node-addon-examples/tests/threadsafe-function/binding.gyp new file mode 100644 index 00000000..80f9fa87 --- /dev/null +++ b/packages/node-addon-examples/tests/threadsafe-function/binding.gyp @@ -0,0 +1,8 @@ +{ + "targets": [ + { + "target_name": "addon", + "sources": [ "addon.c" ] + } + ] +} diff --git a/packages/node-addon-examples/tests/threadsafe-function/package.json b/packages/node-addon-examples/tests/threadsafe-function/package.json new file mode 100644 index 00000000..c2fdf057 --- /dev/null +++ b/packages/node-addon-examples/tests/threadsafe-function/package.json @@ -0,0 +1,14 @@ +{ + "name": "threadsafe-function-test", + "version": "0.0.0", + "description": "Tests of runtime threadsafe functions", + "main": "addon.js", + "private": true, + "dependencies": { + "bindings": "~1.5.0" + }, + "scripts": { + "test": "node addon.js" + }, + "gypfile": true +} From 1b8a9d141cee7cf31468ae9682b2fae619a75d1c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 08:32:33 +0000 Subject: [PATCH 2/4] Trigger CI for the label-gated device lanes The Check workflow only reacts to opened/synchronize/reopened, so the Apple and Android labels added to the PR need a synchronize event to be seen by the job conditions. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6xHHeMFF853z5R6th5CkF From f59db0254a02dd06e49024ec18588b89965999e3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 08:35:29 +0000 Subject: [PATCH 3/4] Trigger CI with the weak-node-api and host labels applied Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6xHHeMFF853z5R6th5CkF From 1b432c83badffb9479770aac5c2a57cbaa0f6e51 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 18:15:04 +0000 Subject: [PATCH 4/4] Address review: truthful teardown outcomes, duplicate-queue drop - JsDispatcher now reports acceptance and WorkItem holds its HostContext strongly: the weak_ptr could never expire (contexts are retained for the process lifetime), so the pool's drop branches were dead code and napi_cancel_async_work could claim success for a completion the dispatcher was about to drop. cancel_work now returns the dispatcher's verdict, and workerMain/postTask log drops where they actually happen. - WorkerPool::enqueue drops a double-queued (loopData, workData) instead of enqueueing it: a second entry meant two completions for one napi_async_work and a use-after-free once the addon deletes the work inside the first. Covered by a new Catch2 test; the saturation helper now uses distinct jobs per worker so it does not trip the detection. - Delete HostContext copy/move: host_.data points at this. - Justify the CallInvoker-liveness assumption at the dispatcher site (RuntimeSchedulerCallInvoker holds a weak RuntimeScheduler owned together with the runtime, so accepted work cannot outlive it) and correct the WorkItem comment: the (loopData, workData) pair separates runtimes/reloads, not envs. - Rework the Catch2 teardown test to model an expired CallInvoker (the state production reaches) instead of dropping the last context ref (which it never does), and cover the rejected post_task path. - Scope the 30s mocha timeout to the threadsafe-function suite so a genuine deadlock elsewhere still fails fast; add a TODO on fatal_exception about routing through RN error handling. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6xHHeMFF853z5R6th5CkF --- apps/test-app/App.tsx | 9 +- packages/host/cpp/CxxNodeApiHostModule.cpp | 30 +++- packages/host/cpp/HermesNapiHost.cpp | 78 +++++----- packages/host/cpp/HermesNapiHost.hpp | 23 ++- packages/host/tests/test_hermes_napi_host.cpp | 143 ++++++++++++++---- 5 files changed, 203 insertions(+), 80 deletions(-) diff --git a/apps/test-app/App.tsx b/apps/test-app/App.tsx index 3006b1fa..3322b136 100644 --- a/apps/test-app/App.tsx +++ b/apps/test-app/App.tsx @@ -39,9 +39,12 @@ function loadTests({ describe(suiteName, () => { for (const [exampleName, requireExample] of Object.entries(examples)) { it(exampleName, async function () { - // Some examples (the threadsafe-function suite in particular) - // marshal thousands of values across threads. - this.timeout(30_000); + if (exampleName === "threadsafe-function") { + // The ported Node.js suite marshals thousands of values across + // threads; every other example keeps the default timeout so a + // genuine deadlock still fails fast. + this.timeout(30_000); + } const test = requireExample(); if (test instanceof Function) { const result = test(); diff --git a/packages/host/cpp/CxxNodeApiHostModule.cpp b/packages/host/cpp/CxxNodeApiHostModule.cpp index 9a3ef8c7..6720d272 100644 --- a/packages/host/cpp/CxxNodeApiHostModule.cpp +++ b/packages/host/cpp/CxxNodeApiHostModule.cpp @@ -17,17 +17,35 @@ CxxNodeApiHostModule::CxxNodeApiHostModule( // The JS-thread dispatcher behind the hermes_napi_host integration: // CallInvoker::invokeAsync is callable from any thread, never runs the - // function inline and delivers in order on the JS thread. The CallInvoker - // is captured weakly so tasks in flight during a runtime teardown are - // dropped instead of dispatched into a dead runtime. + // function inline and delivers in order on the JS thread. + // + // Teardown is the load-bearing case. What the host integration needs is + // that a function handed to this dispatcher either runs on the JS thread + // while the runtime is alive, or is dropped — never invoked against a + // destroyed runtime. In bridgeless React Native the CallInvoker received + // here is a RuntimeSchedulerCallInvoker holding a std::weak_ptr to the + // RuntimeScheduler; the ReactInstance owns scheduler and runtime together + // and invokeAsync no-ops once the scheduler is gone, so work cannot outlive + // the runtime it targets. The weak capture below covers the remaining + // window where this module (and its CallInvoker reference) is released + // during instance teardown. + // + // Dropping is safe precisely because a drop implies that teardown: every + // env this host serves is owned by that same runtime and destroyed with it, + // so the completion or tsfn dispatch being dropped has no live observer. + // The one caller that could still see the difference — + // napi_cancel_async_work — receives the verdict through this dispatcher's + // return value (see HostContext::cancelWork). hostContext_ = HostContext::create( [weakInvoker = std::weak_ptr(callInvoker_)](std::function &&fn) { - if (auto invoker = weakInvoker.lock()) { - invoker->invokeAsync(std::move(fn)); - } else { + auto invoker = weakInvoker.lock(); + if (!invoker) { log_warning( "NapiHost: dropping a task posted after runtime teardown"); + return false; } + invoker->invokeAsync(std::move(fn)); + return true; }); HostContext::retainForProcessLifetime(hostContext_); } diff --git a/packages/host/cpp/HermesNapiHost.cpp b/packages/host/cpp/HermesNapiHost.cpp index 8fb17250..681e16d2 100644 --- a/packages/host/cpp/HermesNapiHost.cpp +++ b/packages/host/cpp/HermesNapiHost.cpp @@ -15,13 +15,14 @@ namespace { struct WorkItem { // Identifies the HostContext that posted the item; matched together with - // workData on cancellation, since the pool is shared by all runtimes and a - // freed napi_async_work address could be reused by another env. + // workData on cancellation. All envs of one runtime share one context, so + // the pair only disambiguates across runtimes (i.e. reloads), where a freed + // napi_async_work address could be reused by a new runtime's env. void *loopData = nullptr; - // The dispatcher may expire while an item is in flight (React Native - // reload); the completion is then dropped, which is safe because the env it - // targets is torn down with its runtime. - std::weak_ptr context; + // Held strongly: contexts are retained for the process lifetime anyway, and + // whether the item's runtime can still receive its completion is reported + // by the context's dispatcher, not by this pointer's liveness. + std::shared_ptr context; void *workData = nullptr; void (*execute)(void *work_data) = nullptr; void (*complete)(void *work_data, napi_status status) = nullptr; @@ -44,11 +45,14 @@ class WorkerPool { if (queued.loopData == item.loopData && queued.workData == item.workData) { // Queueing the same napi_async_work twice is undefined behavior in - // Node (libuv asserts); warn instead of crashing. - log_warning( - "NapiHost: napi_async_work %p was queued while already queued", + // Node (libuv asserts). Drop the duplicate instead of crashing: + // enqueueing it would produce two completions for one work item, + // and the second is a use-after-free once the addon has called + // napi_delete_async_work from inside the first. + log_warning("NapiHost: dropping napi_async_work %p, queued while " + "already queued", item.workData); - break; + return; } } queue_.push_back(std::move(item)); @@ -93,14 +97,13 @@ class WorkerPool { // or removed by tryRemove (complete gets napi_cancelled) — never both, // as both happen under the queue mutex. item.execute(item.workData); - if (auto context = item.context.lock()) { - context->dispatchToJs( - [workData = item.workData, complete = item.complete] { - // No pool state refers to workData at this point, so the - // complete callback is free to napi_delete_async_work it. - complete(workData, napi_ok); - }); - } else { + bool accepted = item.context->dispatchToJs( + [workData = item.workData, complete = item.complete] { + // No pool state refers to workData at this point, so the + // complete callback is free to napi_delete_async_work it. + complete(workData, napi_ok); + }); + if (!accepted) { log_warning("NapiHost: dropping an async work completion posted after " "runtime teardown"); } @@ -185,11 +188,9 @@ void HostContext::postWork(void *loop_data, void *work_data, void (*complete)(void *work_data, napi_status status)) noexcept { auto *self = static_cast(loop_data); - // Called on the JS thread (napi_queue_async_work) while the env — and - // therefore this context — is alive, so weak_from_this() is populated. WorkerPool::instance().enqueue(WorkItem{ .loopData = loop_data, - .context = self->weak_from_this(), + .context = self->shared_from_this(), .workData = work_data, .execute = execute, .complete = complete, @@ -203,17 +204,16 @@ bool HostContext::cancelWork(void *loop_data, void *work_data) noexcept { // and Hermes surfaces napi_generic_failure, like Node. return false; } - if (auto context = item.context.lock()) { - // Deliver the cancelled completion asynchronously, matching Node, where a - // cancelled complete callback still runs on a later loop tick. - context->dispatchToJs([workData = item.workData, complete = item.complete] { - complete(workData, napi_cancelled); - }); - return true; - } - // The runtime is being torn down; the complete callback can never run, so - // report the cancellation as failed. - return false; + // Deliver the cancelled completion asynchronously, matching Node, where a + // cancelled complete callback still runs on a later loop tick. Success is + // only reported while the dispatcher accepts the delivery: once the runtime + // is torn down the complete callback can never run, and claiming success + // would leave the addon waiting for a complete(napi_cancelled) that never + // arrives. + return item.context->dispatchToJs( + [workData = item.workData, complete = item.complete] { + complete(workData, napi_cancelled); + }); } void HostContext::postTask(void *loop_data, void *task_data, @@ -223,8 +223,13 @@ void HostContext::postTask(void *loop_data, void *task_data, // Hermes' tsfnDispatch re-posts itself from inside the callback. The // dispatcher never runs the callback inline (JS would run off-thread) and // never drops it while the runtime is alive — a dropped dispatch would - // permanently wedge the tsfn, as its dispatch_pending flag stays set. - self->dispatchToJs_([task_data, callback] { callback(task_data); }); + // permanently wedge the tsfn, as its dispatch_pending flag stays set. A + // rejected dispatch therefore implies the runtime (and with it the tsfn's + // env) is gone, making the wedged flag unobservable. + if (!self->dispatchToJs_([task_data, callback] { callback(task_data); })) { + log_warning("NapiHost: dropping a thread-safe function dispatch posted " + "after runtime teardown"); + } } void HostContext::fatalException(void *, napi_env env, @@ -234,6 +239,11 @@ void HostContext::fatalException(void *, napi_env env, // error and abort — the same observable outcome as Hermes' null-host // default, but surfaced through the host logger. `err` is only valid for // the duration of this call, so it is stringified before returning. + // TODO: Route through React Native's error handling (ErrorUtils / LogBox), + // with abort() as the fallback, to get closer to Node's observable and + // handleable 'uncaughtException' — note node-addon-api calls + // napi_fatal_exception whenever an exception escapes a thread-safe + // function callback, so today a single throwing tsfn callback is fatal. log_error("napi_fatal_exception: %s", describeError(env, err).c_str()); abort(); } diff --git a/packages/host/cpp/HermesNapiHost.hpp b/packages/host/cpp/HermesNapiHost.hpp index 1013d595..3d8f314a 100644 --- a/packages/host/cpp/HermesNapiHost.hpp +++ b/packages/host/cpp/HermesNapiHost.hpp @@ -71,11 +71,15 @@ namespace callstack::react_native_node_api { /// machinery can be exercised by plain C++ tests. class HostContext : public std::enable_shared_from_this { public: - /// Dispatches a function onto the JS thread. Implementations must be safe - /// to call from arbitrary threads, must never run the function inline and - /// must deliver functions one at a time, in order, on the single JS thread. - /// Dropping a function is only acceptable once the JS runtime is gone. - using JsDispatcher = std::function &&)>; + /// Dispatches a function onto the JS thread, returning whether it was + /// accepted for delivery. Implementations must be safe to call from + /// arbitrary threads, must never run the function inline and must deliver + /// accepted functions one at a time, in order, on the single JS thread. + /// Returning false (and dropping the function) is only acceptable once the + /// JS runtime is gone — callers use the verdict to report outcomes + /// truthfully, e.g. cancel_work only claims success while the cancelled + /// completion can actually be delivered. + using JsDispatcher = std::function &&)>; static std::shared_ptr create(JsDispatcher dispatchToJs); @@ -93,7 +97,14 @@ class HostContext : public std::enable_shared_from_this { /// The struct to pass to hermes_napi_create_env. Owned by this context. hermes_napi_host *host() { return &host_; } - void dispatchToJs(std::function &&fn) { dispatchToJs_(std::move(fn)); } + bool dispatchToJs(std::function &&fn) { + return dispatchToJs_(std::move(fn)); + } + + // host_.data points at this object and the static callbacks cast it back, + // so a copied or moved instance would service callbacks meant for another. + HostContext(const HostContext &) = delete; + HostContext &operator=(const HostContext &) = delete; private: explicit HostContext(JsDispatcher dispatchToJs); diff --git a/packages/host/tests/test_hermes_napi_host.cpp b/packages/host/tests/test_hermes_napi_host.cpp index 06a2bd87..cbe52368 100644 --- a/packages/host/tests/test_hermes_napi_host.cpp +++ b/packages/host/tests/test_hermes_napi_host.cpp @@ -22,18 +22,26 @@ using namespace std::chrono_literals; namespace { // Stands in for the JS thread: functions are queued by the dispatcher (from -// any thread) and only run when the test drains the queue. +// any thread) and only run when the test drains the queue. Flipping +// setAccepting(false) models the CallInvoker expiring on runtime teardown: +// the dispatcher rejects the function and drops it. struct FakeJsQueue { HostContext::JsDispatcher dispatcher() { - return [this](std::function &&fn) { + return [this](std::function &&fn) -> bool { + if (!accepting_.load()) { + return false; + } { std::lock_guard lock(mutex_); queue_.push_back(std::move(fn)); } cv_.notify_all(); + return true; }; } + void setAccepting(bool accepting) { accepting_.store(accepting); } + // Runs queued functions one at a time until the queue is empty, including // functions queued reentrantly while draining. Returns how many ran. size_t drain() { @@ -65,6 +73,7 @@ struct FakeJsQueue { } private: + std::atomic accepting_{true}; std::mutex mutex_; std::condition_variable cv_; std::deque> queue_; @@ -114,6 +123,23 @@ struct GatedWork { // workers keeps a subsequently posted item deterministically queued. constexpr int kWorkerCount = 4; +// Fills every pool worker with its own gate-blocked job, so a subsequently +// posted item deterministically stays queued. Jobs are heap-allocated and +// deliberately leaked: their completions may never run (e.g. when delivery is +// rejected) and workers may still touch them when a test ends. +std::vector saturatePool(hermes_napi_host *host) { + std::vector jobs; + for (int i = 0; i < kWorkerCount; i++) { + auto *job = new GatedWork(); + host->post_work(host->data, job, GatedWork::execute, GatedWork::complete); + jobs.push_back(job); + } + for (auto *job : jobs) { + job->waitForStarted(1); + } + return jobs; +} + } // namespace TEST_CASE("post_work runs execute off the posting thread and delivers " @@ -160,14 +186,9 @@ TEST_CASE("cancel_work cancels queued items and rejects started items") { SECTION("a queued item is cancelled: execute skipped, complete gets " "napi_cancelled, a second cancel fails") { - auto *busy = new GatedWork(); - for (int i = 0; i < kWorkerCount; i++) { - host->post_work(host->data, busy, GatedWork::execute, - GatedWork::complete); - } - busy->waitForStarted(kWorkerCount); + auto busy = saturatePool(host); - // Every worker is blocked on the gate, so this item stays queued. + // Every worker is blocked on a gate, so this item stays queued. auto *target = new GatedWork(); host->post_work(host->data, target, GatedWork::execute, GatedWork::complete); @@ -181,12 +202,14 @@ TEST_CASE("cancel_work cancels queued items and rejects started items") { REQUIRE(target->lastStatus == napi_cancelled); REQUIRE(target->executions.load() == 0); - busy->openGate(); + for (auto *job : busy) { + job->openGate(); + } REQUIRE(js.waitForItems(kWorkerCount)); REQUIRE(js.drain() == kWorkerCount); - REQUIRE(busy->completions.load() == kWorkerCount); - delete busy; - delete target; + for (auto *job : busy) { + REQUIRE(job->completions.load() == 1); + } } SECTION("an item that started executing cannot be cancelled") { @@ -205,6 +228,36 @@ TEST_CASE("cancel_work cancels queued items and rejects started items") { } } +TEST_CASE("queueing the same work item twice drops the duplicate") { + FakeJsQueue js; + auto context = HostContext::create(js.dispatcher()); + hermes_napi_host *host = context->host(); + + auto busy = saturatePool(host); + + auto *target = new GatedWork(); + host->post_work(host->data, target, GatedWork::execute, GatedWork::complete); + // Double-queueing is undefined behavior in Node (libuv aborts); the pool + // drops the duplicate so the one-completion invariant holds. + host->post_work(host->data, target, GatedWork::execute, GatedWork::complete); + + // Exactly one queue entry exists: the first cancel claims it, the second + // finds nothing, and precisely one napi_cancelled completion arrives. + REQUIRE(host->cancel_work(host->data, target)); + REQUIRE(!host->cancel_work(host->data, target)); + REQUIRE(js.waitForItems(1)); + REQUIRE(js.drain() == 1); + REQUIRE(target->completions.load() == 1); + REQUIRE(target->lastStatus == napi_cancelled); + REQUIRE(target->executions.load() == 0); + + for (auto *job : busy) { + job->openGate(); + } + REQUIRE(js.waitForItems(kWorkerCount)); + REQUIRE(js.drain() == kWorkerCount); +} + TEST_CASE("cancel_work racing worker pickup yields exactly one outcome") { FakeJsQueue js; auto context = HostContext::create(js.dispatcher()); @@ -320,26 +373,54 @@ TEST_CASE("post_task delivers exactly once, in order and never inline") { } } -TEST_CASE("work completing after its context died is dropped, not crashed") { +TEST_CASE("a dispatcher that stops accepting (runtime teardown) fails " + "cancellations and drops completions") { + // Models the state production actually reaches: the HostContext is retained + // for the process lifetime, but its dispatcher's CallInvoker expires with + // the runtime, so dispatchToJs starts returning false. FakeJsQueue js; auto context = HostContext::create(js.dispatcher()); hermes_napi_host *host = context->host(); - // Leaked deliberately: complete never runs, so nothing would free it, and - // the worker may still be inside execute when the assertions run. - auto *work = new GatedWork(); - host->post_work(host->data, work, GatedWork::execute, GatedWork::complete); - work->waitForStarted(1); - - // Simulates a React Native runtime teardown: in production the context is - // retained for the process lifetime, but the dispatcher's CallInvoker — - // modelled here by the context itself — can die while work is in flight. - context.reset(); - work->openGate(); - - // The completion cannot be delivered anywhere; give the worker a moment to - // hit the drop path and assert nothing was queued and nothing crashed. - std::this_thread::sleep_for(100ms); - REQUIRE(js.size() == 0); - REQUIRE(work->completions.load() == 0); + SECTION("cancel_work reports failure when the cancelled completion can no " + "longer be delivered") { + auto busy = saturatePool(host); + + auto *target = new GatedWork(); + host->post_work(host->data, target, GatedWork::execute, + GatedWork::complete); + js.setAccepting(false); + // The item is removed from the queue, but the cancelled completion cannot + // be delivered — so the cancellation must not claim success. + REQUIRE(!host->cancel_work(host->data, target)); + for (auto *job : busy) { + job->openGate(); + } + std::this_thread::sleep_for(100ms); + REQUIRE(js.size() == 0); + REQUIRE(target->executions.load() == 0); + REQUIRE(target->completions.load() == 0); + } + + SECTION("a completion for executed work is dropped, not crashed") { + auto *work = new GatedWork(); + host->post_work(host->data, work, GatedWork::execute, GatedWork::complete); + work->waitForStarted(1); + js.setAccepting(false); + work->openGate(); + std::this_thread::sleep_for(100ms); + REQUIRE(js.size() == 0); + REQUIRE(work->completions.load() == 0); + } + + SECTION("a rejected post_task is dropped, not crashed") { + js.setAccepting(false); + std::atomic runs{0}; + host->post_task(host->data, &runs, [](void *data) { + static_cast *>(data)->fetch_add(1); + }); + REQUIRE(js.size() == 0); + REQUIRE(js.drain() == 0); + REQUIRE(runs.load() == 0); + } }