worker: initial implementation

Implement multi-threading support for most of the API.

Thanks to Stephen Belanger for reviewing this change in its
original form, to Olivia Hugger for reviewing the
documentation and some of the tests coming along with it,
and to Alexey Orlenko and Timothy Gu for reviewing other
parts of the tests.

Refs: https://github.com/ayojs/ayo/pull/110
Refs: https://github.com/ayojs/ayo/pull/114
Refs: https://github.com/ayojs/ayo/pull/117

PR-URL: https://github.com/nodejs/node/pull/20876
Reviewed-By: Gireesh Punathil <gpunathi@in.ibm.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Shingo Inoue <leko.noor@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Tiancheng "Timothy" Gu <timothygu99@gmail.com>
Reviewed-By: John-David Dalton <john.david.dalton@gmail.com>
Reviewed-By: Gus Caplan <me@gus.host>

9 files changed, 272 insertions, 12 deletions

diff --git a/lib/inspector.js b/lib/inspector.js
index 3285c1040a..f4ec71fd6c 100644
--- a/lib/inspector.js
+++ b/lib/inspector.js
@@ -12,7 +12,7 @@ const {
 const util = require('util');
 const { Connection, open, url } = process.binding('inspector');
 
-if (!Connection)
+if (!Connection || !require('internal/worker').isMainThread)
   throw new ERR_INSPECTOR_NOT_AVAILABLE();
 
 const connectionSymbol = Symbol('connectionProperty');
diff --git a/lib/internal/bootstrap/node.js b/lib/internal/bootstrap/node.js
index 6477c2d828..4817ec110a 100644
--- a/lib/internal/bootstrap/node.js
+++ b/lib/internal/bootstrap/node.js
@@ -24,6 +24,7 @@
                                 _shouldAbortOnUncaughtToggle },
                               { internalBinding, NativeModule }) {
   const exceptionHandlerState = { captureFn: null };
+  const isMainThread = internalBinding('worker').threadId === 0;
 
   function startup() {
     const EventEmitter = NativeModule.require('events');
@@ -100,7 +101,9 @@
       NativeModule.require('internal/inspector_async_hook').setup();
     }
 
-    _process.setupChannel();
+    if (isMainThread)
+      _process.setupChannel();
+
     _process.setupRawDebug(_rawDebug);
 
     const browserGlobals = !process._noBrowserGlobals;
@@ -175,8 +178,11 @@
     // are running from a script and running the REPL - but there are a few
     // others like the debugger or running --eval arguments. Here we decide
     // which mode we run in.
-
-    if (NativeModule.exists('_third_party_main')) {
+    if (internalBinding('worker').getEnvMessagePort() !== undefined) {
+      // This means we are in a Worker context, and any script execution
+      // will be directed by the worker module.
+      NativeModule.require('internal/worker').setupChild(evalScript);
+    } else if (NativeModule.exists('_third_party_main')) {
       // To allow people to extend Node in different ways, this hook allows
       // one to drop a file lib/_third_party_main.js into the build
       // directory which will be executed instead of Node's normal loading.
@@ -542,7 +548,7 @@
     return `process.binding('inspector').callAndPauseOnStart(${fn}, {})`;
   }
 
-  function evalScript(name) {
+  function evalScript(name, body = wrapForBreakOnFirstLine(process._eval)) {
     const CJSModule = NativeModule.require('internal/modules/cjs/loader');
     const path = NativeModule.require('path');
     const cwd = tryGetCwd(path);
@@ -550,7 +556,6 @@
     const module = new CJSModule(name);
     module.filename = path.join(cwd, name);
     module.paths = CJSModule._nodeModulePaths(cwd);
-    const body = wrapForBreakOnFirstLine(process._eval);
     const script = `global.__filename = ${JSON.stringify(name)};\n` +
                    'global.exports = exports;\n' +
                    'global.module = module;\n' +
diff --git a/lib/internal/errors.js b/lib/internal/errors.js
index 89c0139f8b..d59531debb 100644
--- a/lib/internal/errors.js
+++ b/lib/internal/errors.js
@@ -844,4 +844,9 @@ E('ERR_VM_MODULE_NOT_LINKED',
 E('ERR_VM_MODULE_NOT_MODULE',
   'Provided module is not an instance of Module', Error);
 E('ERR_VM_MODULE_STATUS', 'Module status %s', Error);
+E('ERR_WORKER_NEED_ABSOLUTE_PATH',
+  'The worker script filename must be an absolute path. Received "%s"',
+  TypeError);
+E('ERR_WORKER_UNSERIALIZABLE_ERROR',
+  'Serializing an uncaught exception failed', Error);
 E('ERR_ZLIB_INITIALIZATION_FAILED', 'Initialization failed', Error);
diff --git a/lib/internal/process.js b/lib/internal/process.js
index 0f0e40d6a0..f01be32be4 100644
--- a/lib/internal/process.js
+++ b/lib/internal/process.js
@@ -16,6 +16,7 @@ const util = require('util');
 const constants = process.binding('constants').os.signals;
 const assert = require('assert').strict;
 const { deprecate } = require('internal/util');
+const { isMainThread } = require('internal/worker');
 
 process.assert = deprecate(
   function(x, msg) {
@@ -186,6 +187,11 @@ function setupKillAndExit() {
 
 
 function setupSignalHandlers() {
+  if (!isMainThread) {
+    // Worker threads don't receive signals.
+    return;
+  }
+
   const signalWraps = Object.create(null);
   let Signal;
 
diff --git a/lib/internal/process/methods.js b/lib/internal/process/methods.js
index 91aca398b3..9a954f6a9b 100644
--- a/lib/internal/process/methods.js
+++ b/lib/internal/process/methods.js
@@ -8,11 +8,18 @@ const {
   validateMode,
   validateUint32
 } = require('internal/validators');
+const {
+  isMainThread
+} = require('internal/worker');
 
 function setupProcessMethods(_chdir, _cpuUsage, _hrtime, _memoryUsage,
                              _rawDebug, _umask, _initgroups, _setegid,
                              _seteuid, _setgid, _setuid, _setgroups) {
   // Non-POSIX platforms like Windows don't have certain methods.
+  // Workers also lack these methods since they change process-global state.
+  if (!isMainThread)
+    return;
+
   if (_setgid !== undefined) {
     setupPosixMethods(_initgroups, _setegid, _seteuid,
                       _setgid, _setuid, _setgroups);
diff --git a/lib/internal/process/stdio.js b/lib/internal/process/stdio.js
index eaba4dfca1..76e6ab8514 100644
--- a/lib/internal/process/stdio.js
+++ b/lib/internal/process/stdio.js
@@ -6,6 +6,7 @@ const {
   ERR_UNKNOWN_STDIN_TYPE,
   ERR_UNKNOWN_STREAM_TYPE
 } = require('internal/errors').codes;
+const { isMainThread } = require('internal/worker');
 
 exports.setup = setupStdio;
 
@@ -16,6 +17,8 @@ function setupStdio() {
 
   function getStdout() {
     if (stdout) return stdout;
+    if (!isMainThread)
+      return new (require('stream').Writable)({ write(b, e, cb) { cb(); } });
     stdout = createWritableStdioStream(1);
     stdout.destroySoon = stdout.destroy;
     stdout._destroy = function(er, cb) {
@@ -31,6 +34,8 @@ function setupStdio() {
 
   function getStderr() {
     if (stderr) return stderr;
+    if (!isMainThread)
+      return new (require('stream').Writable)({ write(b, e, cb) { cb(); } });
     stderr = createWritableStdioStream(2);
     stderr.destroySoon = stderr.destroy;
     stderr._destroy = function(er, cb) {
@@ -46,6 +51,8 @@ function setupStdio() {
 
   function getStdin() {
     if (stdin) return stdin;
+    if (!isMainThread)
+      return new (require('stream').Readable)({ read() { this.push(null); } });
 
     const tty_wrap = process.binding('tty_wrap');
     const fd = 0;
diff --git a/lib/internal/util/inspector.js b/lib/internal/util/inspector.js
index 634d330233..3dd73415de 100644
--- a/lib/internal/util/inspector.js
+++ b/lib/internal/util/inspector.js
@@ -1,6 +1,8 @@
 'use strict';
 
-const hasInspector = process.config.variables.v8_enable_inspector === 1;
+// TODO(addaleax): Figure out how to integrate the inspector with workers.
+const hasInspector = process.config.variables.v8_enable_inspector === 1 &&
+  require('internal/worker').isMainThread;
 const inspector = hasInspector ? require('inspector') : undefined;
 
 let session;
diff --git a/lib/internal/worker.js b/lib/internal/worker.js
index 73f7525aa7..c982478b93 100644
--- a/lib/internal/worker.js
+++ b/lib/internal/worker.js
@@ -1,24 +1,49 @@
 'use strict';
 
+const Buffer = require('buffer').Buffer;
 const EventEmitter = require('events');
+const assert = require('assert');
+const path = require('path');
 const util = require('util');
+const {
+  ERR_INVALID_ARG_TYPE,
+  ERR_WORKER_NEED_ABSOLUTE_PATH,
+  ERR_WORKER_UNSERIALIZABLE_ERROR
+} = require('internal/errors').codes;
 
 const { internalBinding } = require('internal/bootstrap/loaders');
 const { MessagePort, MessageChannel } = internalBinding('messaging');
 const { handle_onclose } = internalBinding('symbols');
+const { clearAsyncIdStack } = require('internal/async_hooks');
 
 util.inherits(MessagePort, EventEmitter);
 
+const {
+  Worker: WorkerImpl,
+  getEnvMessagePort,
+  threadId
+} = internalBinding('worker');
+
+const isMainThread = threadId === 0;
+
 const kOnMessageListener = Symbol('kOnMessageListener');
+const kHandle = Symbol('kHandle');
+const kPort = Symbol('kPort');
+const kPublicPort = Symbol('kPublicPort');
+const kDispose = Symbol('kDispose');
+const kOnExit = Symbol('kOnExit');
+const kOnMessage = Symbol('kOnMessage');
+const kOnCouldNotSerializeErr = Symbol('kOnCouldNotSerializeErr');
+const kOnErrorMessage = Symbol('kOnErrorMessage');
 
 const debug = util.debuglog('worker');
 
-// A MessagePort consists of a handle (that wraps around an
+// A communication channel consisting of a handle (that wraps around an
 // uv_async_t) which can receive information from other threads and emits
 // .onmessage events, and a function used for sending data to a MessagePort
 // in some other thread.
 MessagePort.prototype[kOnMessageListener] = function onmessage(payload) {
-  debug('received message', payload);
+  debug(`[${threadId}] received message`, payload);
   // Emit the deserialized object to userland.
   this.emit('message', payload);
 };
@@ -79,6 +104,9 @@ MessagePort.prototype.close = function(cb) {
   originalClose.call(this);
 };
 
+const drainMessagePort = MessagePort.prototype.drain;
+delete MessagePort.prototype.drain;
+
 function setupPortReferencing(port, eventEmitter, eventName) {
   // Keep track of whether there are any workerMessage listeners:
   // If there are some, ref() the channel so it keeps the event loop alive.
@@ -99,7 +127,194 @@ function setupPortReferencing(port, eventEmitter, eventName) {
   });
 }
 
+
+class Worker extends EventEmitter {
+  constructor(filename, options = {}) {
+    super();
+    debug(`[${threadId}] create new worker`, filename, options);
+    if (typeof filename !== 'string') {
+      throw new ERR_INVALID_ARG_TYPE('filename', 'string', filename);
+    }
+
+    if (!options.eval && !path.isAbsolute(filename)) {
+      throw new ERR_WORKER_NEED_ABSOLUTE_PATH(filename);
+    }
+
+    // Set up the C++ handle for the worker, as well as some internal wiring.
+    this[kHandle] = new WorkerImpl();
+    this[kHandle].onexit = (code) => this[kOnExit](code);
+    this[kPort] = this[kHandle].messagePort;
+    this[kPort].on('message', (data) => this[kOnMessage](data));
+    this[kPort].start();
+    this[kPort].unref();
+    debug(`[${threadId}] created Worker with ID ${this.threadId}`);
+
+    const { port1, port2 } = new MessageChannel();
+    this[kPublicPort] = port1;
+    this[kPublicPort].on('message', (message) => this.emit('message', message));
+    setupPortReferencing(this[kPublicPort], this, 'message');
+    this[kPort].postMessage({
+      type: 'loadScript',
+      filename,
+      doEval: !!options.eval,
+      workerData: options.workerData,
+      publicPort: port2
+    }, [port2]);
+    // Actually start the new thread now that everything is in place.
+    this[kHandle].startThread();
+  }
+
+  [kOnExit](code) {
+    debug(`[${threadId}] hears end event for Worker ${this.threadId}`);
+    drainMessagePort.call(this[kPublicPort]);
+    this[kDispose]();
+    this.emit('exit', code);
+    this.removeAllListeners();
+  }
+
+  [kOnCouldNotSerializeErr]() {
+    this.emit('error', new ERR_WORKER_UNSERIALIZABLE_ERROR());
+  }
+
+  [kOnErrorMessage](serialized) {
+    // This is what is called for uncaught exceptions.
+    const error = deserializeError(serialized);
+    this.emit('error', error);
+  }
+
+  [kOnMessage](message) {
+    switch (message.type) {
+      case 'upAndRunning':
+        return this.emit('online');
+      case 'couldNotSerializeError':
+        return this[kOnCouldNotSerializeErr]();
+      case 'errorMessage':
+        return this[kOnErrorMessage](message.error);
+    }
+
+    assert.fail(`Unknown worker message type ${message.type}`);
+  }
+
+  [kDispose]() {
+    this[kHandle].onexit = null;
+    this[kHandle] = null;
+    this[kPort] = null;
+    this[kPublicPort] = null;
+  }
+
+  postMessage(...args) {
+    this[kPublicPort].postMessage(...args);
+  }
+
+  terminate(callback) {
+    if (this[kHandle] === null) return;
+
+    debug(`[${threadId}] terminates Worker with ID ${this.threadId}`);
+
+    if (typeof callback !== 'undefined')
+      this.once('exit', (exitCode) => callback(null, exitCode));
+
+    this[kHandle].stopThread();
+  }
+
+  ref() {
+    if (this[kHandle] === null) return;
+
+    this[kHandle].ref();
+    this[kPublicPort].ref();
+  }
+
+  unref() {
+    if (this[kHandle] === null) return;
+
+    this[kHandle].unref();
+    this[kPublicPort].unref();
+  }
+
+  get threadId() {
+    if (this[kHandle] === null) return -1;
+
+    return this[kHandle].threadId;
+  }
+}
+
+let originalFatalException;
+
+function setupChild(evalScript) {
+  // Called during bootstrap to set up worker script execution.
+  debug(`[${threadId}] is setting up worker child environment`);
+  const port = getEnvMessagePort();
+
+  const publicWorker = require('worker');
+
+  port.on('message', (message) => {
+    if (message.type === 'loadScript') {
+      const { filename, doEval, workerData, publicPort } = message;
+      publicWorker.parentPort = publicPort;
+      setupPortReferencing(publicPort, publicPort, 'message');
+      publicWorker.workerData = workerData;
+      debug(`[${threadId}] starts worker script ${filename} ` +
+            `(eval = ${eval}) at cwd = ${process.cwd()}`);
+      port.unref();
+      port.postMessage({ type: 'upAndRunning' });
+      if (doEval) {
+        evalScript('[worker eval]', filename);
+      } else {
+        process.argv[1] = filename; // script filename
+        require('module').runMain();
+      }
+      return;
+    }
+
+    assert.fail(`Unknown worker message type ${message.type}`);
+  });
+
+  port.start();
+
+  originalFatalException = process._fatalException;
+  process._fatalException = fatalException;
+
+  function fatalException(error) {
+    debug(`[${threadId}] gets fatal exception`);
+    let caught = false;
+    try {
+      caught = originalFatalException.call(this, error);
+    } catch (e) {
+      error = e;
+    }
+    debug(`[${threadId}] fatal exception caught = ${caught}`);
+
+    if (!caught) {
+      let serialized;
+      try {
+        serialized = serializeError(error);
+      } catch {}
+      debug(`[${threadId}] fatal exception serialized = ${!!serialized}`);
+      if (serialized)
+        port.postMessage({ type: 'errorMessage', error: serialized });
+      else
+        port.postMessage({ type: 'couldNotSerializeError' });
+      clearAsyncIdStack();
+    }
+  }
+}
+
+// TODO(addaleax): These can be improved a lot.
+function serializeError(error) {
+  return Buffer.from(util.inspect(error), 'utf8');
+}
+
+function deserializeError(error) {
+  return Buffer.from(error.buffer,
+                     error.byteOffset,
+                     error.byteLength).toString('utf8');
+}
+
 module.exports = {
   MessagePort,
-  MessageChannel
+  MessageChannel,
+  threadId,
+  Worker,
+  setupChild,
+  isMainThread
 };
diff --git a/lib/worker.js b/lib/worker.js
index d67fb4efe4..0609650cd5 100644
--- a/lib/worker.js
+++ b/lib/worker.js
@@ -1,5 +1,18 @@
 'use strict';
 
-const { MessagePort, MessageChannel } = require('internal/worker');
+const {
+  isMainThread,
+  MessagePort,
+  MessageChannel,
+  threadId,
+  Worker
+} = require('internal/worker');
 
-module.exports = { MessagePort, MessageChannel };
+module.exports = {
+  isMainThread,
+  MessagePort,
+  MessageChannel,
+  threadId,
+  Worker,
+  parentPort: null
+};