worker_threads: add everysync

Signed-off-by: Matteo Collina <[email protected]>

Author: mcollina

LICENSE

@@ -2639,3 +2639,28 @@ The externally maintained libraries used by Node.js are:
     OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
     SOFTWARE.
   """
+
+- everysync, located at lib/internal/worker/everysync, is licensed as follows:
+  """
+  MIT License
+
+  Copyright (c) 2024 Matteo Collina
+
+  Permission is hereby granted, free of charge, to any person obtaining a copy
+  of this software and associated documentation files (the "Software"), to deal
+  in the Software without restriction, including without limitation the rights
+  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+  copies of the Software, and to permit persons to whom the Software is
+  furnished to do so, subject to the following conditions:
+
+  The above copyright notice and this permission notice shall be included in all
+  copies or substantial portions of the Software.
+
+  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+  SOFTWARE.
+  """

doc/api/worker_threads.md

@@ -1527,6 +1527,78 @@ Calling `unref()` on a worker allows the thread to exit if this is the only
 active handle in the event system. If the worker is already `unref()`ed calling
 `unref()` again has no effect.
 
+## `worker.makeSync(buffer[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `buffer` {SharedArrayBuffer} A shared memory buffer to use for communication.
+* `options` {Object}
+  * `timeout` {number} The timeout in milliseconds for synchronous calls. **Default:** `5000`.
+  * `expandable` {boolean} Whether the buffer can be resized. **Default:** `true` if the buffer supports `growable` option.
+* Returns: {Object} An object with synchronous methods mirroring those exposed through `wire()`.
+
+Creates a synchronous API facade that communicates with a worker thread over a shared memory buffer.
+The worker thread must call `wire()` on the same buffer to register the methods that can be called.
+
+This function enables making synchronous calls to a worker thread, which is particularly useful
+when code requires blocking operations but still wants to benefit from the worker thread's isolation.
+
+```js
+const { Worker, makeSync } = require('node:worker_threads');
+
+// Create a SharedArrayBuffer for communication
+const buffer = new SharedArrayBuffer(1024, {
+  maxByteLength: 64 * 1024 * 1024
+});
+
+// Create a worker, passing the buffer
+const worker = new Worker('worker-script.js', {
+  workerData: { buffer }
+});
+
+// Create a synchronous API facade
+const api = makeSync(buffer);
+
+// Call a method synchronously - this will block until the worker responds
+const result = api.methodName(arg1, arg2);
+```
+
+## `worker.wire(buffer, methods)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `buffer` {SharedArrayBuffer} A shared memory buffer to use for communication.
+* `methods` {Object} An object whose properties are methods to expose to the main thread.
+
+Exposes methods to the main thread that can be called synchronously using `makeSync()`.
+The methods can be async functions or return promises, and the main thread will wait
+for the promise to resolve or reject.
+
+```js
+const { workerData, wire } = require('node:worker_threads');
+
+// Expose methods synchronously to the main thread
+wire(workerData.buffer, {
+  async methodName(arg1, arg2) {
+    // Do work asynchronously
+    return result;
+  },
+
+  syncMethod(arg) {
+    // Do synchronous work
+    return result;
+  }
+});
+```
+
+The `wire()` function should be called early in the worker's lifecycle to register
+the methods before the main thread attempts to call them. Any values returned by
+these methods are serialized and passed back to the main thread.
+
 ## Notes
 
 ### Synchronous blocking of stdio
@@ -1637,6 +1709,8 @@ thread spawned will spawn another until the application crashes.
 [`worker.postMessage()`]: #workerpostmessagevalue-transferlist
 [`worker.terminate()`]: #workerterminate
 [`worker.threadId`]: #workerthreadid_1
+[`worker.makeSync()`]: #workermakesyncbuffer-options
+[`worker.wire()`]: #workerwirebuffer-methods
 [async-resource-worker-pool]: async_context.md#using-asyncresource-for-a-worker-thread-pool
 [browser `MessagePort`]: https://developer.mozilla.org/en-US/docs/Web/API/MessagePort
 [child processes]: child_process.md

lib/internal/worker/everysync/index.js

@@ -0,0 +1,89 @@
+'use strict';
+
+const { read, write } = require('internal/worker/everysync/objects');
+const {
+  OFFSET,
+  TO_MAIN,
+  TO_WORKER,
+} = require('internal/worker/everysync/indexes');
+
+/**
+ * Creates a synchronous API facade from a shared memory buffer.
+ * This function is meant to be used in the main thread to communicate with
+ * a worker thread that has called `wire()` on the same shared memory.
+ *
+ * @param {SharedArrayBuffer} data - The shared memory buffer for communication
+ * @param {Object} [opts={}] - Options object
+ * @param {number} [opts.timeout=1000] - Timeout in milliseconds for synchronous operations
+ * @returns {Object} - An object with methods that match the ones exposed by the worker
+ */
+function makeSync(data, opts = {}) {
+  const timeout = opts.timeout || 1000;
+  const metaView = new Int32Array(data);
+
+  const res = Atomics.wait(metaView, TO_WORKER, 0, timeout);
+  Atomics.store(metaView, TO_WORKER, 0);
+
+  if (res === 'ok') {
+    const obj = read(data, OFFSET);
+
+    const api = {};
+    for (const key of obj) {
+      api[key] = (...args) => {
+        write(data, { key, args }, OFFSET);
+        Atomics.store(metaView, TO_MAIN, 1);
+        Atomics.notify(metaView, TO_MAIN, 1);
+        const res = Atomics.wait(metaView, TO_WORKER, 0, timeout);
+        Atomics.store(metaView, TO_WORKER, 0);
+        if (res === 'ok') {
+          const obj = read(data, OFFSET);
+          return obj;
+        } else {
+          throw new Error(`The response timed out after ${timeout}ms`);
+        }
+      };
+    }
+
+    return api;
+  } else {
+    throw new Error(`The initialization timed out after ${timeout}ms`);
+  }
+}
+
+/**
+ * Wires up a shared memory buffer to invoke methods on an object.
+ * This function is meant to be used in a worker thread to expose methods
+ * to the main thread that has called `makeSync()` on the same shared memory.
+ *
+ * @param {SharedArrayBuffer} data - The shared memory buffer for communication
+ * @param {Object} obj - Object with methods to expose to the main thread
+ * @returns {Promise<void>} - A promise that never resolves unless there's an error
+ */
+async function wire(data, obj) {
+  write(data, Object.keys(obj), OFFSET);
+
+  const metaView = new Int32Array(data);
+
+  Atomics.store(metaView, TO_WORKER, 1);
+  Atomics.notify(metaView, TO_WORKER);
+
+  while (true) {
+    const waitAsync = Atomics.waitAsync(metaView, TO_MAIN, 0);
+    const res = await waitAsync.value;
+    Atomics.store(metaView, TO_MAIN, 0);
+
+    if (res === 'ok') {
+      const { key, args } = read(data, OFFSET);
+      // This is where the magic happens - invoke the requested method
+      const result = await obj[key](...args);
+      write(data, result, OFFSET);
+      Atomics.store(metaView, TO_WORKER, 1);
+      Atomics.notify(metaView, TO_WORKER, 1);
+    }
+  }
+}
+
+module.exports = {
+  makeSync,
+  wire
+};

lib/internal/worker/everysync/indexes.js

@@ -0,0 +1,27 @@
+'use strict';
+
+/**
+ * Byte offset where the actual data begins in the shared memory
+ * @type {number}
+ */
+const OFFSET = 64;
+
+/**
+ * Index in the Int32Array for signaling from worker to main thread
+ * 0: writing from worker, reading from main
+ * @type {number}
+ */
+const TO_WORKER = 0;
+
+/**
+ * Index in the Int32Array for signaling from main to worker thread
+ * 1: writing from main, reading from worker
+ * @type {number}
+ */
+const TO_MAIN = 1;
+
+module.exports = {
+  OFFSET,
+  TO_WORKER,
+  TO_MAIN
+};

lib/internal/worker/everysync/objects.js

@@ -0,0 +1,47 @@
+'use strict';
+
+const { serialize, deserialize } = require('v8');
+
+/**
+ * Reads an object from a shared memory buffer
+ * 
+ * @param {SharedArrayBuffer} buffer - The shared memory buffer containing serialized data
+ * @param {number} [byteOffset=0] - Byte offset where the data begins
+ * @returns {any} - The deserialized object
+ */
+function read(buffer, byteOffset = 0) {
+  const view = new DataView(buffer, byteOffset);
+  const length = view.getUint32(0, true);
+  const object = deserialize(new Uint8Array(buffer, byteOffset + 4, length));
+  return object;
+}
+
+/**
+ * Writes an object to a shared memory buffer
+ * 
+ * @param {SharedArrayBuffer} buffer - The shared memory buffer to write to
+ * @param {any} object - The object to serialize and write
+ * @param {number} [byteOffset=0] - Byte offset where to write the data
+ * @throws {Error} If the buffer is too small and not growable
+ */
+function write(buffer, object, byteOffset = 0) {
+  const data = serialize(object);
+
+  if (buffer.byteLength < data.byteLength + 4 + byteOffset) {
+    // Check if buffer is growable (has grow method from ShareArrayBuffer.prototype)
+    if (typeof buffer.grow !== 'function') {
+      throw new Error('Buffer is too small and not growable');
+    }
+    
+    buffer.grow(data.byteLength + 4 + byteOffset);
+  }
+  
+  const view = new DataView(buffer, byteOffset);
+  view.setUint32(0, data.byteLength, true);
+  new Uint8Array(buffer, byteOffset + 4).set(data);
+}
+
+module.exports = {
+  read,
+  write
+};

lib/worker_threads.js

@@ -29,6 +29,11 @@ const {
   isMarkedAsUntransferable,
 } = require('internal/buffer');
 
+const {
+  makeSync,
+  wire
+} = require('internal/worker/everysync/index');
+
 module.exports = {
   isInternalThread,
   isMainThread,
@@ -49,4 +54,6 @@ module.exports = {
   BroadcastChannel,
   setEnvironmentData,
   getEnvironmentData,
+  makeSync,
+  wire
 };

test/fixtures/everysync/echo.mjs

@@ -0,0 +1,10 @@
+import { workerData, wire } from 'node:worker_threads';
+
+wire(workerData.data, {
+  async echo(arg) {
+    return arg;
+  },
+});
+
+// Keep the event loop alive
+setInterval(() => {}, 100000);

test/fixtures/everysync/failure.mjs

@@ -0,0 +1,9 @@
+import { workerData, wire } from 'node:worker_threads';
+
+wire(workerData.data, {
+  fail(arg) {
+    return new Promise((resolve, reject) => {
+      // nothing to do here, we will fail
+    });
+  },
+});

test/parallel/test-worker-everysync-base.js

@@ -0,0 +1,57 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { join } = require('path');
+const { Worker, makeSync } = require('worker_threads');
+
+// Test makeSync and wire functionality
+{
+  const buffer = new SharedArrayBuffer(1024, {
+    maxByteLength: 64 * 1024 * 1024,
+  });
+  const worker = new Worker(join(__dirname, '..', 'fixtures', 'everysync', 'echo.mjs'), {
+    workerData: {
+      data: buffer,
+    },
+  });
+
+  const api = makeSync(buffer);
+
+  assert.strictEqual(api.echo(42), 42);
+  assert.strictEqual(api.echo('test'), 'test');
+  assert.deepStrictEqual(api.echo({ foo: 'bar' }), { foo: 'bar' });
+
+  worker.terminate();
+}
+
+// Test timeout failure
+{
+  const buffer = new SharedArrayBuffer(1024, {
+    maxByteLength: 64 * 1024 * 1024,
+  });
+  const worker = new Worker(join(__dirname, '..', 'fixtures', 'everysync', 'failure.mjs'), {
+    workerData: {
+      data: buffer,
+    },
+  });
+
+  const api = makeSync(buffer, { timeout: 100 });
+
+  assert.throws(() => api.fail(), {
+    message: 'The response timed out after 100ms'
+  });
+
+  worker.terminate();
+}
+
+// Test initialization timeout
+{
+  const buffer = new SharedArrayBuffer(1024, {
+    maxByteLength: 64 * 1024 * 1024,
+  });
+
+  assert.throws(() => makeSync(buffer, { timeout: 100 }), {
+    message: 'The initialization timed out after 100ms'
+  });
+}