From 416e7e1e6fb2ec5cc44e1c0084bb29b45c9b20b8 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Thu, 5 Nov 2026 18:42:18 +0000 Subject: [PATCH 01/20] src: add NewIsolate() overload taking a platform delegate The internal NewIsolate() always registers the isolate with the platform using a libuv loop, which ties V8's foreground tasks to that loop. Add an overload that registers an IsolatePlatformDelegate instead, so that embedder-level schedulers can run an isolate's foreground tasks themselves, and share the rest of the setup between both overloads. Assisted-by: Claude Opus 5.5 --- src/api/environment.cc | 38 ++++++++++++++++++++++++++++++-------- src/node_internals.h | 6 ++++++ 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/src/api/environment.cc b/src/api/environment.cc index 4a7b70168e6d..dd044a0a28b5 100644 --- a/src/api/environment.cc +++ b/src/api/environment.cc @@ -342,13 +342,12 @@ bool IsFirstSnapshotBlob(const char* data) { return first_snapshot_recorded && data == first_snapshot_blob.data; } -// TODO(joyeecheung): we may want to expose this, but then we need to be -// careful about what we override in the params. -Isolate* NewIsolate(Isolate::CreateParams* params, - uv_loop_t* event_loop, - MultiIsolatePlatform* platform, - const SnapshotData* snapshot_data, - const IsolateSettings& settings) { +template +static Isolate* NewIsolateImpl(Isolate::CreateParams* params, + MultiIsolatePlatform* platform, + const SnapshotData* snapshot_data, + const IsolateSettings& settings, + RegisterFn register_isolate) { IsolateGroup group = GetOrCreateIsolateGroup(); Isolate* isolate = Isolate::Allocate(group); if (isolate == nullptr) return nullptr; @@ -361,7 +360,7 @@ Isolate* NewIsolate(Isolate::CreateParams* params, // Register the isolate on the platform before the isolate gets initialized, // so that the isolate can access the platform during initialization. - platform->RegisterIsolate(isolate, event_loop); + register_isolate(isolate); // Ensure that there is always a CppHeap. if (settings.cpp_heap == nullptr) { @@ -378,6 +377,29 @@ Isolate* NewIsolate(Isolate::CreateParams* params, return isolate; } +// TODO(joyeecheung): we may want to expose this, but then we need to be +// careful about what we override in the params. +Isolate* NewIsolate(Isolate::CreateParams* params, + uv_loop_t* event_loop, + MultiIsolatePlatform* platform, + const SnapshotData* snapshot_data, + const IsolateSettings& settings) { + return NewIsolateImpl( + params, platform, snapshot_data, settings, [&](Isolate* isolate) { + platform->RegisterIsolate(isolate, event_loop); + }); +} + +Isolate* NewIsolate(Isolate::CreateParams* params, + IsolatePlatformDelegate* delegate, + MultiIsolatePlatform* platform, + const IsolateSettings& settings) { + return NewIsolateImpl( + params, platform, nullptr, settings, [&](Isolate* isolate) { + platform->RegisterIsolate(isolate, delegate); + }); +} + Isolate* NewIsolate(ArrayBufferAllocator* allocator, uv_loop_t* event_loop, MultiIsolatePlatform* platform, diff --git a/src/node_internals.h b/src/node_internals.h index bb45d73e9090..e5dcc76afeae 100644 --- a/src/node_internals.h +++ b/src/node_internals.h @@ -363,6 +363,12 @@ v8::Isolate* NewIsolate(v8::Isolate::CreateParams* params, MultiIsolatePlatform* platform, const SnapshotData* snapshot_data = nullptr, const IsolateSettings& settings = {}); +// Like the overload above, but registers the isolate with `delegate` as its +// foreground task runner instead of binding it to a libuv loop. +v8::Isolate* NewIsolate(v8::Isolate::CreateParams* params, + IsolatePlatformDelegate* delegate, + MultiIsolatePlatform* platform, + const IsolateSettings& settings); // This overload automatically picks the right 'main_script_id' if no callback // was provided by the embedder. v8::MaybeLocal StartExecution( From a66a5ad1d9c24aa3d6c4787394a664ff98b2862c Mon Sep 17 00:00:00 2001 From: James M Snell Date: Thu, 5 Nov 2026 18:42:28 +0000 Subject: [PATCH 02/20] green_threads: add experimental node:green_threads module Add GreenThread, which runs JavaScript in its own V8 isolate without an OS thread, event loop or Node.js environment of its own. Green threads are scheduled onto a process-wide pool of threads, each with a libuv loop. An isolate stays on the pool thread it was assigned to, and runs one task at a time with the isolate locked, followed by a microtask checkpoint. A green thread only gets the ECMAScript built-ins, the default context of the startup snapshot, plus postMessage(), onmessage, close(), queueMicrotask(), timers and a console that is forwarded to the host. The host can send messages, evaluate scripts and terminate the green thread. Uncaught exceptions, unhandled rejections and reaching the heap limit emit 'error' and stop the green thread. V8 foreground tasks for green isolates go through their own task runner, so they run on the right pool thread with the isolate locked. Terminations requested while an isolate is unlocked are tracked separately, because acquiring the lock discards them. Green threads are stopped when their host Environment is torn down, and before the platform is disposed when the process exits. The module is only available with --experimental-green-threads. Assisted-by: Claude Opus 5.5 --- doc/api/cli.md | 12 + doc/api/errors.md | 16 + doc/api/green_threads.md | 236 ++++ doc/api/index.md | 1 + doc/node-config-schema.json | 4 + doc/node.1 | 5 + lib/green_threads.js | 197 +++ lib/internal/bootstrap/realm.js | 1 + lib/internal/errors.js | 3 + node.gyp | 1 + src/api/environment.cc | 1 + src/async_wrap.h | 1 + src/node_binding.cc | 1 + src/node_builtins.cc | 1 + src/node_external_reference.h | 1 + src/node_green_thread.cc | 1254 +++++++++++++++++ src/node_internals.h | 6 + src/node_messaging.cc | 4 + src/node_messaging.h | 5 + src/node_options.cc | 4 + src/node_options.h | 1 + test/parallel/test-green-threads-basic.js | 179 +++ test/parallel/test-green-threads-errors.js | 70 + test/parallel/test-green-threads-flag.js | 36 + test/parallel/test-green-threads-lifetime.js | 64 + test/sequential/test-async-wrap-getasyncid.js | 7 + 26 files changed, 2111 insertions(+) create mode 100644 doc/api/green_threads.md create mode 100644 lib/green_threads.js create mode 100644 src/node_green_thread.cc create mode 100644 test/parallel/test-green-threads-basic.js create mode 100644 test/parallel/test-green-threads-errors.js create mode 100644 test/parallel/test-green-threads-flag.js create mode 100644 test/parallel/test-green-threads-lifetime.js diff --git a/doc/api/cli.md b/doc/api/cli.md index 415b65ac5efa..84e14219b555 100644 --- a/doc/api/cli.md +++ b/doc/api/cli.md @@ -1469,6 +1469,16 @@ added: Enable exposition of [EventSource Web API][] on the global scope. +### `--experimental-green-threads` + + + +> Stability: 1 - Experimental + +Enable the experimental [`node:green_threads`][] module. + ### `--experimental-import-meta-resolve` + + + +> Stability: 1 - Experimental + + + +The `node:green_threads` module runs JavaScript in lightweight, isolated +threads of execution. It is available only when the +[`--experimental-green-threads`][] flag is used. + +```mjs +import { GreenThread } from 'node:green_threads'; +``` + +```cjs +const { GreenThread } = require('node:green_threads'); +``` + +This module is only available under the `node:` scheme. Loading it emits an +`ExperimentalWarning`. + +Each green thread has its own V8 isolate, so it has its own heap and garbage +collector, and it can be terminated without affecting any other thread. +Unlike a [`Worker`][], a green thread does not get its own OS thread, event +loop, or Node.js environment. Green threads are scheduled onto a small, +process-wide pool of OS threads, which makes them much cheaper to create. + +A green thread only has the JavaScript built-ins defined by ECMAScript, plus +the following globals: + +* `postMessage(value)`: Sends a structured clone of `value` to the host. The + host receives it as a [`'message'`][] event. +* `onmessage`: Set this to a function to receive messages sent with + [`greenThread.postMessage()`][]. The function is called with an object + whose `data` property holds the message. +* `close()`: Stops the green thread once the current task has completed. The + exit code is `0`. +* `queueMicrotask(callback)` +* `setTimeout(callback, delay)`, `clearTimeout(id)`, + `setInterval(callback, delay)`, `clearInterval(id)`. Extra arguments are not + passed to `callback`. +* `console.log()`, `console.info()`, `console.warn()`, `console.error()`, + `console.debug()`: Forwarded to the host's `console`. Arguments that can't + be cloned are forwarded as strings. + +Green threads have no access to `require()`, `import`, `process`, or any +other Node.js API. + +```cjs +const { GreenThread } = require('node:green_threads'); + +const greenThread = new GreenThread(` + onmessage = ({ data }) => postMessage(data * 2); +`); +greenThread.on('message', (value) => { + console.log(value); // Prints: 42 + greenThread.terminate(); +}); +greenThread.postMessage(21); +``` + +An uncaught exception or unhandled promise rejection inside a green thread +emits an [`'error'`][] event on the host, followed by an [`'exit'`][] event +with exit code `1`. + +## Class: `GreenThread` + + + +* Extends: {EventEmitter} + +### `new GreenThread(source[, options])` + + + +* `source` {string} The script to run in the green thread. +* `options` {Object} + * `filename` {string} The file name used in stack traces. + **Default:** `'green-thread.js'`. + * `resourceLimits` {Object} + * `maxOldGenerationSizeMb` {number} The maximum size of the main heap, in + MB. `0` uses the V8 default. **Default:** `0`. + * `maxYoungGenerationSizeMb` {number} The maximum size of the heap space + for recently created objects, in MB. `0` uses the V8 default. + **Default:** `0`. + +Starts a green thread that runs `source`. If the heap limit is reached, the +green thread emits an [`'error'`][] event with an +[`ERR_GREEN_THREAD_OUT_OF_MEMORY`][] error and exits. + +### Event: `'error'` + + + +* `error` {any} + +Emitted when the green thread throws an uncaught exception, has an unhandled +promise rejection, or runs out of memory. The green thread then exits. + +### Event: `'exit'` + + + +* `exitCode` {integer} + +Emitted once the green thread has stopped. The exit code is `0` if the green +thread called `close()`, and `1` otherwise. + +### Event: `'message'` + + + +* `value` {any} + +Emitted when the green thread calls `postMessage()`. + +### `greenThread.evaluate(source[, options])` + + + +* `source` {string} +* `options` {Object} + * `filename` {string} The file name used in stack traces. + **Default:** `'green-thread-eval.js'`. +* Returns: {Promise} + +Runs `source` as a script in the green thread. The promise is fulfilled with a +structured clone of the script's completion value. If that value is a promise, +the green thread waits for it to settle first. The promise is rejected if the +script throws, or if the green thread exits first, with an +[`ERR_GREEN_THREAD_EXITED`][] error. + +Exceptions thrown by `source` do not stop the green thread. + +```cjs +const { GreenThread } = require('node:green_threads'); + +const greenThread = new GreenThread('globalThis.counter = 0;'); +(async () => { + console.log(await greenThread.evaluate('++counter')); // Prints: 1 + await greenThread.terminate(); +})(); +``` + +### `greenThread.exitCode` + + + +* Type: {integer|undefined} + +The exit code, or `undefined` while the green thread is running. + +### `greenThread.postMessage(value)` + + + +* `value` {any} + +Sends a structured clone of `value` to the green thread's `onmessage` handler. +Messages are dropped if `onmessage` is not set. Calling this after the green +thread has exited has no effect. + +### `greenThread.ref()` + + + +* Returns: {GreenThread} + +Undoes a previous `unref()` call. This is the default: a running green thread +keeps the event loop alive. + +### `greenThread.terminate()` + + + +* Returns: {Promise} Fulfilled with the exit code. + +Stops the green thread as soon as possible, interrupting any JavaScript that +is running. + +### `greenThread.unref()` + + + +* Returns: {GreenThread} + +Allows the process to exit while the green thread is still running. Any +green threads that are still running when the process exits are terminated. + +### `greenThread[Symbol.asyncDispose]()` + + + +* Returns: {Promise} + +Calls [`greenThread.terminate()`][]. + +[`'error'`]: #event-error +[`'exit'`]: #event-exit +[`'message'`]: #event-message +[`--experimental-green-threads`]: cli.md#--experimental-green-threads +[`ERR_GREEN_THREAD_EXITED`]: errors.md#err_green_thread_exited +[`ERR_GREEN_THREAD_OUT_OF_MEMORY`]: errors.md#err_green_thread_out_of_memory +[`Worker`]: worker_threads.md#class-worker +[`greenThread.postMessage()`]: #greenthreadpostmessagevalue +[`greenThread.terminate()`]: #greenthreadterminate diff --git a/doc/api/index.md b/doc/api/index.md index a30724c064e1..33d89ec642dc 100644 --- a/doc/api/index.md +++ b/doc/api/index.md @@ -29,6 +29,7 @@ * [Events](events.md) * [File system](fs.md) * [FFI](ffi.md) +* [Green threads](green_threads.md) * [Globals](globals.md) * [HTTP](http.md) * [HTTP/2](http2.md) diff --git a/doc/node-config-schema.json b/doc/node-config-schema.json index 0d2cb83b517d..6b5a6f2d76cc 100644 --- a/doc/node-config-schema.json +++ b/doc/node-config-schema.json @@ -351,6 +351,10 @@ "type": "boolean", "description": "expose experimental Navigator API on the global scope" }, + "experimental-green-threads": { + "type": "boolean", + "description": "experimental node:green_threads module" + }, "experimental-import-meta-resolve": { "type": "boolean", "description": "experimental ES Module import.meta.resolve() parentURL support" diff --git a/doc/node.1 b/doc/node.1 index 7419eb9a34d2..5f6ee3e5362c 100644 --- a/doc/node.1 +++ b/doc/node.1 @@ -861,6 +861,9 @@ dtls documentation for details. .It Fl -experimental-eventsource Enable exposition of EventSource Web API on the global scope. . +.It Fl -experimental-green-threads +Enable the experimental \fBnode:green_threads\fR module. +. .It Fl -experimental-import-meta-resolve Enable experimental \fBimport.meta.resolve()\fR parent URL support, which allows passing a second \fBparentURL\fR argument for contextual resolution. @@ -2269,6 +2272,8 @@ one is included in the list below. .It \fB--experimental-eventsource\fR .It +\fB--experimental-green-threads\fR +.It \fB--experimental-import-meta-resolve\fR .It \fB--experimental-import-text\fR diff --git a/lib/green_threads.js b/lib/green_threads.js new file mode 100644 index 000000000000..66a160ebe087 --- /dev/null +++ b/lib/green_threads.js @@ -0,0 +1,197 @@ +'use strict'; + +const { + PromiseReject, + PromiseResolve, + PromiseWithResolvers, + ReflectApply, + SafeMap, + SymbolAsyncDispose, +} = primordials; + +const EventEmitter = require('events'); +const { + codes: { + ERR_GREEN_THREAD_EXITED, + ERR_GREEN_THREAD_OUT_OF_MEMORY, + }, +} = require('internal/errors'); +const { + emitExperimentalWarning, + kEmptyObject, +} = require('internal/util'); +const { + validateNumber, + validateObject, + validateString, +} = require('internal/validators'); +const { + GreenThread: GreenThreadHandle, + kEventMessage, + kEventConsole, + kEventEvalResult, + kEventError, + kEventExit, + kErrorOutOfMemory, +} = internalBinding('green_thread'); + +emitExperimentalWarning('Green threads'); + +const kConsoleMethods = ['log', 'info', 'warn', 'error', 'debug']; + +let hostConsole; + +class GreenThread extends EventEmitter { + #handle; + #exitCode; + #exited = PromiseWithResolvers(); + #nextEvalId = 0; + #pendingEvals = new SafeMap(); + + /** + * @param {string} source + * @param {{ + * filename?: string, + * resourceLimits?: { + * maxOldGenerationSizeMb?: number, + * maxYoungGenerationSizeMb?: number, + * }, + * }} [options] + */ + constructor(source, options = kEmptyObject) { + super(); + validateString(source, 'source'); + validateObject(options, 'options'); + const { + filename = 'green-thread.js', + resourceLimits = kEmptyObject, + } = options; + validateString(filename, 'options.filename'); + validateObject(resourceLimits, 'options.resourceLimits'); + const { + maxOldGenerationSizeMb = 0, + maxYoungGenerationSizeMb = 0, + } = resourceLimits; + validateNumber(maxOldGenerationSizeMb, + 'options.resourceLimits.maxOldGenerationSizeMb', 0); + validateNumber(maxYoungGenerationSizeMb, + 'options.resourceLimits.maxYoungGenerationSizeMb', 0); + + this.#handle = new GreenThreadHandle(source, + filename, + maxOldGenerationSizeMb, + maxYoungGenerationSizeMb); + this.#handle.onevent = (type, id, ok, value) => { + this.#onEvent(type, id, ok, value); + }; + } + + /** + * The exit code, or `undefined` while the green thread is running. + * @type {number|undefined} + */ + get exitCode() { + return this.#exitCode; + } + + /** + * Sends a structured clone of `value` to the green thread's `onmessage`. + * @param {any} value + */ + postMessage(value) { + if (this.#exitCode !== undefined) return; + this.#handle.postMessage(value); + } + + /** + * Runs `source` as a script in the green thread. Resolves with a + * structured clone of the completion value, awaiting it if it is a promise. + * @param {string} source + * @param {{ filename?: string }} [options] + * @returns {Promise} + */ + evaluate(source, options = kEmptyObject) { + validateString(source, 'source'); + validateObject(options, 'options'); + const { filename = 'green-thread-eval.js' } = options; + validateString(filename, 'options.filename'); + if (this.#exitCode !== undefined) { + return PromiseReject(new ERR_GREEN_THREAD_EXITED()); + } + const id = ++this.#nextEvalId; + const pending = PromiseWithResolvers(); + this.#pendingEvals.set(id, pending); + this.#handle.evaluate(id, source, filename); + return pending.promise; + } + + /** + * Stops the green thread, interrupting any running code. + * @returns {Promise} The exit code. + */ + terminate() { + if (this.#exitCode !== undefined) return PromiseResolve(this.#exitCode); + this.#handle.terminate(); + return this.#exited.promise; + } + + ref() { + if (this.#exitCode === undefined) this.#handle.ref(); + return this; + } + + unref() { + if (this.#exitCode === undefined) this.#handle.unref(); + return this; + } + + async [SymbolAsyncDispose]() { + await this.terminate(); + } + + #onEvent(type, id, ok, value) { + switch (type) { + case kEventMessage: + this.emit('message', value); + break; + case kEventConsole: { + hostConsole ??= require('internal/console/global'); + ReflectApply(hostConsole[kConsoleMethods[id]], hostConsole, value); + break; + } + case kEventEvalResult: { + const pending = this.#pendingEvals.get(id); + if (pending === undefined) break; + this.#pendingEvals.delete(id); + if (ok) { + pending.resolve(value); + } else { + pending.reject(value); + } + break; + } + case kEventError: + this.emit('error', id === kErrorOutOfMemory ? + new ERR_GREEN_THREAD_OUT_OF_MEMORY() : value); + break; + case kEventExit: + this.#onExit(id); + break; + } + } + + #onExit(code) { + this.#exitCode = code; + this.#handle.close(); + for (const { reject } of this.#pendingEvals.values()) { + reject(new ERR_GREEN_THREAD_EXITED()); + } + this.#pendingEvals.clear(); + this.#exited.resolve(code); + this.emit('exit', code); + } +} + +module.exports = { + GreenThread, +}; diff --git a/lib/internal/bootstrap/realm.js b/lib/internal/bootstrap/realm.js index fd3e4eb64c80..8eec8664152d 100644 --- a/lib/internal/bootstrap/realm.js +++ b/lib/internal/bootstrap/realm.js @@ -201,6 +201,7 @@ const builtinModulePolicies = [ ['bench/reporters', true, '--experimental-bench'], ['dtls', true, '--experimental-dtls'], ['ffi', true, '--experimental-ffi'], + ['green_threads', true, '--experimental-green-threads'], ['sea', true, null], ['sqlite', true, '--experimental-sqlite'], ['quic', true, '--experimental-quic'], diff --git a/lib/internal/errors.js b/lib/internal/errors.js index e2ee2cdd340d..73d93ef8f81d 100644 --- a/lib/internal/errors.js +++ b/lib/internal/errors.js @@ -1259,6 +1259,9 @@ E('ERR_FS_CP_UNKNOWN', 'Cannot copy an unknown file type', SystemError); E('ERR_FS_EISDIR', 'Path is a directory', SystemError, HideStackFramesError); E('ERR_FS_FILE_TOO_LARGE', 'File size (%s) is greater than 2 GiB', RangeError); E('ERR_FS_WATCH_QUEUE_OVERFLOW', 'fs.watch() queued more than %d events', Error); +E('ERR_GREEN_THREAD_EXITED', 'The green thread has exited', Error); +E('ERR_GREEN_THREAD_OUT_OF_MEMORY', + 'The green thread was terminated due to reaching its memory limit', Error); E('ERR_HTTP2_ALTSVC_INVALID_ORIGIN', 'HTTP/2 ALTSVC frames require a valid origin', TypeError); E('ERR_HTTP2_ALTSVC_LENGTH', diff --git a/node.gyp b/node.gyp index 714609b03e66..b19d5c13b7d4 100644 --- a/node.gyp +++ b/node.gyp @@ -140,6 +140,7 @@ 'src/node_errors.cc', 'src/node_external_reference.cc', 'src/node_file.cc', + 'src/node_green_thread.cc', 'src/node_file_utils.cc', 'src/node_http_parser.cc', 'src/node_http2.cc', diff --git a/src/api/environment.cc b/src/api/environment.cc index dd044a0a28b5..8cdd1189caf4 100644 --- a/src/api/environment.cc +++ b/src/api/environment.cc @@ -1190,6 +1190,7 @@ void DefaultProcessExitHandlerInternal(Environment* env, ExitCode exit_code) { env->set_stopping(true); env->set_can_call_into_js(false); env->stop_sub_worker_contexts(); + green_thread::TerminateAll(); env->isolate()->DumpAndResetStats(); // The tracing agent could be in the process of writing data using the // threadpool. Stop it before shutting down libuv. The rest of the tracing diff --git a/src/async_wrap.h b/src/async_wrap.h index a14d84088b38..883e6f87bcb4 100644 --- a/src/async_wrap.h +++ b/src/async_wrap.h @@ -42,6 +42,7 @@ namespace node { V(FSEVENTWRAP) \ V(FSREQCALLBACK) \ V(GLOBREQUEST) \ + V(GREENTHREAD) \ V(FSREQPROMISE) \ V(GETADDRINFOREQWRAP) \ V(GETNAMEINFOREQWRAP) \ diff --git a/src/node_binding.cc b/src/node_binding.cc index 3c7dee9bb33d..f940dd1f6d90 100644 --- a/src/node_binding.cc +++ b/src/node_binding.cc @@ -71,6 +71,7 @@ V(fs_dir) \ V(fs_event_wrap) \ V(glob) \ + V(green_thread) \ V(heap_utils) \ V(http2) \ V(http_parser) \ diff --git a/src/node_builtins.cc b/src/node_builtins.cc index 21af988c95d8..a73f3229ce32 100644 --- a/src/node_builtins.cc +++ b/src/node_builtins.cc @@ -162,6 +162,7 @@ BuiltinLoader::BuiltinCategories BuiltinLoader::GetBuiltinCategories() const { "bench/reporters", // Experimental. "dtls", // Experimental. "ffi", // Experimental. + "green_threads", // Experimental. "quic", // Experimental. "sqlite", // Experimental. "stream/iter", // Experimental. diff --git a/src/node_external_reference.h b/src/node_external_reference.h index abdf6a36cbcb..e3a7836440cc 100644 --- a/src/node_external_reference.h +++ b/src/node_external_reference.h @@ -78,6 +78,7 @@ class ExternalReferenceRegistry { V(fs_dir) \ V(fs_event_wrap) \ V(glob) \ + V(green_thread) \ V(handle_wrap) \ V(heap_utils) \ V(http_parser) \ diff --git a/src/node_green_thread.cc b/src/node_green_thread.cc new file mode 100644 index 000000000000..d3eb9d7b9e95 --- /dev/null +++ b/src/node_green_thread.cc @@ -0,0 +1,1254 @@ +#include "async_wrap-inl.h" +#include "env-inl.h" +#include "handle_wrap.h" +#include "node_external_reference.h" +#include "node_internals.h" +#include "node_messaging.h" +#include "node_mutex.h" +#include "util-inl.h" +#include "uv.h" +#include "v8.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace node { +namespace green_thread { + +using v8::Array; +using v8::Boolean; +using v8::ConstructorBehavior; +using v8::Context; +using v8::Exception; +using v8::Function; +using v8::FunctionCallbackInfo; +using v8::FunctionTemplate; +using v8::Global; +using v8::HandleScope; +using v8::IdleTask; +using v8::Int32; +using v8::Integer; +using v8::Isolate; +using v8::Just; +using v8::Local; +using v8::LocalVector; +using v8::Locker; +using v8::Maybe; +using v8::MaybeLocal; +using v8::Message; +using v8::Name; +using v8::Nothing; +using v8::Number; +using v8::Object; +using v8::Promise; +using v8::PromiseRejectEvent; +using v8::PromiseRejectMessage; +using v8::PropertyAttribute; +using v8::Script; +using v8::ScriptCompiler; +using v8::ScriptOrigin; +using v8::SourceLocation; +using v8::String; +using v8::Task; +using v8::TaskRunner; +using v8::TryCatch; +using v8::Uint32; +using v8::Undefined; +using v8::Value; +using v8::ValueDeserializer; +using v8::ValueSerializer; + +namespace { + +constexpr size_t kStackSize = 4 * 1024 * 1024; +constexpr size_t kStackBufferSize = 192 * 1024; +constexpr unsigned int kMaxPoolThreads = 8; +constexpr size_t kMB = 1024 * 1024; +constexpr size_t kExtraHeapAllowance = 16 * kMB; + +// Event types sent from a green thread to its host. +enum EventType : int32_t { + kEventMessage = 0, + kEventConsole, + kEventEvalResult, + kEventError, + kEventExit, +}; + +// Values of the `id` field of kEventError events. +enum ErrorKind : int32_t { + kErrorException = 0, + kErrorOutOfMemory, +}; + +using Bytes = std::vector; + +// A pool thread owns a libuv loop and runs the tasks posted to it, in order, +// on that loop. Pool threads live until the process exits. +class PoolThread { + public: + using Callback = std::function; + + explicit PoolThread(size_t index) : index_(index) {} + + void Start() { + CHECK_EQ(uv_loop_init(&loop_), 0); + CHECK_EQ(uv_async_init(&loop_, &wakeup_, OnWakeup), 0); + + uv_thread_options_t options; + options.flags = UV_THREAD_HAS_STACK_SIZE; + options.stack_size = kStackSize; + CHECK_EQ( + uv_thread_create_ex( + &tid_, + &options, + [](void* arg) { + PoolThread* self = static_cast(arg); + const uintptr_t stack_top = reinterpret_cast(&arg); + // Leave some room for C++ code below the V8 limit. + self->stack_limit_ = stack_top - (kStackSize - kStackBufferSize); + char name[16]; + snprintf(name, sizeof(name), "GreenThread%zu", self->index_); + uv_thread_setname(name); + uv_run(&self->loop_, UV_RUN_DEFAULT); + }, + this), + 0); + } + + // May be called from any thread. + void Post(Callback callback) { + { + Mutex::ScopedLock lock(mutex_); + queue_.push_back(std::move(callback)); + } + CHECK_EQ(uv_async_send(&wakeup_), 0); + } + + uv_loop_t* loop() { return &loop_; } + uintptr_t stack_limit() const { return stack_limit_; } + + private: + static void OnWakeup(uv_async_t* handle) { + PoolThread* self = ContainerOf(&PoolThread::wakeup_, handle); + std::deque callbacks; + { + Mutex::ScopedLock lock(self->mutex_); + callbacks.swap(self->queue_); + } + for (Callback& callback : callbacks) callback(); + } + + const size_t index_; + uv_thread_t tid_; + uv_loop_t loop_; + uv_async_t wakeup_; + Mutex mutex_; + std::deque queue_; + uintptr_t stack_limit_ = 0; +}; + +// The process-wide set of pool threads. It is created on first use and +// intentionally never destroyed. +class GreenThreadPool { + public: + static GreenThreadPool* Get() { + static GreenThreadPool* pool = new GreenThreadPool(); + return pool; + } + + PoolThread* Next() { + size_t index = next_.fetch_add(1, std::memory_order_relaxed); + return threads_[index % threads_.size()].get(); + } + + private: + GreenThreadPool() { + unsigned int count = + std::clamp(uv_available_parallelism(), 1u, kMaxPoolThreads); + for (unsigned int i = 0; i < count; i++) { + threads_.push_back(std::make_unique(i)); + threads_.back()->Start(); + } + } + + std::vector> threads_; + std::atomic next_{0}; +}; + +class SerializerDelegate final : public ValueSerializer::Delegate { + public: + SerializerDelegate(Isolate* isolate, Local context, bool is_host) + : isolate_(isolate), context_(context), is_host_(is_host) {} + + void ThrowDataCloneError(Local message) override { + if (is_host_) { + worker::ThrowDataCloneException(context_, message); + return; + } + // Green threads have no DOMException, so use a plain Error with the + // same name. + Local error = Exception::Error(message); + error.As() + ->Set(context_, + FIXED_ONE_BYTE_STRING(isolate_, "name"), + FIXED_ONE_BYTE_STRING(isolate_, "DataCloneError")) + .Check(); + isolate_->ThrowException(error); + } + + private: + Isolate* isolate_; + Local context_; + bool is_host_; +}; + +// Throws a JS exception and returns std::nullopt if `value` is not cloneable. +std::optional Serialize(Isolate* isolate, + Local context, + Local value, + bool is_host) { + SerializerDelegate delegate(isolate, context, is_host); + ValueSerializer serializer(isolate, &delegate); + serializer.WriteHeader(); + if (serializer.WriteValue(context, value).IsNothing()) return std::nullopt; + std::pair buffer = serializer.Release(); + Bytes bytes(buffer.first, buffer.first + buffer.second); + free(buffer.first); + return bytes; +} + +MaybeLocal Deserialize(Isolate* isolate, + Local context, + const Bytes& bytes) { + ValueDeserializer deserializer(isolate, bytes.data(), bytes.size()); + if (deserializer.ReadHeader(context).IsNothing()) return {}; + return deserializer.ReadValue(context); +} + +struct HostEvent { + EventType type; + int32_t id = 0; + bool ok = true; + Bytes payload; +}; + +// Carries events from a green thread to its host. The host closes it before +// closing its uv_async_t, after which events are dropped. +class HostChannel { + public: + explicit HostChannel(uv_async_t* async) : async_(async) {} + + void Push(HostEvent&& event) { + Mutex::ScopedLock lock(mutex_); + if (async_ == nullptr) return; + queue_.push_back(std::move(event)); + CHECK_EQ(uv_async_send(async_), 0); + } + + std::deque Take() { + std::deque events; + Mutex::ScopedLock lock(mutex_); + events.swap(queue_); + return events; + } + + void Close() { + Mutex::ScopedLock lock(mutex_); + async_ = nullptr; + queue_.clear(); + } + + private: + Mutex mutex_; + uv_async_t* async_; + std::deque queue_; +}; + +class GreenIsolate; + +// NOLINTNEXTLINE(runtime/thread_local) +thread_local GreenIsolate* current_green_isolate = nullptr; + +// All green isolates that have not been disposed yet, so that they can be +// stopped before the process exits. +Mutex live_green_isolates_mutex; +std::unordered_map> + live_green_isolates; + +// The foreground task runner V8 uses for a green isolate. Tasks are run on the +// green isolate's pool thread, with the isolate locked. +class GreenTaskRunner final : public TaskRunner { + public: + explicit GreenTaskRunner(std::weak_ptr owner) + : owner_(std::move(owner)) {} + + bool IdleTasksEnabled() override { return false; } + bool NonNestableTasksEnabled() const override { return true; } + bool NonNestableDelayedTasksEnabled() const override { return true; } + + private: + void PostTaskImpl(std::unique_ptr task, + const SourceLocation& location) override { + Post(std::move(task), 0); + } + void PostNonNestableTaskImpl(std::unique_ptr task, + const SourceLocation& location) override { + Post(std::move(task), 0); + } + void PostDelayedTaskImpl(std::unique_ptr task, + double delay_in_seconds, + const SourceLocation& location) override { + Post(std::move(task), delay_in_seconds); + } + void PostNonNestableDelayedTaskImpl(std::unique_ptr task, + double delay_in_seconds, + const SourceLocation& location) override { + Post(std::move(task), delay_in_seconds); + } + void PostIdleTaskImpl(std::unique_ptr task, + const SourceLocation& location) override { + UNREACHABLE(); + } + + inline void Post(std::unique_ptr task, double delay_in_seconds); + + std::weak_ptr owner_; +}; + +// The state of one green thread: a V8 isolate pinned to a pool thread, plus +// its timers. Except where noted, methods run on the pool thread. +class GreenIsolate final : public IsolatePlatformDelegate, + public std::enable_shared_from_this { + public: + struct Options { + std::string source; + std::string filename; + size_t max_old_generation_size = 0; + size_t max_young_generation_size = 0; + }; + + GreenIsolate(MultiIsolatePlatform* platform, + std::shared_ptr host) + : platform_(platform), + thread_(GreenThreadPool::Get()->Next()), + host_(std::move(host)) {} + + // Must be called once, from the host thread, right after construction. + void Start(Options options) { + task_runner_ = std::make_shared(weak_from_this()); + { + Mutex::ScopedLock lock(live_green_isolates_mutex); + live_green_isolates.emplace(this, weak_from_this()); + } + Post([options = std::move(options)](GreenIsolate* self) { + self->Initialize(options); + }); + } + + // IsolatePlatformDelegate. May be called from any thread. + std::shared_ptr GetForegroundTaskRunner() override { + return task_runner_; + } + bool IdleTasksEnabled() override { return false; } + + // May be called from any thread. + void PostMessage(Bytes data) { + Post([data = std::move(data)](GreenIsolate* self) { + self->RunTask([&](Isolate* isolate, Local context) { + self->DeliverMessage(isolate, context, data); + }); + }); + } + + // May be called from any thread. + void Evaluate(int32_t id, std::string source, std::string filename) { + Post([=](GreenIsolate* self) { + self->RunTask([&](Isolate* isolate, Local context) { + self->EvaluateSource(isolate, context, id, source, filename); + }); + }); + } + + // May be called from any thread. Interrupts running JS, if any. + void Terminate() { + terminate_requested_ = true; + { + Mutex::ScopedLock lock(isolate_mutex_); + if (isolate_ != nullptr) isolate_->TerminateExecution(); + } + Post([](GreenIsolate* self) { self->Shutdown(); }); + } + + // May be called from any thread. + void WaitForDisposal() { + Mutex::ScopedLock lock(disposal_mutex_); + while (!disposed_) disposal_cv_.Wait(lock); + } + + // May be called from any thread. + void PostV8Task(std::unique_ptr task, double delay_in_seconds) { + auto holder = std::make_shared>(std::move(task)); + Post([holder, delay_in_seconds](GreenIsolate* self) { + if (delay_in_seconds <= 0) { + self->RunV8Task(std::move(*holder)); + } else { + self->ScheduleV8Task(std::move(*holder), delay_in_seconds); + } + }); + } + + private: + enum class State { kPending, kRunning, kStopped }; + + struct Timer { + uv_timer_t handle; + GreenIsolate* owner; + uint32_t id; + bool repeat; + Global callback; + }; + + struct DelayedV8Task { + uv_timer_t handle; + GreenIsolate* owner; + std::unique_ptr task; + }; + + class CurrentScope { + public: + explicit CurrentScope(GreenIsolate* green) + : previous_(current_green_isolate) { + current_green_isolate = green; + } + ~CurrentScope() { current_green_isolate = previous_; } + + private: + GreenIsolate* previous_; + }; + + template + void Post(Fn&& fn) { + thread_->Post([self = shared_from_this(), fn = std::forward(fn)]() { + fn(self.get()); + }); + } + + void Initialize(const Options& options) { + CHECK_EQ(state_, State::kPending); + if (terminate_requested_) return Shutdown(); + + Isolate::CreateParams params; + params.array_buffer_allocator_shared = ArrayBufferAllocator::Create(); + if (options.max_old_generation_size > 0) { + params.constraints.set_max_old_generation_size_in_bytes( + options.max_old_generation_size); + } + if (options.max_young_generation_size > 0) { + params.constraints.set_max_young_generation_size_in_bytes( + options.max_young_generation_size); + } + + // Node.js' default callbacks expect a Node.js context, so skip or replace + // them. + IsolateSettings settings; + settings.flags = SHOULD_NOT_SET_PREPARE_STACK_TRACE_CALLBACK | + SHOULD_NOT_SET_WASM_STREAMING_CALLBACK; + settings.promise_reject_callback = OnPromiseReject; + settings.allow_wasm_code_generation_callback = + [](Local, Local) { return true; }; + + Isolate* isolate = NewIsolate(¶ms, this, platform_, settings); + if (isolate == nullptr) { + fatal_ = true; + return Shutdown(); + } + { + Mutex::ScopedLock lock(isolate_mutex_); + isolate_ = isolate; + } + + { + Locker locker(isolate); + Isolate::Scope isolate_scope(isolate); + isolate->SetStackLimit(thread_->stack_limit()); + HandleScope handle_scope(isolate); + isolate->AddNearHeapLimitCallback(NearHeapLimit, this); + isolate->AddMessageListenerWithErrorLevel( + OnMessage, Isolate::MessageErrorLevel::kMessageError); + // The default context of the startup snapshot contains only what V8 + // creates. + Local context = Context::New(isolate); + if (context.IsEmpty()) { + fatal_ = true; + } else { + Context::Scope context_scope(context); + context_.Reset(isolate, context); + if (!InstallGlobals(isolate, context)) fatal_ = true; + } + } + if (fatal_) return Shutdown(); + + state_ = State::kRunning; + RunTask([&](Isolate* isolate, Local context) { + Local