test: run in both a child process and a worker thread

package.json

@@ -281,7 +281,7 @@
     "prebenchmark": "node-gyp rebuild -C benchmark",
     "benchmark": "node benchmark",
     "pretest": "node-gyp rebuild -C test",
-    "test": "node test",
+    "test": "node --experimental-worker test",
     "predev": "node-gyp rebuild -C test --debug",
     "dev": "node test",
     "predev:incremental": "node-gyp configure build -C test --debug",

test/index.js

@@ -1,10 +1,34 @@
 'use strict';
 
+// Command line:
+// None to launch all tests, otherwise
+// $2: name of test
+// $3: mode ("worker" or "child")
+
+const { Worker, workerData, isMainThread } = require('worker_threads');
+
 process.config.target_defaults.default_configuration =
   require('fs')
     .readdirSync(require('path').join(__dirname, 'build'))
     .filter((item) => (item === 'Debug' || item === 'Release'))[0];
 
+if (!isMainThread) {
+  require('./' + workerData);
+  return;
+}
+
+if (process.argv.length === 4) {
+  if (process.argv[3] === 'child') {
+    require('./' + process.argv[2]);
+  } else if (process.argv[3] === 'worker') {
+    (new Worker(__filename, { workerData: process.argv[2] }))
+      .on('exit', (code) => {
+        process.exit(code);
+      });
+  }
+  return;
+}
+
 // FIXME: We might need a way to load test modules automatically without
 // explicit declaration as follows.
 let testModules = [
@@ -60,7 +84,6 @@ let testModules = [
 ];
 
 const napiVersion = Number(process.versions.napi)
-const majorNodeVersion = process.versions.node.split('.')[0]
 
 if (napiVersion < 3) {
   testModules.splice(testModules.indexOf('callbackscope'), 1);
@@ -88,35 +111,28 @@ if (napiVersion < 6) {
   testModules.splice(testModules.indexOf('addon_data'), 1);
 }
 
-if (typeof global.gc === 'function') {
-  console.log(`Testing with N-API Version '${napiVersion}'.`);
-
-  console.log('Starting test suite\n');
-
-  // Requiring each module runs tests in the module.
-  testModules.forEach(name => {
-    console.log(`Running test '${name}'`);
-    require('./' + name);
-  });
-
-  console.log('\nAll tests passed!');
-} else {
-  // Construct the correct (version-dependent) command-line args.
-  let args = ['--expose-gc', '--no-concurrent-array-buffer-freeing'];
-  if (majorNodeVersion >= 14) {
-    args.push('--no-concurrent-array-buffer-sweeping');
-  }
-  args.push(__filename);
-
-  const child = require('./napi_child').spawnSync(process.argv[0], args, {
-    stdio: 'inherit',
-  });
+console.log(`Testing with N-API Version '${napiVersion}'.`);
+console.log('Starting test suite\n');
 
+function runOneChild(name, mode) {
+  console.log(`Running test '${name}' as ${mode}`);
+  const child = require('./napi_child').spawnSync(process.execPath, [
+    __filename, name, mode
+  ], { stdio: 'inherit' });
   if (child.signal) {
-    console.error(`Tests aborted with ${child.signal}`);
-    process.exitCode = 1;
-  } else {
+    console.error(`Test ${name} run as ${mode} aborted with ${child.signal}`);
+    if (!process.exitCode) {
+      process.exitCode = 1;
+    }
+  }
+  if (!process.exitCode && child.status !== 0) {
     process.exitCode = child.status;
   }
-  process.exit(process.exitCode);
 }
+
+testModules.forEach((name) => {
+  runOneChild(name, 'child');
+  runOneChild(name, 'worker');
+});
+
+process.exit(process.exitCode);

test/napi_child.js

@@ -1,14 +1,20 @@
 // Makes sure that child processes are spawned appropriately.
+const child_process = require('child_process');
+
+const majorNodeVersion = process.versions.node.split('.')[0];
+defaultArgs = [
+  '--expose-gc',
+  '--experimental-worker',
+  '--no-concurrent-array-buffer-freeing'
+];
+if (majorNodeVersion >= 14) {
+  defaultArgs.push('--no-concurrent-array-buffer-sweeping');
+}
+
 exports.spawnSync = function(command, args, options) {
-  if (require('../index').needsFlag) {
-    args.splice(0, 0, '--napi-modules');
-  }
-  return require('child_process').spawnSync(command, args, options);
+  return child_process.spawnSync(command, defaultArgs.concat(args), options);
 };
 
 exports.spawn = function(command, args, options) {
-  if (require('../index').needsFlag) {
-    args.splice(0, 0, '--napi-modules');
-  }
-  return require('child_process').spawn(command, args, options);
-};
+  return child_process.spawn(command, defaultArgs.concat(args), options);
+}

test/threadsafe_function/threadsafe_function_sum.cc

@@ -13,12 +13,21 @@ namespace {
 struct TestData {
 
   TestData(Promise::Deferred&& deferred) : deferred(std::move(deferred)) {};
+
+  // These variables are accessed only from the main thread. They keep track of
+  // the number of expected incoming calls that have completed. We do not want
+  // to release the thread-safe function until all expected calls are complete.
+  size_t threadsCreated = 0;
+  size_t callsCompleted = 0;
+  // A value of true for this variable indicates that no more new threads will
+  // be created.
+  bool threadsStopped = false;
 
   // Native Promise returned to JavaScript
   Promise::Deferred deferred;
 
-  // List of threads created for test. This list only ever accessed via main
-  // thread.
+  // List of threads created for test. This list is only ever accessed via the
+  // main thread.
   std::vector<std::thread> threads = {};
 
   ThreadSafeFunction tsfn = ThreadSafeFunction();
@@ -142,11 +151,16 @@ static Value TestDelayedTSFN(const CallbackInfo &info) {
   return testData->deferred.Promise();
 }
 
-void entryAcquire(ThreadSafeFunction tsfn, int threadId) {
+void entryAcquire(ThreadSafeFunction tsfn, int threadId, TestData* testData) {
   tsfn.Acquire();
   std::this_thread::sleep_for(std::chrono::milliseconds(std::rand() % 100 + 1));
   tsfn.BlockingCall( [=](Napi::Env env, Function callback) {
     callback.Call( { Number::New(env, static_cast<double>(threadId))});
+    testData->callsCompleted++;
+    if (testData->threadsStopped &&
+        testData->callsCompleted == testData->threadsCreated) {
+      testData->tsfn.Release();
+    }
   });
   tsfn.Release();
 }
@@ -156,14 +170,15 @@ static Value CreateThread(const CallbackInfo& info) {
   ThreadSafeFunction tsfn = testData->tsfn;
   int threadId = testData->threads.size();
   // A copy of the ThreadSafeFunction will go to the thread entry point
-  testData->threads.push_back( std::thread(entryAcquire, tsfn, threadId) );
+  testData->threads
+    .push_back(std::thread(entryAcquire, tsfn, threadId, testData));
+  testData->threadsCreated++;
   return Number::New(info.Env(), threadId);
 }
 
 static Value StopThreads(const CallbackInfo& info) {
   TestData* testData = static_cast<TestData*>(info.Data());
-  ThreadSafeFunction tsfn = testData->tsfn;
-  tsfn.Release();
+  testData->threadsStopped = true;
   return info.Env().Undefined();
 }