fix: execute tsfn finalizer after queue drains when aborted (nodejs/node#61956)

Author: toyobayashi

packages/emnapi/src/threadsafe-function.ts

@@ -524,7 +524,7 @@ const emnapiTSFN = {
     emnapiTSFN.releaseResources(func)
     _free(to64('func') as number)
   },
-  emptyQueueAndMaybeDelete (func: number) {
+  emptyQueue (func: number) {
     const drainQueue: number[] = []
     emnapiTSFN.getMutex(func).execute(() => {
       while (emnapiTSFN.getQueueSize(func) > 0) {
@@ -541,6 +541,8 @@ const emnapiTSFN = {
         makeDynCall('vpppp', 'callJsCb')(to64('0'), to64('0'), to64('context'), to64('data'))
       }
     }
+  },
+  maybeDelete (func: number) {
     let shouldDelete = false
     emnapiTSFN.getMutex(func).execute(() => {
       if (emnapiTSFN.getThreadCount(func) > 0) {
@@ -574,6 +576,7 @@ const emnapiTSFN = {
     }
 
     try {
+      emnapiTSFN.emptyQueue(func)
       if (finalize) {
         if (emnapiNodeBinding) {
           const resource = emnapiTSFN.getResource(func)
@@ -590,7 +593,7 @@ const emnapiTSFN = {
           f()
         }
       }
-      emnapiTSFN.emptyQueueAndMaybeDelete(func)
+      emnapiTSFN.maybeDelete(func)
     } finally {
       emnapiCtx.closeScope(envObject)
     }

packages/emnapi/src/threadsafe_function.c

@@ -167,7 +167,7 @@ static napi_status _emnapi_tsfn_init(napi_threadsafe_function func) {
   return napi_generic_failure;
 }
 
-static void _emnapi_tsfn_empty_queue_and_maybe_delete(napi_threadsafe_function func) {
+static void _emnapi_tsfn_empty_queue(napi_threadsafe_function func) {
   struct uv__queue drain_queue;
   uv__queue_init(&drain_queue);
   pthread_mutex_lock(&func->mutex);
@@ -186,6 +186,9 @@ static void _emnapi_tsfn_empty_queue_and_maybe_delete(napi_threadsafe_function f
     uv__queue_init(q);
     free(node);
   }
+}
+
+static void _emnapi_tsfn_maybe_delete(napi_threadsafe_function func) {
   pthread_mutex_lock(&func->mutex);
   if (func->thread_count > 0) {
     _emnapi_tsfn_release_resources(func);
@@ -207,6 +210,7 @@ static napi_value _emnapi_tsfn_finalize_in_callback_scope(napi_env env, napi_cal
 
 static void _emnapi_tsfn_finalize(napi_threadsafe_function func) {
   napi_handle_scope scope = _emnapi_open_handle_scope();
+  _emnapi_tsfn_empty_queue(func);
   if (func->finalize_cb) {
     if (emnapi_is_node_binding_available()) {
       napi_value resource, cb;
@@ -224,7 +228,7 @@ static void _emnapi_tsfn_finalize(napi_threadsafe_function func) {
       _emnapi_call_finalizer(0, func->env, func->finalize_cb, func->finalize_data, func->context);
     }
   }
-  _emnapi_tsfn_empty_queue_and_maybe_delete(func);
+  _emnapi_tsfn_maybe_delete(func);
   _emnapi_close_handle_scope(scope);
 }
 

packages/test/CMakeLists.txt

@@ -303,6 +303,8 @@ if((NOT IS_WASM) OR IS_EMSCRIPTEN OR IS_WASI_THREADS)
   add_test("pool" "./pool/binding.c" ON)
   add_test("tsfn" "./tsfn/binding.c" ON)
   add_test("tsfn_shutdown" "./tsfn_shutdown/binding.cc" ON)
+  add_test("tsfn_abort" "./tsfn_abort/binding.cc" ON)
+  target_compile_options("tsfn_abort" PRIVATE "-fno-rtti")
   add_test("async_cleanup_hook" "./async_cleanup_hook/binding.c" ON)
   add_test("uv_threadpool_size" "./uv_threadpool_size/binding.c" ON)
   add_test("trap_in_thread" "./trap_in_thread/binding.c" ON)

packages/test/script/test.js

@@ -22,6 +22,7 @@ const pthread = [
   'pool/**/*',
   'tsfn/**/*',
   'tsfn_shutdown/**/*',
+  'tsfn_abort/**/*',
   'async_cleanup_hook/**/*',
   'string/string-pthread.test.js',
   'uv_threadpool_size/**/*',

packages/test/tsfn_abort/binding.cc

@@ -0,0 +1,111 @@
+#include <js_native_api.h>
+#include <node_api.h>
+#include <node_api_types.h>
+
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <type_traits>
+
+template <typename R, auto func, typename... Args>
+inline auto call(const char* name, Args&&... args) -> R {
+  napi_status status;
+  if constexpr (std::is_same_v<R, void>) {
+    status = func(std::forward<Args>(args)...);
+    if (status == napi_ok) {
+      return;
+    }
+  } else {
+    R ret;
+    status = func(std::forward<Args>(args)..., &ret);
+    if (status == napi_ok) {
+      return ret;
+    }
+  }
+  std::fprintf(stderr, "%s: %d\n", name, status);
+  std::abort();
+}
+
+#define NAPI_CALL(ret_type, func, ...)                                         \
+  call<ret_type, func>(#func, ##__VA_ARGS__)
+
+class Context {
+ public:
+  ~Context() {
+    std::fprintf(stderr, "Context: destructor called\n");
+    get = [](void*) {
+      std::abort();
+      return 0;
+    };
+  }
+
+  std::function<int*(int)> create = [](int value) {
+    std::fprintf(stderr, "Context: create called\n");
+    return new int(value);
+  };
+
+  std::function<int(void*)> get = [](void* ptr) {
+    std::fprintf(stderr, "Context: get called\n");
+    return *static_cast<int*>(ptr);
+  };
+
+  std::function<void(void*)> deleter = [](void* ptr) {
+    std::fprintf(stderr, "Context: deleter called\n");
+    delete static_cast<int*>(ptr);
+  };
+};
+
+void tsfn_callback(napi_env env, napi_value js_cb, void* ctx_p, void* data) {
+  auto ctx = static_cast<Context*>(ctx_p);
+  std::fprintf(stderr, "tsfn_callback: env=%p data=%d\n", env, ctx->get(data));
+  ctx->deleter(data);
+}
+
+void tsfn_finalize(napi_env env, void* finalize_data, void* finalize_hint) {
+  auto ctx = static_cast<Context*>(finalize_hint);
+  std::fprintf(stderr,
+               "tsfn_finalize: env=%p finalize_data=%p finalize_hint=%p\n",
+               env,
+               finalize_data,
+               finalize_hint);
+  delete ctx;
+}
+
+auto run(napi_env env, napi_callback_info info) -> napi_value {
+  auto global = NAPI_CALL(napi_value, napi_get_global, env);
+  auto undefined = NAPI_CALL(napi_value, napi_get_undefined, env);
+  auto ctx = new Context();
+  auto tsfn = NAPI_CALL(napi_threadsafe_function,
+                        napi_create_threadsafe_function,
+                        env,
+                        nullptr,
+                        global,
+                        undefined,
+                        0,
+                        1 /* initial_thread_count */,
+                        nullptr,
+                        tsfn_finalize,
+                        ctx,
+                        tsfn_callback);
+
+  NAPI_CALL(void,
+            napi_call_threadsafe_function,
+            tsfn,
+            ctx->create(1),
+            napi_tsfn_blocking);
+
+  NAPI_CALL(void, napi_unref_threadsafe_function, env, tsfn);
+
+  NAPI_CALL(void,
+            napi_release_threadsafe_function,
+            tsfn,
+            napi_threadsafe_function_release_mode::napi_tsfn_abort);
+  return NAPI_CALL(napi_value, napi_get_undefined, env);
+}
+
+napi_value init(napi_env env, napi_value exports) {
+  return NAPI_CALL(
+      napi_value, napi_create_function, env, nullptr, 0, run, nullptr);
+}
+
+NAPI_MODULE(NODE_GYP_MODULE_NAME, init)

packages/test/tsfn_abort/tsfn_abort.test.js

@@ -0,0 +1,8 @@
+'use strict'
+const { load } = require('../util')
+
+module.exports = new Promise((resolve, reject) => {
+  load('tsfn_abort', { nodeBinding: require('@emnapi/node-binding') }).then((binding) => {
+    binding()
+  }).catch(reject)
+})