diff --git a/benchmark/fixtures/echo.worker.js b/benchmark/fixtures/echo.worker.js
new file mode 100644
index 00000000000000..167a28ad23e830
--- /dev/null
+++ b/benchmark/fixtures/echo.worker.js
@@ -0,0 +1,7 @@
+'use strict';
+
+const { parentPort } = require('worker_threads');
+
+parentPort.on('message', (msg) => {
+ parentPort.postMessage(msg);
+});
diff --git a/benchmark/worker/echo.js b/benchmark/worker/echo.js
new file mode 100644
index 00000000000000..32c4bddf7d5f77
--- /dev/null
+++ b/benchmark/worker/echo.js
@@ -0,0 +1,73 @@
+'use strict';
+
+const common = require('../common.js');
+const path = require('path');
+const bench = common.createBenchmark(main, {
+ workers: [1],
+ payload: ['string', 'object'],
+ sendsPerBroadcast: [1, 10],
+ n: [1e5]
+}, { flags: ['--experimental-worker'] });
+
+const workerPath = path.resolve(__dirname, '..', 'fixtures', 'echo.worker.js');
+
+function main(conf) {
+ const { Worker } = require('worker_threads');
+
+ const n = +conf.n;
+ const workers = +conf.workers;
+ const sends = +conf.sendsPerBroadcast;
+ const expectedPerBroadcast = sends * workers;
+ var payload;
+ var readies = 0;
+ var broadcasts = 0;
+ var msgCount = 0;
+
+ switch (conf.payload) {
+ case 'string':
+ payload = 'hello world!';
+ break;
+ case 'object':
+ payload = { action: 'pewpewpew', powerLevel: 9001 };
+ break;
+ default:
+ throw new Error('Unsupported payload type');
+ }
+
+ const workerObjs = [];
+
+ for (var i = 0; i < workers; ++i) {
+ const worker = new Worker(workerPath);
+ workerObjs.push(worker);
+ worker.on('online', onOnline);
+ worker.on('message', onMessage);
+ }
+
+ function onOnline() {
+ if (++readies === workers) {
+ bench.start();
+ broadcast();
+ }
+ }
+
+ function broadcast() {
+ if (broadcasts++ === n) {
+ bench.end(n);
+ for (const worker of workerObjs) {
+ worker.unref();
+ }
+ return;
+ }
+ for (const worker of workerObjs) {
+ for (var i = 0; i < sends; ++i)
+ worker.postMessage(payload);
+ }
+ }
+
+ function onMessage() {
+ if (++msgCount === expectedPerBroadcast) {
+ msgCount = 0;
+ broadcast();
+ }
+ }
+}
diff --git a/doc/api/_toc.md b/doc/api/_toc.md
index 9b487b50a55031..e307d52ae8544e 100644
--- a/doc/api/_toc.md
+++ b/doc/api/_toc.md
@@ -53,6 +53,7 @@
* [Utilities](util.html)
* [V8](v8.html)
* [VM](vm.html)
+* [Worker Threads](worker_threads.html)
* [ZLIB](zlib.html)
diff --git a/doc/api/all.md b/doc/api/all.md
index d013f07bd328fc..47216b695d3351 100644
--- a/doc/api/all.md
+++ b/doc/api/all.md
@@ -46,4 +46,5 @@
@include util
@include v8
@include vm
+@include worker_threads
@include zlib
diff --git a/doc/api/async_hooks.md b/doc/api/async_hooks.md
index b97bc73304a4d7..f8449ec423740a 100644
--- a/doc/api/async_hooks.md
+++ b/doc/api/async_hooks.md
@@ -21,6 +21,9 @@ A resource can also be closed before the callback is called. `AsyncHook` does
not explicitly distinguish between these different cases but will represent them
as the abstract concept that is a resource.
+If [`Worker`][]s are used, each thread has an independent `async_hooks`
+interface, and each thread will use a new set of async IDs.
+
## Public API
### Overview
@@ -224,7 +227,7 @@ clearTimeout(setTimeout(() => {}, 10));
```
Every new resource is assigned an ID that is unique within the scope of the
-current process.
+current Node.js instance.
###### `type`
@@ -733,3 +736,4 @@ never be called.
[Hook Callbacks]: #async_hooks_hook_callbacks
[PromiseHooks]: https://docs.google.com/document/d/1rda3yKGHimKIhg5YeoAmCOtyURgsbTH_qaYR79FELlk
[promise execution tracking]: #async_hooks_promise_execution_tracking
+[`Worker`]: worker_threads.html#worker_threads_class_worker
diff --git a/doc/api/errors.md b/doc/api/errors.md
index fc1bcd3e6e994b..972d0971c4b059 100644
--- a/doc/api/errors.md
+++ b/doc/api/errors.md
@@ -629,6 +629,12 @@ An operation outside the bounds of a `Buffer` was attempted.
An attempt has been made to create a `Buffer` larger than the maximum allowed
size.
+
+### ERR_CANNOT_TRANSFER_OBJECT
+
+The value passed to `postMessage()` contained an object that is not supported
+for transferring.
+
### ERR_CANNOT_WATCH_SIGINT
@@ -650,12 +656,23 @@ Used when a child process is being forked without specifying an IPC channel.
Used when the main process is trying to read data from the child process's
STDERR / STDOUT, and the data's length is longer than the `maxBuffer` option.
+
+### ERR_CLOSED_MESSAGE_PORT
+
+There was an attempt to use a `MessagePort` instance in a closed
+state, usually after `.close()` has been called.
+
### ERR_CONSOLE_WRITABLE_STREAM
`Console` was instantiated without `stdout` stream, or `Console` has a
non-writable `stdout` or `stderr` stream.
+
+### ERR_CONSTRUCT_CALL_REQUIRED
+
+A constructor for a class was called without `new`.
+
### ERR_CPU_USAGE
@@ -1203,6 +1220,11 @@ urlSearchParams.has.call(buf, 'foo');
// Throws a TypeError with code 'ERR_INVALID_THIS'
```
+
+### ERR_INVALID_TRANSFER_OBJECT
+
+An invalid transfer object was passed to `postMessage()`.
+
### ERR_INVALID_TUPLE
@@ -1278,6 +1300,12 @@ strict compliance with the API specification (which in some cases may accept
`func(undefined)` and `func()` are treated identically, and the
[`ERR_INVALID_ARG_TYPE`][] error code may be used instead.
+
+### ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST
+
+A `MessagePort` was found in the object passed to a `postMessage()` call,
+but not provided in the `transferList` for that call.
+
### ERR_MISSING_MODULE
@@ -1285,6 +1313,13 @@ strict compliance with the API specification (which in some cases may accept
An [ES6 module][] could not be resolved.
+
+### ERR_MISSING_PLATFORM_FOR_WORKER
+
+The V8 platform used by this instance of Node.js does not support creating
+Workers. This is caused by lack of embedder support for Workers. In particular,
+this error will not occur with standard builds of Node.js.
+
### ERR_MODULE_RESOLUTION_LEGACY
@@ -1694,6 +1729,22 @@ The fulfilled value of a linking promise is not a `vm.Module` object.
The current module's status does not allow for this operation. The specific
meaning of the error depends on the specific function.
+
+### ERR_WORKER_NEED_ABSOLUTE_PATH
+
+The path for the main script of a worker is not an absolute path.
+
+
+### ERR_WORKER_UNSERIALIZABLE_ERROR
+
+All attempts at serializing an uncaught exception from a worker thread failed.
+
+
+### ERR_WORKER_UNSUPPORTED_EXTENSION
+
+The pathname used for the main script of a worker has an
+unknown file extension.
+
### ERR_ZLIB_INITIALIZATION_FAILED
diff --git a/doc/api/process.md b/doc/api/process.md
index 8205efbd780300..92b39d6da4a97d 100644
--- a/doc/api/process.md
+++ b/doc/api/process.md
@@ -410,6 +410,8 @@ added: v0.7.0
The `process.abort()` method causes the Node.js process to exit immediately and
generate a core file.
+This feature is not available in [`Worker`][] threads.
+
## process.arch
+
+> Stability: 1 - Experimental
+
+The `worker` module provides a way to create multiple environments running
+on independent threads, and to create message channels between them. It
+can be accessed using:
+
+```js
+const worker = require('worker_threads');
+```
+
+Workers are useful for performing CPU-intensive JavaScript operations; do not
+use them for I/O, since Node.js’s built-in mechanisms for performing operations
+asynchronously already treat it more efficiently than Worker threads can.
+
+Workers, unlike child processes or when using the `cluster` module, can also
+share memory efficiently by transferring `ArrayBuffer` instances or sharing
+`SharedArrayBuffer` instances between them.
+
+## Example
+
+```js
+const {
+ Worker, isMainThread, parentPort, workerData
+} = require('worker_threads');
+
+if (isMainThread) {
+ module.exports = async function parseJSAsync(script) {
+ return new Promise((resolve, reject) => {
+ const worker = new Worker(__filename, {
+ workerData: script
+ });
+ worker.on('message', resolve);
+ worker.on('error', reject);
+ worker.on('exit', (code) => {
+ if (code !== 0)
+ reject(new Error(`Worker stopped with exit code ${code}`));
+ });
+ });
+ };
+} else {
+ const { parse } = require('some-js-parsing-library');
+ const script = workerData;
+ parentPort.postMessage(parse(script));
+}
+```
+
+Note that this example spawns a Worker thread for each `parse` call.
+In practice, it is strongly recommended to use a pool of Workers for these
+kinds of tasks, since the overhead of creating Workers would likely exceed the
+benefit of handing the work off to it.
+
+## worker.isMainThread
+
+
+* {boolean}
+
+Is `true` if this code is not running inside of a [`Worker`][] thread.
+
+## worker.parentPort
+
+
+* {null|MessagePort}
+
+If this thread was spawned as a [`Worker`][], this will be a [`MessagePort`][]
+allowing communication with the parent thread. Messages sent using
+`parentPort.postMessage()` will be available in the parent thread
+using `worker.on('message')`, and messages sent from the parent thread
+using `worker.postMessage()` will be available in this thread using
+`parentPort.on('message')`.
+
+## worker.threadId
+
+
+* {integer}
+
+An integer identifier for the current thread. On the corresponding worker object
+(if there is any), it is available as [`worker.threadId`][].
+
+## worker.workerData
+
+
+An arbitrary JavaScript value that contains a clone of the data passed
+to this thread’s `Worker` constructor.
+
+## Class: MessageChannel
+
+
+Instances of the `worker.MessageChannel` class represent an asynchronous,
+two-way communications channel.
+The `MessageChannel` has no methods of its own. `new MessageChannel()`
+yields an object with `port1` and `port2` properties, which refer to linked
+[`MessagePort`][] instances.
+
+```js
+const { MessageChannel } = require('worker_threads');
+
+const { port1, port2 } = new MessageChannel();
+port1.on('message', (message) => console.log('received', message));
+port2.postMessage({ foo: 'bar' });
+// prints: received { foo: 'bar' } from the `port1.on('message')` listener
+```
+
+## Class: MessagePort
+
+
+* Extends: {EventEmitter}
+
+Instances of the `worker.MessagePort` class represent one end of an
+asynchronous, two-way communications channel. It can be used to transfer
+structured data, memory regions and other `MessagePort`s between different
+[`Worker`][]s.
+
+With the exception of `MessagePort`s being [`EventEmitter`][]s rather
+than `EventTarget`s, this implementation matches [browser `MessagePort`][]s.
+
+### Event: 'close'
+
+
+The `'close'` event is emitted once either side of the channel has been
+disconnected.
+
+### Event: 'message'
+
+
+* `value` {any} The transmitted value
+
+The `'message'` event is emitted for any incoming message, containing the cloned
+input of [`port.postMessage()`][].
+
+Listeners on this event will receive a clone of the `value` parameter as passed
+to `postMessage()` and no further arguments.
+
+### port.close()
+
+
+Disables further sending of messages on either side of the connection.
+This method can be called once you know that no further communication
+will happen over this `MessagePort`.
+
+### port.postMessage(value[, transferList])
+
+
+* `value` {any}
+* `transferList` {Object[]}
+
+Sends a JavaScript value to the receiving side of this channel.
+`value` will be transferred in a way which is compatible with
+the [HTML structured clone algorithm][]. In particular, it may contain circular
+references and objects like typed arrays that the `JSON` API is not able
+to stringify.
+
+`transferList` may be a list of `ArrayBuffer` and `MessagePort` objects.
+After transferring, they will not be usable on the sending side of the channel
+anymore (even if they are not contained in `value`). Unlike with
+[child processes][], transferring handles such as network sockets is currently
+not supported.
+
+If `value` contains [`SharedArrayBuffer`][] instances, those will be accessible
+from either thread. They cannot be listed in `transferList`.
+
+`value` may still contain `ArrayBuffer` instances that are not in
+`transferList`; in that case, the underlying memory is copied rather than moved.
+
+Because the object cloning uses the structured clone algorithm,
+non-enumerable properties, property accessors, and object prototypes are
+not preserved. In particular, [`Buffer`][] objects will be read as
+plain [`Uint8Array`][]s on the receiving side.
+
+The message object will be cloned immediately, and can be modified after
+posting without having side effects.
+
+For more information on the serialization and deserialization mechanisms
+behind this API, see the [serialization API of the `v8` module][v8.serdes].
+
+### port.ref()
+
+
+Opposite of `unref()`. Calling `ref()` on a previously `unref()`ed port will
+*not* let the program exit if it's the only active handle left (the default
+behavior). If the port is `ref()`ed, calling `ref()` again will have no effect.
+
+If listeners are attached or removed using `.on('message')`, the port will
+be `ref()`ed and `unref()`ed automatically depending on whether
+listeners for the event exist.
+
+### port.start()
+
+
+Starts receiving messages on this `MessagePort`. When using this port
+as an event emitter, this will be called automatically once `'message'`
+listeners are attached.
+
+### port.unref()
+
+
+Calling `unref()` on a port will allow the thread to exit if this is the only
+active handle in the event system. If the port is already `unref()`ed calling
+`unref()` again will have no effect.
+
+If listeners are attached or removed using `.on('message')`, the port will
+be `ref()`ed and `unref()`ed automatically depending on whether
+listeners for the event exist.
+
+## Class: Worker
+
+
+The `Worker` class represents an independent JavaScript execution thread.
+Most Node.js APIs are available inside of it.
+
+Notable differences inside a Worker environment are:
+
+- The [`process.stdin`][], [`process.stdout`][] and [`process.stderr`][]
+ may be redirected by the parent thread.
+- The [`require('worker_threads').isMainThread`][] property is set to `false`.
+- The [`require('worker_threads').parentPort`][] message port is available,
+- [`process.exit()`][] does not stop the whole program, just the single thread,
+ and [`process.abort()`][] is not available.
+- [`process.chdir()`][] and `process` methods that set group or user ids
+ are not available.
+- [`process.env`][] is a read-only reference to the environment variables.
+- [`process.title`][] cannot be modified.
+- Signals will not be delivered through [`process.on('...')`][Signals events].
+- Execution may stop at any point as a result of [`worker.terminate()`][]
+ being invoked.
+- IPC channels from parent processes are not accessible.
+
+Currently, the following differences also exist until they are addressed:
+
+- The [`inspector`][] module is not available yet.
+- Native addons are not supported yet.
+
+Creating `Worker` instances inside of other `Worker`s is possible.
+
+Like [Web Workers][] and the [`cluster` module][], two-way communication can be
+achieved through inter-thread message passing. Internally, a `Worker` has a
+built-in pair of [`MessagePort`][]s that are already associated with each other
+when the `Worker` is created. While the `MessagePort` object on the parent side
+is not directly exposed, its functionalities are exposed through
+[`worker.postMessage()`][] and the [`worker.on('message')`][] event
+on the `Worker` object for the parent thread.
+
+To create custom messaging channels (which is encouraged over using the default
+global channel because it facilitates separation of concerns), users can create
+a `MessageChannel` object on either thread and pass one of the
+`MessagePort`s on that `MessageChannel` to the other thread through a
+pre-existing channel, such as the global one.
+
+See [`port.postMessage()`][] for more information on how messages are passed,
+and what kind of JavaScript values can be successfully transported through
+the thread barrier.
+
+For example:
+
+```js
+const assert = require('assert');
+const {
+ Worker, MessageChannel, MessagePort, isMainThread
+} = require('worker_threads');
+if (isMainThread) {
+ const worker = new Worker(__filename);
+ const subChannel = new MessageChannel();
+ worker.postMessage({ hereIsYourPort: subChannel.port1 }, [subChannel.port1]);
+ subChannel.port2.on('message', (value) => {
+ console.log('received:', value);
+ });
+} else {
+ require('worker_threads').once('workerMessage', (value) => {
+ assert(value.hereIsYourPort instanceof MessagePort);
+ value.hereIsYourPort.postMessage('the worker is sending this');
+ value.hereIsYourPort.close();
+ });
+}
+```
+
+### new Worker(filename, options)
+
+* `filename` {string} The absolute path to the Worker’s main script.
+ If `options.eval` is true, this is a string containing JavaScript code rather
+ than a path.
+* `options` {Object}
+ * `eval` {boolean} If true, interpret the first argument to the constructor
+ as a script that is executed once the worker is online.
+ * `data` {any} Any JavaScript value that will be cloned and made
+ available as [`require('worker_threads').workerData`][]. The cloning will
+ occur as described in the [HTML structured clone algorithm][], and an error
+ will be thrown if the object cannot be cloned (e.g. because it contains
+ `function`s).
+ * stdin {boolean} If this is set to `true`, then `worker.stdin` will
+ provide a writable stream whose contents will appear as `process.stdin`
+ inside the Worker. By default, no data is provided.
+ * stdout {boolean} If this is set to `true`, then `worker.stdout` will
+ not automatically be piped through to `process.stdout` in the parent.
+ * stderr {boolean} If this is set to `true`, then `worker.stderr` will
+ not automatically be piped through to `process.stderr` in the parent.
+
+### Event: 'error'
+
+
+* `err` {Error}
+
+The `'error'` event is emitted if the worker thread throws an uncaught
+exception. In that case, the worker will be terminated.
+
+### Event: 'exit'
+
+
+* `exitCode` {integer}
+
+The `'exit'` event is emitted once the worker has stopped. If the worker
+exited by calling [`process.exit()`][], the `exitCode` parameter will be the
+passed exit code. If the worker was terminated, the `exitCode` parameter will
+be `1`.
+
+### Event: 'message'
+
+
+* `value` {any} The transmitted value
+
+The `'message'` event is emitted when the worker thread has invoked
+[`require('worker_threads').postMessage()`][]. See the [`port.on('message')`][]
+event for more details.
+
+### Event: 'online'
+
+
+The `'online'` event is emitted when the worker thread has started executing
+JavaScript code.
+
+### worker.postMessage(value[, transferList])
+
+
+* `value` {any}
+* `transferList` {Object[]}
+
+Send a message to the worker that will be received via
+[`require('worker_threads').on('workerMessage')`][].
+See [`port.postMessage()`][] for more details.
+
+### worker.ref()
+
+
+Opposite of `unref()`, calling `ref()` on a previously `unref()`ed worker will
+*not* let the program exit if it's the only active handle left (the default
+behavior). If the worker is `ref()`ed, calling `ref()` again will have
+no effect.
+
+### worker.stderr
+
+
+* {stream.Readable}
+
+This is a readable stream which contains data written to [`process.stderr`][]
+inside the worker thread. If `stderr: true` was not passed to the
+[`Worker`][] constructor, then data will be piped to the parent thread's
+[`process.stderr`][] stream.
+
+### worker.stdin
+
+
+* {null|stream.Writable}
+
+If `stdin: true` was passed to the [`Worker`][] constructor, this is a
+writable stream. The data written to this stream will be made available in
+the worker thread as [`process.stdin`][].
+
+### worker.stdout
+
+
+* {stream.Readable}
+
+This is a readable stream which contains data written to [`process.stdout`][]
+inside the worker thread. If `stdout: true` was not passed to the
+[`Worker`][] constructor, then data will be piped to the parent thread's
+[`process.stdout`][] stream.
+
+### worker.terminate([callback])
+
+
+* `callback` {Function}
+
+Stop all JavaScript execution in the worker thread as soon as possible.
+`callback` is an optional function that is invoked once this operation is known
+to have completed.
+
+**Warning**: Currently, not all code in the internals of Node.js is prepared to
+expect termination at arbitrary points in time and may crash if it encounters
+that condition. Consequently, you should currently only call `.terminate()` if
+it is known that the Worker thread is not accessing Node.js core modules other
+than what is exposed in the `worker` module.
+
+### worker.threadId
+
+
+* {integer}
+
+An integer identifier for the referenced thread. Inside the worker thread,
+it is available as [`require('worker_threads').threadId`][].
+
+### worker.unref()
+
+
+Calling `unref()` on a worker will allow 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 will have no effect.
+
+[`Buffer`]: buffer.html
+[`EventEmitter`]: events.html
+[`MessagePort`]: #worker_threads_class_messageport
+[`port.postMessage()`]: #worker_threads_port_postmessage_value_transferlist
+[`Worker`]: #worker_threads_class_worker
+[`worker.terminate()`]: #worker_threads_worker_terminate_callback
+[`worker.postMessage()`]: #worker_threads_worker_postmessage_value_transferlist_1
+[`worker.on('message')`]: #worker_threads_event_message_1
+[`worker.threadId`]: #worker_threads_worker_threadid_1
+[`port.on('message')`]: #worker_threads_event_message
+[`process.exit()`]: process.html#process_process_exit_code
+[`process.abort()`]: process.html#process_process_abort
+[`process.chdir()`]: process.html#process_process_chdir_directory
+[`process.env`]: process.html#process_process_env
+[`process.stdin`]: process.html#process_process_stdin
+[`process.stderr`]: process.html#process_process_stderr
+[`process.stdout`]: process.html#process_process_stdout
+[`process.title`]: process.html#process_process_title
+[`require('worker_threads').workerData`]: #worker_threads_worker_workerdata
+[`require('worker_threads').on('workerMessage')`]: #worker_threads_event_workermessage
+[`require('worker_threads').postMessage()`]: #worker_threads_worker_postmessage_value_transferlist
+[`require('worker_threads').isMainThread`]: #worker_threads_worker_ismainthread
+[`require('worker_threads').threadId`]: #worker_threads_worker_threadid
+[`cluster` module]: cluster.html
+[`inspector`]: inspector.html
+[v8.serdes]: v8.html#v8_serialization_api
+[`SharedArrayBuffer`]: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/SharedArrayBuffer
+[Signals events]: process.html#process_signal_events
+[`Uint8Array`]: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Uint8Array
+[browser `MessagePort`]: https://developer.mozilla.org/en-US/docs/Web/API/MessagePort
+[child processes]: child_process.html
+[HTML structured clone algorithm]: https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm
+[Web Workers]: https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API
diff --git a/lib/inspector.js b/lib/inspector.js
index 3285c1040a7132..f4ec71fd6c2105 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/loaders.js b/lib/internal/bootstrap/loaders.js
index ff809a91291bee..4291092532ec94 100644
--- a/lib/internal/bootstrap/loaders.js
+++ b/lib/internal/bootstrap/loaders.js
@@ -194,7 +194,9 @@
};
NativeModule.isInternal = function(id) {
- return id.startsWith('internal/');
+ return id.startsWith('internal/') ||
+ (id === 'worker_threads' &&
+ !process.binding('config').experimentalWorker);
};
}
diff --git a/lib/internal/bootstrap/node.js b/lib/internal/bootstrap/node.js
index 6477c2d8282f43..4817ec110a99e5 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/error-serdes.js b/lib/internal/error-serdes.js
new file mode 100644
index 00000000000000..9da1a864171607
--- /dev/null
+++ b/lib/internal/error-serdes.js
@@ -0,0 +1,121 @@
+'use strict';
+
+const Buffer = require('buffer').Buffer;
+const { serialize, deserialize } = require('v8');
+const { SafeSet } = require('internal/safe_globals');
+
+const kSerializedError = 0;
+const kSerializedObject = 1;
+const kInspectedError = 2;
+
+const GetPrototypeOf = Object.getPrototypeOf;
+const GetOwnPropertyDescriptor = Object.getOwnPropertyDescriptor;
+const GetOwnPropertyNames = Object.getOwnPropertyNames;
+const DefineProperty = Object.defineProperty;
+const Assign = Object.assign;
+const ObjectPrototypeToString =
+ Function.prototype.call.bind(Object.prototype.toString);
+const ForEach = Function.prototype.call.bind(Array.prototype.forEach);
+const Call = Function.prototype.call.bind(Function.prototype.call);
+
+const errors = {
+ Error, TypeError, RangeError, URIError, SyntaxError, ReferenceError, EvalError
+};
+const errorConstructorNames = new SafeSet(Object.keys(errors));
+
+function TryGetAllProperties(object, target = object) {
+ const all = Object.create(null);
+ if (object === null)
+ return all;
+ Assign(all, TryGetAllProperties(GetPrototypeOf(object), target));
+ const keys = GetOwnPropertyNames(object);
+ ForEach(keys, (key) => {
+ const descriptor = GetOwnPropertyDescriptor(object, key);
+ const getter = descriptor.get;
+ if (getter && key !== '__proto__') {
+ try {
+ descriptor.value = Call(getter, target);
+ } catch {}
+ }
+ if ('value' in descriptor && typeof descriptor.value !== 'function') {
+ delete descriptor.get;
+ delete descriptor.set;
+ all[key] = descriptor;
+ }
+ });
+ return all;
+}
+
+function GetConstructors(object) {
+ const constructors = [];
+
+ for (var current = object;
+ current !== null;
+ current = GetPrototypeOf(current)) {
+ const desc = GetOwnPropertyDescriptor(current, 'constructor');
+ if (desc && desc.value) {
+ DefineProperty(constructors, constructors.length, {
+ value: desc.value, enumerable: true
+ });
+ }
+ }
+
+ return constructors;
+}
+
+function GetName(object) {
+ const desc = GetOwnPropertyDescriptor(object, 'name');
+ return desc && desc.value;
+}
+
+let util;
+function lazyUtil() {
+ if (!util)
+ util = require('util');
+ return util;
+}
+
+function serializeError(error) {
+ try {
+ if (typeof error === 'object' &&
+ ObjectPrototypeToString(error) === '[object Error]') {
+ const constructors = GetConstructors(error);
+ for (var i = constructors.length - 1; i >= 0; i--) {
+ const name = GetName(constructors[i]);
+ if (errorConstructorNames.has(name)) {
+ try { error.stack; } catch {}
+ const serialized = serialize({
+ constructor: name,
+ properties: TryGetAllProperties(error)
+ });
+ return Buffer.concat([Buffer.from([kSerializedError]), serialized]);
+ }
+ }
+ }
+ } catch {}
+ try {
+ const serialized = serialize(error);
+ return Buffer.concat([Buffer.from([kSerializedObject]), serialized]);
+ } catch {}
+ return Buffer.concat([Buffer.from([kInspectedError]),
+ Buffer.from(lazyUtil().inspect(error), 'utf8')]);
+}
+
+function deserializeError(error) {
+ switch (error[0]) {
+ case kSerializedError:
+ const { constructor, properties } = deserialize(error.subarray(1));
+ const ctor = errors[constructor];
+ return Object.create(ctor.prototype, properties);
+ case kSerializedObject:
+ return deserialize(error.subarray(1));
+ case kInspectedError:
+ const buf = Buffer.from(error.buffer,
+ error.byteOffset + 1,
+ error.byteLength - 1);
+ return buf.toString('utf8');
+ }
+ require('assert').fail('This should not happen');
+}
+
+module.exports = { serializeError, deserializeError };
diff --git a/lib/internal/errors.js b/lib/internal/errors.js
index 89c0139f8b6fde..54201d0d1e7f4c 100644
--- a/lib/internal/errors.js
+++ b/lib/internal/errors.js
@@ -844,4 +844,12 @@ 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_WORKER_UNSUPPORTED_EXTENSION',
+ 'The worker script extension must be ".js" or ".mjs". Received "%s"',
+ TypeError);
E('ERR_ZLIB_INITIALIZATION_FAILED', 'Initialization failed', Error);
diff --git a/lib/internal/modules/cjs/helpers.js b/lib/internal/modules/cjs/helpers.js
index 60346c5841c7df..5b5199c262ae3b 100644
--- a/lib/internal/modules/cjs/helpers.js
+++ b/lib/internal/modules/cjs/helpers.js
@@ -105,6 +105,11 @@ const builtinLibs = [
'v8', 'vm', 'zlib'
];
+if (process.binding('config').experimentalWorker) {
+ builtinLibs.push('worker_threads');
+ builtinLibs.sort();
+}
+
if (typeof process.binding('inspector').open === 'function') {
builtinLibs.push('inspector');
builtinLibs.sort();
diff --git a/lib/internal/process.js b/lib/internal/process.js
index 0f0e40d6a0cdbc..f01be32be4b6a3 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 91aca398b346d4..9a954f6a9b93cf 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 eaba4dfca13a47..e5bb50e8ed81de 100644
--- a/lib/internal/process/stdio.js
+++ b/lib/internal/process/stdio.js
@@ -6,6 +6,10 @@ const {
ERR_UNKNOWN_STDIN_TYPE,
ERR_UNKNOWN_STREAM_TYPE
} = require('internal/errors').codes;
+const {
+ isMainThread,
+ workerStdio
+} = require('internal/worker');
exports.setup = setupStdio;
@@ -16,6 +20,7 @@ function setupStdio() {
function getStdout() {
if (stdout) return stdout;
+ if (!isMainThread) return workerStdio.stdout;
stdout = createWritableStdioStream(1);
stdout.destroySoon = stdout.destroy;
stdout._destroy = function(er, cb) {
@@ -31,6 +36,7 @@ function setupStdio() {
function getStderr() {
if (stderr) return stderr;
+ if (!isMainThread) return workerStdio.stderr;
stderr = createWritableStdioStream(2);
stderr.destroySoon = stderr.destroy;
stderr._destroy = function(er, cb) {
@@ -46,6 +52,7 @@ function setupStdio() {
function getStdin() {
if (stdin) return stdin;
+ if (!isMainThread) return workerStdio.stdin;
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 634d3302333584..3dd73415ded862 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
new file mode 100644
index 00000000000000..de00f20d4f0e1d
--- /dev/null
+++ b/lib/internal/worker.js
@@ -0,0 +1,476 @@
+'use strict';
+
+const EventEmitter = require('events');
+const assert = require('assert');
+const path = require('path');
+const util = require('util');
+const { Readable, Writable } = require('stream');
+const {
+ ERR_INVALID_ARG_TYPE,
+ ERR_WORKER_NEED_ABSOLUTE_PATH,
+ ERR_WORKER_UNSERIALIZABLE_ERROR,
+ ERR_WORKER_UNSUPPORTED_EXTENSION,
+} = 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');
+const { serializeError, deserializeError } = require('internal/error-serdes');
+
+util.inherits(MessagePort, EventEmitter);
+
+const {
+ Worker: WorkerImpl,
+ getEnvMessagePort,
+ threadId
+} = internalBinding('worker');
+
+const isMainThread = threadId === 0;
+
+const kOnMessageListener = Symbol('kOnMessageListener');
+const kHandle = Symbol('kHandle');
+const kName = Symbol('kName');
+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 kParentSideStdio = Symbol('kParentSideStdio');
+const kWritableCallbacks = Symbol('kWritableCallbacks');
+const kStdioWantsMoreDataCallback = Symbol('kStdioWantsMoreDataCallback');
+const kStartedReading = Symbol('kStartedReading');
+const kWaitingStreams = Symbol('kWaitingStreams');
+const kIncrementsPortRef = Symbol('kIncrementsPortRef');
+
+const debug = util.debuglog('worker');
+
+// 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(`[${threadId}] received message`, payload);
+ // Emit the deserialized object to userland.
+ this.emit('message', payload);
+};
+
+// This is for compatibility with the Web's MessagePort API. It makes sense to
+// provide it as an `EventEmitter` in Node.js, but if somebody overrides
+// `onmessage`, we'll switch over to the Web API model.
+Object.defineProperty(MessagePort.prototype, 'onmessage', {
+ enumerable: true,
+ configurable: true,
+ get() {
+ return this[kOnMessageListener];
+ },
+ set(value) {
+ this[kOnMessageListener] = value;
+ if (typeof value === 'function') {
+ this.ref();
+ this.start();
+ } else {
+ this.unref();
+ this.stop();
+ }
+ }
+});
+
+// This is called from inside the `MessagePort` constructor.
+function oninit() {
+ setupPortReferencing(this, this, 'message');
+}
+
+Object.defineProperty(MessagePort.prototype, 'oninit', {
+ enumerable: true,
+ writable: false,
+ value: oninit
+});
+
+// This is called after the underlying `uv_async_t` has been closed.
+function onclose() {
+ if (typeof this.onclose === 'function') {
+ // Not part of the Web standard yet, but there aren't many reasonable
+ // alternatives in a non-EventEmitter usage setting.
+ // Refs: https://github.com/whatwg/html/issues/1766
+ this.onclose();
+ }
+ this.emit('close');
+}
+
+Object.defineProperty(MessagePort.prototype, handle_onclose, {
+ enumerable: false,
+ writable: false,
+ value: onclose
+});
+
+const originalClose = MessagePort.prototype.close;
+MessagePort.prototype.close = function(cb) {
+ if (typeof cb === 'function')
+ this.once('close', 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.
+ // If there are none or all are removed, unref() the channel so the worker
+ // can shutdown gracefully.
+ port.unref();
+ eventEmitter.on('newListener', (name) => {
+ if (name === eventName && eventEmitter.listenerCount(eventName) === 0) {
+ port.ref();
+ port.start();
+ }
+ });
+ eventEmitter.on('removeListener', (name) => {
+ if (name === eventName && eventEmitter.listenerCount(eventName) === 0) {
+ port.stop();
+ port.unref();
+ }
+ });
+}
+
+
+class ReadableWorkerStdio extends Readable {
+ constructor(port, name) {
+ super();
+ this[kPort] = port;
+ this[kName] = name;
+ this[kIncrementsPortRef] = true;
+ this[kStartedReading] = false;
+ this.on('end', () => {
+ if (this[kIncrementsPortRef] && --this[kPort][kWaitingStreams] === 0)
+ this[kPort].unref();
+ });
+ }
+
+ _read() {
+ if (!this[kStartedReading] && this[kIncrementsPortRef]) {
+ this[kStartedReading] = true;
+ if (this[kPort][kWaitingStreams]++ === 0)
+ this[kPort].ref();
+ }
+
+ this[kPort].postMessage({
+ type: 'stdioWantsMoreData',
+ stream: this[kName]
+ });
+ }
+}
+
+class WritableWorkerStdio extends Writable {
+ constructor(port, name) {
+ super({ decodeStrings: false });
+ this[kPort] = port;
+ this[kName] = name;
+ this[kWritableCallbacks] = [];
+ }
+
+ _write(chunk, encoding, cb) {
+ this[kPort].postMessage({
+ type: 'stdioPayload',
+ stream: this[kName],
+ chunk,
+ encoding
+ });
+ this[kWritableCallbacks].push(cb);
+ if (this[kPort][kWaitingStreams]++ === 0)
+ this[kPort].ref();
+ }
+
+ _final(cb) {
+ this[kPort].postMessage({
+ type: 'stdioPayload',
+ stream: this[kName],
+ chunk: null
+ });
+ cb();
+ }
+
+ [kStdioWantsMoreDataCallback]() {
+ const cbs = this[kWritableCallbacks];
+ this[kWritableCallbacks] = [];
+ for (const cb of cbs)
+ cb();
+ if ((this[kPort][kWaitingStreams] -= cbs.length) === 0)
+ this[kPort].unref();
+ }
+}
+
+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) {
+ if (!path.isAbsolute(filename)) {
+ throw new ERR_WORKER_NEED_ABSOLUTE_PATH(filename);
+ }
+ const ext = path.extname(filename);
+ if (ext !== '.js' && ext !== '.mjs') {
+ throw new ERR_WORKER_UNSUPPORTED_EXTENSION(ext);
+ }
+ }
+
+ // 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();
+ this[kPort][kWaitingStreams] = 0;
+ debug(`[${threadId}] created Worker with ID ${this.threadId}`);
+
+ let stdin = null;
+ if (options.stdin)
+ stdin = new WritableWorkerStdio(this[kPort], 'stdin');
+ const stdout = new ReadableWorkerStdio(this[kPort], 'stdout');
+ if (!options.stdout) {
+ stdout[kIncrementsPortRef] = false;
+ pipeWithoutWarning(stdout, process.stdout);
+ }
+ const stderr = new ReadableWorkerStdio(this[kPort], 'stderr');
+ if (!options.stderr) {
+ stderr[kIncrementsPortRef] = false;
+ pipeWithoutWarning(stderr, process.stderr);
+ }
+
+ this[kParentSideStdio] = { stdin, stdout, stderr };
+
+ 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,
+ hasStdin: !!options.stdin
+ }, [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);
+ case 'stdioPayload':
+ {
+ const { stream, chunk, encoding } = message;
+ return this[kParentSideStdio][stream].push(chunk, encoding);
+ }
+ case 'stdioWantsMoreData':
+ {
+ const { stream } = message;
+ return this[kParentSideStdio][stream][kStdioWantsMoreDataCallback]();
+ }
+ }
+
+ assert.fail(`Unknown worker message type ${message.type}`);
+ }
+
+ [kDispose]() {
+ this[kHandle].onexit = null;
+ this[kHandle] = null;
+ this[kPort] = null;
+ this[kPublicPort] = null;
+
+ const { stdout, stderr } = this[kParentSideStdio];
+ this[kParentSideStdio] = null;
+
+ if (!stdout._readableState.ended) {
+ debug(`[${threadId}] explicitly closes stdout for ${this.threadId}`);
+ stdout.push(null);
+ }
+ if (!stderr._readableState.ended) {
+ debug(`[${threadId}] explicitly closes stderr for ${this.threadId}`);
+ stderr.push(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;
+ }
+
+ get stdin() {
+ return this[kParentSideStdio].stdin;
+ }
+
+ get stdout() {
+ return this[kParentSideStdio].stdout;
+ }
+
+ get stderr() {
+ return this[kParentSideStdio].stderr;
+ }
+}
+
+const workerStdio = {};
+if (!isMainThread) {
+ const port = getEnvMessagePort();
+ port[kWaitingStreams] = 0;
+ workerStdio.stdin = new ReadableWorkerStdio(port, 'stdin');
+ workerStdio.stdout = new WritableWorkerStdio(port, 'stdout');
+ workerStdio.stderr = new WritableWorkerStdio(port, 'stderr');
+}
+
+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_threads');
+
+ port.on('message', (message) => {
+ if (message.type === 'loadScript') {
+ const { filename, doEval, workerData, publicPort, hasStdin } = message;
+ publicWorker.parentPort = publicPort;
+ setupPortReferencing(publicPort, publicPort, 'message');
+ publicWorker.workerData = workerData;
+
+ if (!hasStdin)
+ workerStdio.stdin.push(null);
+
+ 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;
+ } else if (message.type === 'stdioPayload') {
+ const { stream, chunk, encoding } = message;
+ workerStdio[stream].push(chunk, encoding);
+ return;
+ } else if (message.type === 'stdioWantsMoreData') {
+ const { stream } = message;
+ workerStdio[stream][kStdioWantsMoreDataCallback]();
+ 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();
+ }
+ }
+}
+
+function pipeWithoutWarning(source, dest) {
+ const sourceMaxListeners = source._maxListeners;
+ const destMaxListeners = dest._maxListeners;
+ source.setMaxListeners(Infinity);
+ dest.setMaxListeners(Infinity);
+
+ source.pipe(dest);
+
+ source._maxListeners = sourceMaxListeners;
+ dest._maxListeners = destMaxListeners;
+}
+
+module.exports = {
+ MessagePort,
+ MessageChannel,
+ threadId,
+ Worker,
+ setupChild,
+ isMainThread,
+ workerStdio
+};
diff --git a/lib/worker_threads.js b/lib/worker_threads.js
new file mode 100644
index 00000000000000..0609650cd5731d
--- /dev/null
+++ b/lib/worker_threads.js
@@ -0,0 +1,18 @@
+'use strict';
+
+const {
+ isMainThread,
+ MessagePort,
+ MessageChannel,
+ threadId,
+ Worker
+} = require('internal/worker');
+
+module.exports = {
+ isMainThread,
+ MessagePort,
+ MessageChannel,
+ threadId,
+ Worker,
+ parentPort: null
+};
diff --git a/node.gyp b/node.gyp
index 4b94c1dd6b2ad9..14bf5211e97362 100644
--- a/node.gyp
+++ b/node.gyp
@@ -78,6 +78,7 @@
'lib/util.js',
'lib/v8.js',
'lib/vm.js',
+ 'lib/worker_threads.js',
'lib/zlib.js',
'lib/internal/assert.js',
'lib/internal/async_hooks.js',
@@ -101,6 +102,7 @@
'lib/internal/constants.js',
'lib/internal/encoding.js',
'lib/internal/errors.js',
+ 'lib/internal/error-serdes.js',
'lib/internal/fixed_queue.js',
'lib/internal/freelist.js',
'lib/internal/fs/promises.js',
@@ -156,6 +158,7 @@
'lib/internal/validators.js',
'lib/internal/stream_base_commons.js',
'lib/internal/vm/module.js',
+ 'lib/internal/worker.js',
'lib/internal/streams/lazy_transform.js',
'lib/internal/streams/async_iterator.js',
'lib/internal/streams/buffer_list.js',
@@ -334,6 +337,7 @@
'src/node_file.cc',
'src/node_http2.cc',
'src/node_http_parser.cc',
+ 'src/node_messaging.cc',
'src/node_os.cc',
'src/node_platform.cc',
'src/node_perf.cc',
@@ -346,10 +350,12 @@
'src/node_v8.cc',
'src/node_stat_watcher.cc',
'src/node_watchdog.cc',
+ 'src/node_worker.cc',
'src/node_zlib.cc',
'src/node_i18n.cc',
'src/pipe_wrap.cc',
'src/process_wrap.cc',
+ 'src/sharedarraybuffer_metadata.cc',
'src/signal_wrap.cc',
'src/spawn_sync.cc',
'src/string_bytes.cc',
@@ -391,6 +397,7 @@
'src/node_http2_state.h',
'src/node_internals.h',
'src/node_javascript.h',
+ 'src/node_messaging.h',
'src/node_mutex.h',
'src/node_perf.h',
'src/node_perf_common.h',
@@ -402,12 +409,14 @@
'src/node_wrap.h',
'src/node_revert.h',
'src/node_i18n.h',
+ 'src/node_worker.h',
'src/pipe_wrap.h',
'src/tty_wrap.h',
'src/tcp_wrap.h',
'src/udp_wrap.h',
'src/req_wrap.h',
'src/req_wrap-inl.h',
+ 'src/sharedarraybuffer_metadata.h',
'src/string_bytes.h',
'src/string_decoder.h',
'src/string_decoder-inl.h',
diff --git a/src/async_wrap-inl.h b/src/async_wrap-inl.h
index c9f12333243092..5763b17aa08bc4 100644
--- a/src/async_wrap-inl.h
+++ b/src/async_wrap-inl.h
@@ -65,6 +65,22 @@ inline v8::MaybeLocal AsyncWrap::MakeCallback(
const v8::Local symbol,
int argc,
v8::Local* argv) {
+ return MakeCallback(symbol.As(), argc, argv);
+}
+
+
+inline v8::MaybeLocal AsyncWrap::MakeCallback(
+ const v8::Local symbol,
+ int argc,
+ v8::Local* argv) {
+ return MakeCallback(symbol.As(), argc, argv);
+}
+
+
+inline v8::MaybeLocal AsyncWrap::MakeCallback(
+ const v8::Local symbol,
+ int argc,
+ v8::Local* argv) {
v8::Local cb_v = object()->Get(symbol);
CHECK(cb_v->IsFunction());
return MakeCallback(cb_v.As(), argc, argv);
diff --git a/src/async_wrap.h b/src/async_wrap.h
index 451bcfe12e6717..b2f96477b490e0 100644
--- a/src/async_wrap.h
+++ b/src/async_wrap.h
@@ -49,6 +49,7 @@ namespace node {
V(HTTP2SETTINGS) \
V(HTTPPARSER) \
V(JSSTREAM) \
+ V(MESSAGEPORT) \
V(PIPECONNECTWRAP) \
V(PIPESERVERWRAP) \
V(PIPEWRAP) \
@@ -66,6 +67,7 @@ namespace node {
V(TTYWRAP) \
V(UDPSENDWRAP) \
V(UDPWRAP) \
+ V(WORKER) \
V(WRITEWRAP) \
V(ZLIB)
@@ -158,10 +160,18 @@ class AsyncWrap : public BaseObject {
v8::MaybeLocal MakeCallback(const v8::Local cb,
int argc,
v8::Local* argv);
+ inline v8::MaybeLocal MakeCallback(
+ const v8::Local symbol,
+ int argc,
+ v8::Local* argv);
inline v8::MaybeLocal MakeCallback(
const v8::Local symbol,
int argc,
v8::Local* argv);
+ inline v8::MaybeLocal MakeCallback(
+ const v8::Local symbol,
+ int argc,
+ v8::Local* argv);
inline v8::MaybeLocal MakeCallback(uint32_t index,
int argc,
v8::Local* argv);
diff --git a/src/base_object-inl.h b/src/base_object-inl.h
index 3bd854639b2c6d..06a29223973c5d 100644
--- a/src/base_object-inl.h
+++ b/src/base_object-inl.h
@@ -65,6 +65,14 @@ v8::Local BaseObject::object() {
return PersistentToLocal(env_->isolate(), persistent_handle_);
}
+v8::Local BaseObject::object(v8::Isolate* isolate) {
+ v8::Local handle = object();
+#ifdef DEBUG
+ CHECK_EQ(handle->CreationContext()->GetIsolate(), isolate);
+ CHECK_EQ(env_->isolate(), isolate);
+#endif
+ return handle;
+}
Environment* BaseObject::env() const {
return env_;
diff --git a/src/base_object.h b/src/base_object.h
index e0b60843401681..38291d598feb1c 100644
--- a/src/base_object.h
+++ b/src/base_object.h
@@ -43,6 +43,10 @@ class BaseObject {
// persistent.IsEmpty() is true.
inline v8::Local object();
+ // Same as the above, except it additionally verifies that this object
+ // is associated with the passed Isolate in debug mode.
+ inline v8::Local object(v8::Isolate* isolate);
+
inline Persistent& persistent();
inline Environment* env() const;
diff --git a/src/bootstrapper.cc b/src/bootstrapper.cc
index 6c7c1af3e31cf6..f9db02562d9c8a 100644
--- a/src/bootstrapper.cc
+++ b/src/bootstrapper.cc
@@ -17,6 +17,7 @@ using v8::Object;
using v8::Promise;
using v8::PromiseRejectEvent;
using v8::PromiseRejectMessage;
+using v8::String;
using v8::Value;
void SetupProcessObject(const FunctionCallbackInfo& args) {
@@ -113,15 +114,17 @@ void SetupBootstrapObject(Environment* env,
BOOTSTRAP_METHOD(_umask, Umask);
#if defined(__POSIX__) && !defined(__ANDROID__) && !defined(__CloudABI__)
- BOOTSTRAP_METHOD(_initgroups, InitGroups);
- BOOTSTRAP_METHOD(_setegid, SetEGid);
- BOOTSTRAP_METHOD(_seteuid, SetEUid);
- BOOTSTRAP_METHOD(_setgid, SetGid);
- BOOTSTRAP_METHOD(_setuid, SetUid);
- BOOTSTRAP_METHOD(_setgroups, SetGroups);
+ if (env->is_main_thread()) {
+ BOOTSTRAP_METHOD(_initgroups, InitGroups);
+ BOOTSTRAP_METHOD(_setegid, SetEGid);
+ BOOTSTRAP_METHOD(_seteuid, SetEUid);
+ BOOTSTRAP_METHOD(_setgid, SetGid);
+ BOOTSTRAP_METHOD(_setuid, SetUid);
+ BOOTSTRAP_METHOD(_setgroups, SetGroups);
+ }
#endif // __POSIX__ && !defined(__ANDROID__) && !defined(__CloudABI__)
- auto should_abort_on_uncaught_toggle =
+ Local should_abort_on_uncaught_toggle =
FIXED_ONE_BYTE_STRING(env->isolate(), "_shouldAbortOnUncaughtToggle");
CHECK(bootstrapper->Set(env->context(),
should_abort_on_uncaught_toggle,
@@ -130,4 +133,21 @@ void SetupBootstrapObject(Environment* env,
}
#undef BOOTSTRAP_METHOD
+namespace symbols {
+
+void Initialize(Local