[WIP] Implement ThreadSafeFunction class

This PR is implementing ThreadSafeFunction class wraps
napi_threadsafe_function*.

Fixes nodejs#312.

napi-inl.h

@@ -3676,6 +3676,59 @@ inline void AsyncWorker::OnWorkComplete(
   delete self;
 }
 
+////////////////////////////////////////////////////////////////////////////////
+// ThreadSafeFunction class
+////////////////////////////////////////////////////////////////////////////////
+
+inline ThreadSafeFunction::ThreadSafeFunction() {
+}
+
+inline ThreadSafeFunction::ThreadSafeFunction(napi_env env,
+                                              const Function& callback,
+                                              const char* resource_name,
+                                              const Object& resource,
+                                              size_t max_queue_size,
+                                              size_t initial_thread_count)
+  : _env(env) {
+  napi_value resource_id;
+  napi_status status = napi_create_string_utf8(
+      _env, resource_name, NAPI_AUTO_LENGTH, &resource_id);
+  NAPI_THROW_IF_FAILED_VOID(_env, status);
+
+  status = napi_create_threadsafe_function(
+      _env, callback, resource, resource_id, max_queue_size,
+      initial_thread_count, nullptr, nullptr, nullptr, CallJS,
+      &_ts_fn);
+  NAPI_THROW_IF_FAILED_VOID(_env, status);
+}
+
+inline napi_status ThreadSafeFunction::Call(Callback callback, napi_threadsafe_function_call_mode mode) const {
+  Callback* callback_container = new Callback();
+  *callback_container = std::move(callback);
+  return napi_call_threadsafe_function(_ts_fn, callback_container, mode);
+}
+
+inline bool ThreadSafeFunction::Acquire() const {
+  return napi_acquire_threadsafe_function(_ts_fn) == napi_ok;
+}
+
+inline bool ThreadSafeFunction::Release(napi_threadsafe_function_release_mode mode) const {
+  return napi_release_threadsafe_function(_ts_fn, mode) == napi_ok;
+}
+
+//static
+inline void ThreadSafeFunction::CallJS(napi_env env,
+                                       napi_value js_callback,
+                                       void* /* hint */,
+                                       void* data) {
+  if (env == nullptr && js_callback == nullptr)
+    return;
+
+  Callback* callback_container = static_cast<Callback*>(data);
+  (*callback_container)(env, Function(env, js_callback));
+  delete callback_container;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // Memory Management class
 ////////////////////////////////////////////////////////////////////////////////

napi.h

@@ -1,6 +1,7 @@
 #ifndef SRC_NAPI_H_
 #define SRC_NAPI_H_
 
+#define NAPI_EXPERIMENTAL
 #include "node_api.h"
 #include <functional>
 #include <initializer_list>
@@ -1760,6 +1761,32 @@ namespace Napi {
     std::string _error;
   };
 
+  class ThreadSafeFunction {
+  public:
+    using Callback = std::function<void(Napi::Env, Napi::Function)>;
+
+    ThreadSafeFunction();
+    ThreadSafeFunction(napi_env env,
+                       const Function& callback,
+                       const char* resource_name,
+                       const Object& resource,
+                       size_t max_queue_size,
+                       size_t initial_thread_count);
+
+    napi_status Call(Callback callback, napi_threadsafe_function_call_mode mode) const;
+    bool Acquire() const;
+    bool Release(napi_threadsafe_function_release_mode mode) const;
+
+  private:
+    static void CallJS(napi_env env,
+                       napi_value callback,
+                       void* hint,
+                       void* data);
+
+    napi_env _env;
+    napi_threadsafe_function _ts_fn;
+  };
+
   // Memory management.
   class MemoryManagement {
     public:

package.json

@@ -53,7 +53,7 @@
     "url": "git://github.com/nodejs/node-addon-api.git"
   },
   "scripts": {
-    "pretest": "node-gyp rebuild -C test",
+    "pretest": "node-gyp build -C test",
     "test": "node test",
     "doc": "doxygen doc/Doxyfile"
   },

test/binding.cc

@@ -32,6 +32,9 @@ Object InitObject(Env env);
 Object InitObjectDeprecated(Env env);
 #endif // !NODE_ADDON_API_DISABLE_DEPRECATED
 Object InitPromise(Env env);
+#if (NAPI_VERSION > 3)
+Object InitThreadSafeFunction(Env env);
+#endif
 Object InitTypedArray(Env env);
 Object InitObjectWrap(Env env);
 Object InitObjectReference(Env env);
@@ -69,6 +72,9 @@ Object Init(Env env, Object exports) {
   exports.Set("object_deprecated", InitObjectDeprecated(env));
 #endif // !NODE_ADDON_API_DISABLE_DEPRECATED
   exports.Set("promise", InitPromise(env));
+#if (NAPI_VERSION > 3)
+  exports.Set("threadsafe_function", InitThreadSafeFunction(env));
+#endif
   exports.Set("typedarray", InitTypedArray(env));
   exports.Set("objectwrap", InitObjectWrap(env));
   exports.Set("objectreference", InitObjectReference(env));

test/binding.gyp

@@ -31,6 +31,7 @@
         'object/object.cc',
         'object/set_property.cc',
         'promise.cc',
+        'threadsafe_function/threadsafe_function.cc',
         'typedarray.cc',
         'objectwrap.cc',
         'objectreference.cc',

test/index.js

@@ -34,6 +34,7 @@ let testModules = [
   'object/object_deprecated',
   'object/set_property',
   'promise',
+  'threadsafe_function/threadsafe_function',
   'typedarray',
   'typedarray-bigint',
   'objectwrap',

test/threadsafe_function/threadsafe_function.cc

@@ -0,0 +1,85 @@
+#define NAPI_EXPERIMENTAL
+#include <uv.h>
+#include "napi.h"
+
+using namespace Napi;
+
+const size_t ARRAY_LENGTH = 10;
+const size_t MAX_QUEUE_SIZE = 2;
+
+// Thread data to transmit to JS
+static int ints[ARRAY_LENGTH];
+
+static uv_thread_t uv_threads[2];
+static napi_threadsafe_function ts_fn;
+static ThreadSafeFunction ts_fn_wrap;
+
+int sum = 0;
+
+static void data_source_thread(void* data) {
+  int args = 15;
+  ts_fn_wrap.Call([args](Napi::Env env, Napi::Function jsCallback) {
+    jsCallback.Call({ Number::New(env, args) });
+  }, napi_tsfn_blocking);
+  sum += 5;
+  if (data) {}
+}
+
+static Value StartThreadInternal(const CallbackInfo& info, bool block_on_full) {
+  ts_fn_wrap = ThreadSafeFunction(info.Env(), info[0].As<Function>(), "", Object(), 10, 2);
+
+  if (uv_thread_create(&uv_threads[0], data_source_thread, (void*)ts_fn) == 0) {
+    printf("created thread1\n");
+  }
+  if (uv_thread_create(&uv_threads[1], data_source_thread, (void*)ts_fn) == 0) {
+    printf("created thread2\n");
+  }
+
+  uv_thread_join(&uv_threads[1]);
+  uv_thread_join(&uv_threads[0]);
+  printf("%d\n", sum);
+  if (block_on_full) {}
+
+  return Object::New(info.Env());
+}
+
+static Value StartThread(const CallbackInfo& info) {
+  return StartThreadInternal(info, true);
+}
+
+static Value StartThreadNoNative(const CallbackInfo& info) {
+  return StartThreadInternal(info, true);
+}
+
+static Value StartThreadNonblocking(const CallbackInfo& info) {
+  return StartThreadInternal(info, false);
+}
+
+static Value StopThread(const CallbackInfo& info) {
+  return Object::New(info.Env());
+}
+
+static Value Unref(const CallbackInfo& info) {
+  return Object::New(info.Env());
+}
+
+static Value Release(const CallbackInfo& info) {
+  return Object::New(info.Env());
+}
+
+Object InitThreadSafeFunction(Env env) {
+  for (size_t index = 0; index < ARRAY_LENGTH; index++) {
+    ints[index] = index;
+  }
+
+  Object exports = Object::New(env);
+  exports["ARRAY_LENGTH"] = Number::New(env, ARRAY_LENGTH);
+  exports["MAX_QUEUE_SIZE"] = Number::New(env, MAX_QUEUE_SIZE);
+  exports["startThread"] = Function::New(env, StartThread);
+  exports["startThreadNoNative"] = Function::New(env, StartThreadNoNative);
+  exports["startThreadNonblocking"] = Function::New(env, StartThreadNonblocking);
+  exports["stopThread"] = Function::New(env, StopThread);
+  exports["unref"] = Function::New(env, Unref);
+  exports["release"] = Function::New(env, Release);
+  return exports;
+}

test/threadsafe_function/threadsafe_function.js

@@ -0,0 +1,15 @@
+'use strict';
+
+const buildType = process.config.target_defaults.default_configuration;
+const assert = require('assert');
+
+test(require(`../build/${buildType}/binding.node`));
+test(require(`../build/${buildType}/binding_noexcept.node`));
+
+function test(binding) {
+  binding.threadsafe_function.startThread(function(number) {
+    console.log("callback-ok?" + number);
+  }, "hello");
+  console.log(binding.threadsafe_function.ARRAY_LENGTH);
+  console.log(binding.threadsafe_function.MAX_QUEUE_SIZE);
+}