feat: support Atomics.waitAsync (#687)

Co-authored-by: Robert Nagy <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Alan Agius <[email protected]>
Co-authored-by: Tomasz Mikos <[email protected]>
Co-authored-by: Chocobozzz <[email protected]>
Co-authored-by: Charles <[email protected]>
Co-authored-by: Rafael Gonzaga <[email protected]>

Author: 8 people

README.md

@@ -278,7 +278,7 @@ const { resolve } = require('path');
 const Piscina = require('piscina');
 const piscina = new Piscina({
   filename: resolve(__dirname, 'worker.js'),
-  useAtomics: false
+  atomics: 'disabled'
 });
 
 async function main () {
@@ -362,12 +362,16 @@ This class extends [`EventEmitter`][] from Node.js.
     only makes sense to specify if there is some kind of asynchronous component
     to the task. Keep in mind that Worker threads are generally not built for
     handling I/O in parallel.
-  * `useAtomics`: (`boolean`) Use the [`Atomics`][] API for faster communication
+  * `atomics`: (`sync` | `async` | `disabled`) Use the [`Atomics`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics) API for faster communication
     between threads. This is on by default. You can disable `Atomics` globally by
-    setting the environment variable `PISCINA_DISABLE_ATOMICS` to `1`. 
-    If `useAtomics` is `true`, it will cause to pause threads (stoping all execution)
-    between tasks. Ideally, threads should wait for all operations to finish before 
-    returning control to the main thread (avoid having open handles within a thread).
+    setting the environment variable `PISCINA_DISABLE_ATOMICS` to `1` .
+    If `atomics` is `sync`, it will cause to pause threads (stoping all execution)
+    between tasks. Ideally, threads should wait for all operations to finish before
+    returning control to the main thread (avoid having open handles within a thread). If still want to have the possibility
+    of having open handles or handle asynchrnous tasks, you can set the environment variable `PISCINA_ENABLE_ASYNC_ATOMICS` to `1` or setting `options.atomics` to `async`.
+
+   > **Note**: The `async` mode comes with performance penalties and can lead to undesired behaviour if open handles are not tracked correctly.
+  
   * `resourceLimits`: (`object`) See [Node.js new Worker options][]
     * `maxOldGenerationSizeMb`: (`number`) The maximum size of each worker threads
       main heap in MB.

benchmark/simple-benchmark-async.js

@@ -0,0 +1,29 @@
+'use strict';
+const { Piscina } = require('../dist');
+const { resolve } = require('path');
+
+async function simpleBenchmark ({ duration = 10000 } = {}) {
+  const pool = new Piscina({ filename: resolve(__dirname, 'fixtures/add.js'), atomics: 'async' });
+  let done = 0;
+
+  const results = [];
+  const start = process.hrtime.bigint();
+  while (pool.queueSize === 0) {
+    results.push(scheduleTasks());
+  }
+
+  async function scheduleTasks () {
+    while ((process.hrtime.bigint() - start) / 1_000_000n < duration) {
+      await pool.run({ a: 4, b: 6 });
+      done++;
+    }
+  }
+
+  await Promise.all(results);
+
+  return done / duration * 1e3;
+}
+
+simpleBenchmark().then((opsPerSecond) => {
+  console.log(`opsPerSecond: ${opsPerSecond} (with default taskQueue)`);
+});

docs/docs/api-reference/class.md

@@ -47,12 +47,20 @@ This class extends [`EventEmitter`](https://nodejs.org/api/events.html) from Nod
     only makes sense to specify if there is some kind of asynchronous component
     to the task. Keep in mind that Worker threads are generally not built for
     handling I/O in parallel.
-  - `useAtomics`: (`boolean`) Use the [`Atomics`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics) API for faster communication
+  - `atomics`: (`sync` | `async` | `disabled`) Use the [`Atomics`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics) API for faster communication
     between threads. This is on by default. You can disable `Atomics` globally by
-    setting the environment variable `PISCINA_DISABLE_ATOMICS` to `1`.
-    If `useAtomics` is `true`, it will cause to pause threads (stoping all execution)
+    setting the environment variable `PISCINA_DISABLE_ATOMICS` to `1` .
+    If `atomics` is `sync`, it will cause to pause threads (stoping all execution)
     between tasks. Ideally, threads should wait for all operations to finish before
-    returning control to the main thread (avoid having open handles within a thread).
+    returning control to the main thread (avoid having open handles within a thread). If still want to have the possibility
+    of having open handles or handle asynchrnous tasks, you can set the environment variable `PISCINA_ENABLE_ASYNC_ATOMICS` to `1` or setting `options.atomics` to `async`.
+
+**    :::info
+    **Note**: The `async` mode comes with performance penalties and can lead to undesired behaviour if open handles are not tracked correctly.
+    Workers should be designed to wait for all operations to finish before returning control to the main thread, if any background operations are still running
+    `async` can be of help (e.g. for cache warming, etc).
+    :::
+**
   - `resourceLimits`: (`object`) See [Node.js new Worker options](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options)
     - `maxOldGenerationSizeMb`: (`number`) The maximum size of each worker threads
       main heap in MB.
@@ -93,6 +101,9 @@ This class extends [`EventEmitter`](https://nodejs.org/api/events.html) from Nod
   - `workerHistogram`: (`boolean`) By default `false`. It will hint the Worker pool to record statistics for each individual Worker
   - `loadBalancer`: ([`PiscinaLoadBalancer`](#piscinaloadbalancer)) By default, Piscina uses a least-busy algorithm. The `loadBalancer`
     option can be used to provide an alternative implementation. See [Custom Load Balancers](../advanced-topics/loadbalancer.mdx) for additional detail.
+  - `workerHistogram`: (`boolean`) By default `false`. It will hint the Worker pool to record statistics for each individual Worker
+  - `loadBalancer`: ([`PiscinaLoadBalancer`](#piscinaloadbalancer)) By default, Piscina uses a least-busy algorithm. The `loadBalancer`
+    option can be used to provide an alternative implementation. See [Custom Load Balancers](../advanced-topics/loadbalancer.mdx) for additional detail.
 
 :::caution
 Use caution when setting resource limits. Setting limits that are too low may

docs/docs/examples/broadcast.mdx

@@ -20,7 +20,7 @@ const { resolve } = require('path');
 const Piscina = require('piscina');
 const piscina = new Piscina({
   filename: resolve(__dirname, 'worker.js'),
-  useAtomics: false
+  atomics: 'disabled'
 });
 
 async function main () {

examples/broadcast/main.js

@@ -6,8 +6,8 @@ const { resolve } = require('path');
 const Piscina = require('piscina');
 const piscina = new Piscina({
   filename: resolve(__dirname, 'worker.js'),
-  // Set useAtomics to false to avoid threads being blocked when idle
-  useAtomics: false
+  // Set atomics to disabled to avoid threads being blocked when idle
+  atomics: 'disabled'
 });
 
 async function main () {

package.json

@@ -19,13 +19,14 @@
     "test:ci": "npm run lint && npm run build && npm run test:coverage",
     "test:coverage": "c8 --reporter=lcov tap --cov",
     "prepack": "npm run build",
-    "bench": "npm run bench:taskqueue && npm run bench:piscina",
-    "bench:piscina": "npm run benchmark:piscina-default &&npm run benchmark:piscina-fixed-queue && npm run benchmark:piscina-comparison",
-    "bench:taskqueue": "npm run benchmark:queue-comparison",
-    "benchmark:piscina-default": "node benchmark/simple-benchmark.js",
-    "benchmark:piscina-fixed-queue": "node benchmark/simple-benchmark-fixed-queue.js",
-    "benchmark:piscina-comparison": "node benchmark/piscina-queue-comparison.js",
-    "benchmark:queue-comparison": "node benchmark/queue-comparison.js"
+    "benchmark": "npm run bench:queue && npm run benchmark:piscina",
+    "benchmark:piscina": "npm run benchmark:default &&npm run benchmark:queue:fixed && npm run benchmark:default:comparison",
+    "benchmark:default": "node benchmark/simple-benchmark.js",
+    "benchmark:default:async": "node benchmark/simple-benchmark.js",
+    "benchmark:default:comparison": "node benchmark/piscina-queue-comparison.js",
+    "benchmark:queue": "npm run benchmark:queue-comparison",
+    "benchmark:queue:fixed": "node benchmark/simple-benchmark-fixed-queue.js",
+    "benchmark:queue:comparison": "node benchmark/queue-comparison.js"
   },
   "repository": {
     "type": "git",

src/index.ts

@@ -65,7 +65,7 @@ interface Options {
   idleTimeout? : number,
   maxQueue? : number | 'auto',
   concurrentTasksPerWorker? : number,
-  useAtomics? : boolean,
+  atomics? : 'sync' | 'async' | 'disabled',
   resourceLimits? : ResourceLimits,
   argv? : string[],
   execArgv? : string[],
@@ -88,7 +88,7 @@ interface FilledOptions extends Options {
   idleTimeout : number,
   maxQueue : number,
   concurrentTasksPerWorker : number,
-  useAtomics: boolean,
+  atomics: Options['atomics'],
   taskQueue : TaskQueue,
   niceIncrement : number,
   closeTimeout : number,
@@ -122,7 +122,7 @@ const kDefaultOptions : FilledOptions = {
   idleTimeout: 0,
   maxQueue: Infinity,
   concurrentTasksPerWorker: 1,
-  useAtomics: true,
+  atomics: 'sync',
   taskQueue: new ArrayTaskQueue(),
   niceIncrement: 0,
   trackUnmanagedFds: true,
@@ -274,7 +274,7 @@ class ThreadPool {
       name: this.options.name,
       port: port2,
       sharedBuffer: workerInfo.sharedBuffer,
-      useAtomics: this.options.useAtomics,
+      atomics: this.options.atomics!,
       niceIncrement: this.options.niceIncrement
     };
     worker.postMessage(message, [port2]);
@@ -379,7 +379,7 @@ class ThreadPool {
   }
 
   _processPendingMessages () {
-    if (this.inProcessPendingMessages || !this.options.useAtomics) {
+    if (this.inProcessPendingMessages || this.options.atomics === 'disabled') {
       return;
     }
 
@@ -755,9 +755,9 @@ export default class Piscina<T = any, R = any> extends EventEmitterAsyncResource
       throw new TypeError(
         'options.concurrentTasksPerWorker must be a positive integer');
     }
-    if (options.useAtomics !== undefined &&
-        typeof options.useAtomics !== 'boolean') {
-      throw new TypeError('options.useAtomics must be a boolean value');
+    if (options.atomics != null && (typeof options.atomics !== 'string' ||
+        !['sync', 'async', 'disabled'].includes(options.atomics))) {
+      throw new TypeError('options.atomics should be a value of sync, sync or disabled.');
     }
     if (options.resourceLimits !== undefined &&
         (typeof options.resourceLimits !== 'object' ||

src/types.ts

@@ -8,7 +8,7 @@ export interface StartupMessage {
   name: string
   port: MessagePort
   sharedBuffer: Int32Array
-  useAtomics: boolean
+  atomics: 'async' | 'sync' | 'disabled'
   niceIncrement: number
 }