diff --git a/lib/internal/bootstrap/switches/is_main_thread.js b/lib/internal/bootstrap/switches/is_main_thread.js index e9e37275b17f..3e18fe6e3228 100644 --- a/lib/internal/bootstrap/switches/is_main_thread.js +++ b/lib/internal/bootstrap/switches/is_main_thread.js @@ -312,6 +312,9 @@ if (isBuildingSnapshot()) { require('internal/modules/typescript'); require('internal/blob'); require('internal/dns/utils'); + // Workers restore this context before loading internal/worker/io, which + // needs the initialized stream constructors for its stdio subclasses. + require('stream'); } // Needed to refresh the time origin. diff --git a/lib/internal/streams/duplex.js b/lib/internal/streams/duplex.js index dce8bd6e0fdf..6d31f62d0998 100644 --- a/lib/internal/streams/duplex.js +++ b/lib/internal/streams/duplex.js @@ -46,15 +46,14 @@ const { const destroyImpl = require('internal/streams/destroy'); const { kOnConstructed } = require('internal/streams/utils'); -ObjectSetPrototypeOf(Duplex.prototype, Readable.prototype); -ObjectSetPrototypeOf(Duplex, Readable); - { const keys = ObjectKeys(Writable.prototype); // Allow the keys array to be GC'ed. for (let i = 0; i < keys.length; i++) { const method = keys[i]; - Duplex.prototype[method] ||= Writable.prototype[method]; + if (!Readable.prototype[method]) { + Duplex.prototype[method] = Writable.prototype[method]; + } } } @@ -201,3 +200,8 @@ Duplex.from = function(body) { duplexify ??= require('internal/streams/duplexify'); return duplexify(body, 'body'); }; + +// Define own methods before inheriting from constructors or prototypes that +// user code may have frozen before this module was loaded. +ObjectSetPrototypeOf(Duplex.prototype, Readable.prototype); +ObjectSetPrototypeOf(Duplex, Readable); diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js index a5cbe1af428f..87b82b65b56b 100644 --- a/lib/internal/streams/operators.js +++ b/lib/internal/streams/operators.js @@ -6,10 +6,13 @@ const { MathFloor, Number, NumberIsNaN, + ObjectDefineProperty, + ObjectKeys, Promise, PromisePrototypeThen, PromiseReject, PromiseResolve, + ReflectApply, Symbol, } = primordials; @@ -18,6 +21,7 @@ const { AbortController, AbortSignal } = require('internal/abort_controller'); const { AbortError, codes: { + ERR_ILLEGAL_CONSTRUCTOR, ERR_MISSING_ARGS, ERR_OUT_OF_RANGE, }, @@ -31,6 +35,7 @@ const { const { kWeakHandler, kResistStopPropagation } = require('internal/event_target'); const destroyImpl = require('internal/streams/destroy'); const { finished } = require('internal/streams/end-of-stream'); +const Stream = require('stream'); const kEmpty = Symbol('kEmpty'); const kEof = Symbol('kEof'); @@ -373,7 +378,7 @@ function take(number, options = undefined) { }.call(this); } -module.exports.streamReturningOperators = { +const streamReturningOperators = { drop, filter, flatMap, @@ -381,7 +386,7 @@ module.exports.streamReturningOperators = { take, }; -module.exports.promiseReturningOperators = { +const promiseReturningOperators = { every, forEach, reduce, @@ -389,3 +394,33 @@ module.exports.promiseReturningOperators = { some, find, }; + +const streamKeys = ObjectKeys(streamReturningOperators); +for (let i = 0; i < streamKeys.length; i++) { + const key = streamKeys[i]; + const op = streamReturningOperators[key]; + function fn(...args) { + if (new.target) { + throw new ERR_ILLEGAL_CONSTRUCTOR(); + } + return Stream.Readable.from(ReflectApply(op, this, args)); + } + ObjectDefineProperty(fn, 'name', { __proto__: null, value: op.name }); + ObjectDefineProperty(fn, 'length', { __proto__: null, value: op.length }); + module.exports[key] = fn; +} + +const promiseKeys = ObjectKeys(promiseReturningOperators); +for (let i = 0; i < promiseKeys.length; i++) { + const key = promiseKeys[i]; + const op = promiseReturningOperators[key]; + function fn(...args) { + if (new.target) { + throw new ERR_ILLEGAL_CONSTRUCTOR(); + } + return ReflectApply(op, this, args); + } + ObjectDefineProperty(fn, 'name', { __proto__: null, value: op.name }); + ObjectDefineProperty(fn, 'length', { __proto__: null, value: op.length }); + module.exports[key] = fn; +} diff --git a/lib/internal/streams/passthrough.js b/lib/internal/streams/passthrough.js index d37f9caf0116..587361295111 100644 --- a/lib/internal/streams/passthrough.js +++ b/lib/internal/streams/passthrough.js @@ -32,8 +32,6 @@ const { module.exports = PassThrough; const Transform = require('internal/streams/transform'); -ObjectSetPrototypeOf(PassThrough.prototype, Transform.prototype); -ObjectSetPrototypeOf(PassThrough, Transform); function PassThrough(options) { if (!(this instanceof PassThrough)) @@ -45,3 +43,7 @@ function PassThrough(options) { PassThrough.prototype._transform = function(chunk, encoding, cb) { cb(null, chunk); }; + +// Install the override before inheriting a potentially frozen prototype. +ObjectSetPrototypeOf(PassThrough.prototype, Transform.prototype); +ObjectSetPrototypeOf(PassThrough, Transform); diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js index 546ff579d8eb..6937d4354340 100644 --- a/lib/internal/streams/pipeline.js +++ b/lib/internal/streams/pipeline.js @@ -5,13 +5,17 @@ const { ArrayIsArray, + ObjectDefineProperty, Promise, SymbolAsyncIterator, SymbolDispose, } = primordials; const { eos } = require('internal/streams/end-of-stream'); -const { once } = require('internal/util'); +const { + once, + promisify: { custom: customPromisify }, +} = require('internal/util'); const destroyImpl = require('internal/streams/destroy'); const Duplex = require('internal/streams/duplex'); const { @@ -469,4 +473,12 @@ function pipe(src, dst, finish, finishOnlyHandleError, { end }) { return eos(dst, { readable: false, writable: true }, finish); } +ObjectDefineProperty(pipeline, customPromisify, { + __proto__: null, + enumerable: true, + get() { + return require('stream/promises').pipeline; + }, +}); + module.exports = { pipelineImpl, pipeline }; diff --git a/lib/internal/streams/transform.js b/lib/internal/streams/transform.js index 2f4d498bc780..2d16f2568edc 100644 --- a/lib/internal/streams/transform.js +++ b/lib/internal/streams/transform.js @@ -74,8 +74,6 @@ const { } = require('internal/errors').codes; const Duplex = require('internal/streams/duplex'); const { getHighWaterMark } = require('internal/streams/state'); -ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype); -ObjectSetPrototypeOf(Transform, Duplex); const kCallback = Symbol('kCallback'); @@ -202,3 +200,7 @@ Transform.prototype._read = function() { callback(); } }; + +// Install overrides before inheriting potentially frozen prototype methods. +ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype); +ObjectSetPrototypeOf(Transform, Duplex); diff --git a/lib/stream.js b/lib/stream.js index 078fa092f483..5538df766a7b 100644 --- a/lib/stream.js +++ b/lib/stream.js @@ -23,32 +23,20 @@ const { ObjectDefineProperty, - ObjectKeys, - ReflectApply, + ObjectSetPrototypeOf, } = primordials; const { + defineLazyProperties, promisify: { custom: customPromisify }, } = require('internal/util'); -const { - streamReturningOperators, - promiseReturningOperators, -} = require('internal/streams/operators'); - -const { - codes: { - ERR_ILLEGAL_CONSTRUCTOR, - }, -} = require('internal/errors'); -const compose = require('internal/streams/compose'); const { setDefaultHighWaterMark, getDefaultHighWaterMark } = require('internal/streams/state'); -const { pipeline } = require('internal/streams/pipeline'); const { destroyer } = require('internal/streams/destroy'); const { eos } = require('internal/streams/end-of-stream'); const internalBuffer = require('internal/buffer'); -const promises = require('stream/promises'); +let promises; const utils = require('internal/streams/utils'); const { isArrayBufferView, isUint8Array } = require('internal/util/types'); @@ -61,57 +49,29 @@ Stream.isReadable = utils.isReadable; Stream.isWritable = utils.isWritable; Stream.Readable = require('internal/streams/readable'); -const streamKeys = ObjectKeys(streamReturningOperators); -for (let i = 0; i < streamKeys.length; i++) { - const key = streamKeys[i]; - const op = streamReturningOperators[key]; - function fn(...args) { - if (new.target) { - throw new ERR_ILLEGAL_CONSTRUCTOR(); - } - return Stream.Readable.from(ReflectApply(op, this, args)); - } - ObjectDefineProperty(fn, 'name', { __proto__: null, value: op.name }); - ObjectDefineProperty(fn, 'length', { __proto__: null, value: op.length }); - ObjectDefineProperty(Stream.Readable.prototype, key, { - __proto__: null, - value: fn, - enumerable: false, - configurable: true, - writable: true, - }); -} -const promiseKeys = ObjectKeys(promiseReturningOperators); -for (let i = 0; i < promiseKeys.length; i++) { - const key = promiseKeys[i]; - const op = promiseReturningOperators[key]; - function fn(...args) { - if (new.target) { - throw new ERR_ILLEGAL_CONSTRUCTOR(); - } - return ReflectApply(op, this, args); - } - ObjectDefineProperty(fn, 'name', { __proto__: null, value: op.name }); - ObjectDefineProperty(fn, 'length', { __proto__: null, value: op.length }); - ObjectDefineProperty(Stream.Readable.prototype, key, { - __proto__: null, - value: fn, - enumerable: false, - configurable: true, - writable: true, - }); -} +defineLazyProperties(Stream.Readable.prototype, 'internal/streams/operators', [ + 'drop', 'filter', 'flatMap', 'map', 'take', + 'every', 'forEach', 'reduce', 'toArray', 'some', 'find', +], false); Stream.Writable = require('internal/streams/writable'); -Stream.Duplex = require('internal/streams/duplex'); -Stream.Transform = require('internal/streams/transform'); -Stream.PassThrough = require('internal/streams/passthrough'); -Stream.duplexPair = require('internal/streams/duplexpair'); -Stream.pipeline = pipeline; +// These modules export a single function rather than named properties. +// Set the prototype after creation to retain fast properties. +const lazy = ObjectSetPrototypeOf({ + Duplex: 'internal/streams/duplex', + Transform: 'internal/streams/transform', + PassThrough: 'internal/streams/passthrough', + duplexPair: 'internal/streams/duplexpair', + compose: 'internal/streams/compose', +}, null); +defineLazyProperties(Stream, lazy, [ + 'Duplex', 'Transform', 'PassThrough', 'duplexPair', +]); +defineLazyProperties(Stream, 'internal/streams/pipeline', ['pipeline']); const { addAbortSignal } = require('internal/streams/add-abort-signal'); Stream.addAbortSignal = addAbortSignal; Stream.finished = eos; Stream.destroy = destroyer; -Stream.compose = compose; +defineLazyProperties(Stream, lazy, ['compose']); Stream.setDefaultHighWaterMark = setDefaultHighWaterMark; Stream.getDefaultHighWaterMark = getDefaultHighWaterMark; @@ -120,15 +80,7 @@ ObjectDefineProperty(Stream, 'promises', { configurable: true, enumerable: true, get() { - return promises; - }, -}); - -ObjectDefineProperty(pipeline, customPromisify, { - __proto__: null, - enumerable: true, - get() { - return promises.pipeline; + return promises ??= require('stream/promises'); }, }); @@ -136,6 +88,7 @@ ObjectDefineProperty(eos, customPromisify, { __proto__: null, enumerable: true, get() { + promises ??= require('stream/promises'); return promises.finished; }, }); diff --git a/lib/stream/promises.js b/lib/stream/promises.js index a8b65d62b096..9d36d4da4f15 100644 --- a/lib/stream/promises.js +++ b/lib/stream/promises.js @@ -11,12 +11,13 @@ const { isWebStream, } = require('internal/streams/utils'); -const { pipelineImpl: pl } = require('internal/streams/pipeline'); +let pl; const { finished } = require('internal/streams/end-of-stream'); require('stream'); function pipeline(...streams) { + pl ??= require('internal/streams/pipeline').pipelineImpl; return new Promise((resolve, reject) => { let signal; let end; diff --git a/src/node_util.cc b/src/node_util.cc index 6d02ecea1daa..d71a7bd2b298 100644 --- a/src/node_util.cc +++ b/src/node_util.cc @@ -352,6 +352,7 @@ static void IsInsideNodeModules(const FunctionCallbackInfo& args) { args.GetReturnValue().Set(result); } +template static void DefineLazyPropertiesGetter( Local name, const v8::PropertyCallbackInfo& info) { Isolate* isolate = info.GetIsolate(); @@ -381,6 +382,10 @@ static void DefineLazyPropertiesGetter( // V8 will have scheduled an error to be thrown. return; } + if constexpr (return_module) { + info.GetReturnValue().Set(require_result); + return; + } Local ret; if (!require_result.As()->Get(context, name).ToLocal(&ret)) { // V8 will have scheduled an error to be thrown. @@ -390,13 +395,15 @@ static void DefineLazyPropertiesGetter( } static void DefineLazyProperties(const FunctionCallbackInfo& args) { - // target: object, id: string, keys: string[][, enumerable = true] + // target: object, source: string | object, keys: string[], + // enumerable: boolean = true CHECK_GE(args.Length(), 3); // target: Object where to define the lazy properties. CHECK(args[0]->IsObject()); - // id: Internal module to lazy-load where the API to expose are implemented. - CHECK(args[1]->IsString()); - // keys: Keys to map from `require(id)` and `target`. + // source: Internal module ID for named exports, or a map from property names + // to module IDs for modules whose entire exports object is the lazy value. + CHECK(args[1]->IsString() || args[1]->IsObject()); + // keys: Property names to expose on `target`. CHECK(args[2]->IsArray()); // enumerable: Whether the property should be enumerable. CHECK(args.Length() == 3 || args[3]->IsBoolean()); @@ -404,7 +411,7 @@ static void DefineLazyProperties(const FunctionCallbackInfo& args) { auto context = args.GetIsolate()->GetCurrentContext(); auto target = args[0].As(); - Local id = args[1]; + Local source = args[1]; v8::PropertyAttribute attribute = args.Length() == 3 || args[3]->IsTrue() ? v8::None : v8::DontEnum; @@ -417,12 +424,19 @@ static void DefineLazyProperties(const FunctionCallbackInfo& args) { return; } CHECK(key->IsString()); + Local id = source; + auto getter = DefineLazyPropertiesGetter; + if (source->IsObject()) { + if (!source.As()->Get(context, key).ToLocal(&id)) { + // V8 will have scheduled an error to be thrown. + return; + } + CHECK(id->IsString()); + getter = DefineLazyPropertiesGetter; + } if (target - ->SetLazyDataProperty(context, - key.As(), - DefineLazyPropertiesGetter, - id, - attribute) + ->SetLazyDataProperty( + context, key.As(), getter, id, attribute) .IsNothing()) { // V8 will have scheduled an error to be thrown. return; @@ -499,7 +513,8 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) { registry->Register(ParseEnv); registry->Register(IsInsideNodeModules); registry->Register(DefineLazyProperties); - registry->Register(DefineLazyPropertiesGetter); + registry->Register(DefineLazyPropertiesGetter); + registry->Register(DefineLazyPropertiesGetter); registry->Register(ConstructSharedArrayBuffer); registry->Register(MarkPromiseAsHandled); } diff --git a/test/parallel/test-bootstrap-modules.js b/test/parallel/test-bootstrap-modules.js index e6a40d45e26f..1c5db787a8bc 100644 --- a/test/parallel/test-bootstrap-modules.js +++ b/test/parallel/test-bootstrap-modules.js @@ -146,6 +146,16 @@ if (contextFromSnapshot) { 'NativeModule internal/mime', 'NativeModule internal/modules/typescript', 'NativeModule internal/net', + 'NativeModule internal/streams/add-abort-signal', + 'NativeModule internal/streams/destroy', + 'NativeModule internal/streams/end-of-stream', + 'NativeModule internal/streams/from', + 'NativeModule internal/streams/legacy', + 'NativeModule internal/streams/readable', + 'NativeModule internal/streams/state', + 'NativeModule internal/streams/writable', + 'NativeModule stream', + 'NativeModule string_decoder', ].forEach(expected.beforePreExec.add.bind(expected.beforePreExec)); } else if (isMainThread) { expected.beforePreExec.delete(getFormatNativeModule); @@ -154,29 +164,20 @@ if (contextFromSnapshot) { if (!isMainThread) { [ 'NativeModule diagnostics_channel', - 'NativeModule internal/abort_controller', 'NativeModule internal/process/worker_thread_only', 'NativeModule internal/streams/add-abort-signal', - 'NativeModule internal/streams/compose', 'NativeModule internal/streams/destroy', - 'NativeModule internal/streams/duplex', - 'NativeModule internal/streams/duplexpair', 'NativeModule internal/streams/end-of-stream', 'NativeModule internal/streams/from', 'NativeModule internal/streams/legacy', - 'NativeModule internal/streams/operators', - 'NativeModule internal/streams/passthrough', - 'NativeModule internal/streams/pipeline', 'NativeModule internal/streams/readable', 'NativeModule internal/streams/state', - 'NativeModule internal/streams/transform', 'NativeModule internal/streams/utils', 'NativeModule internal/streams/writable', 'NativeModule internal/worker', 'NativeModule internal/worker/io', 'NativeModule internal/worker/messaging', 'NativeModule stream', - 'NativeModule stream/promises', 'NativeModule string_decoder', 'NativeModule worker_threads', ].forEach(expected.atRunTime.add.bind(expected.atRunTime)); diff --git a/test/parallel/test-worker-snapshot-streams.js b/test/parallel/test-worker-snapshot-streams.js new file mode 100644 index 000000000000..bda73a129c8c --- /dev/null +++ b/test/parallel/test-worker-snapshot-streams.js @@ -0,0 +1,41 @@ +// Flags: --expose-internals +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { Worker, isMainThread, workerData } = require('worker_threads'); +const { internalBinding } = require('internal/test/binding'); + +const contextFromSnapshot = + process.config.variables.node_use_node_snapshot && + !process.execArgv.includes('--no-node-snapshot') && + (isMainThread || !process.execArgv.includes('--no-worker-snapshot')); +const { compiledInSnapshot } = internalBinding('builtins').getCacheUsage(); +for (const id of ['stream', 'internal/streams/readable', 'internal/streams/writable']) { + assert.strictEqual(compiledInSnapshot.includes(id), Boolean(contextFromSnapshot), id); +} + +const { Readable, Writable, getDefaultHighWaterMark, setDefaultHighWaterMark } = require('stream'); + +if (isMainThread) { + const highWaterMark = getDefaultHighWaterMark(false); + // Runtime changes in the parent must not become the worker's initial state. + setDefaultHighWaterMark(false, highWaterMark + 1); + Readable.prototype.parentOnly = true; + const worker = new Worker(__filename, { workerData: { highWaterMark } }); + worker.on('error', common.mustNotCall()); + worker.on('exit', common.mustCall((code) => assert.strictEqual(code, 0))); +} else { + assert.strictEqual(Readable.prototype.parentOnly, undefined); + if (workerData?.highWaterMark !== undefined) { + assert.strictEqual(getDefaultHighWaterMark(false), workerData.highWaterMark); + } + assert(process.stdin instanceof Readable); + assert(process.stdout instanceof Writable); + assert(process.stderr instanceof Writable); + assert.strictEqual(typeof process.stdin.map, 'function'); + assert.strictEqual(typeof process.stdin.toArray, 'function'); + Readable.from([1, 2, 3]).map((value) => value * 2).toArray().then((values) => { + assert.deepStrictEqual(values, [2, 4, 6]); + }).then(common.mustCall()); +}