Skip to content

Commit

Permalink
stream: optimize creation
Browse files Browse the repository at this point in the history 
Refs: nodejs/performance#79

PR-URL: #50337
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: Vinícius Lourenço Claro Cardoso <contact@viniciusl.com.br>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
  • Loading branch information
@ronag @targos
ronag authored and targos committed Nov 11, 2023
1 parent e2c3b01 commit e433fa5
Show file tree
Hide file tree
Showing 4 changed files with 83 additions and 57 deletions.
42 changes: 40 additions & 2 deletions lib/internal/streams/duplex.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,17 @@ const {

module.exports = Duplex;

const Stream = require('internal/streams/legacy').Stream;
const Readable = require('internal/streams/readable');
const Writable = require('internal/streams/writable');

const {
addAbortSignal,
} = require('internal/streams/add-abort-signal');

const destroyImpl = require('internal/streams/destroy');
const { kOnConstructed } = require('internal/streams/utils');

ObjectSetPrototypeOf(Duplex.prototype, Readable.prototype);
ObjectSetPrototypeOf(Duplex, Readable);

Expand All @@ -55,8 +63,8 @@ function Duplex(options) {
if (!(this instanceof Duplex))
return new Duplex(options);

Readable.call(this, options);
Writable.call(this, options);
this._readableState = new Readable.ReadableState(options, this, true);
this._writableState = new Writable.WritableState(options, this, true);

if (options) {
this.allowHalfOpen = options.allowHalfOpen !== false;
Expand All @@ -73,9 +81,39 @@ function Duplex(options) {
this._writableState.ended = true;
this._writableState.finished = true;
}

if (typeof options.read === 'function')
this._read = options.read;

if (typeof options.write === 'function')
this._write = options.write;

if (typeof options.writev === 'function')
this._writev = options.writev;

if (typeof options.destroy === 'function')
this._destroy = options.destroy;

if (typeof options.final === 'function')
this._final = options.final;

if (typeof options.construct === 'function')
this._construct = options.construct;

if (options.signal)
addAbortSignal(options.signal, this);
} else {
this.allowHalfOpen = true;
}

Stream.call(this, options);

if (this._construct != null) {
destroyImpl.construct(this, () => {
this._readableState[kOnConstructed](this);
this._writableState[kOnConstructed](this);
});
}
}

ObjectDefineProperties(Duplex.prototype, {
Expand Down
36 changes: 16 additions & 20 deletions lib/internal/streams/readable.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ const {
AbortError,
} = require('internal/errors');
const { validateObject } = require('internal/validators');
const { kOnConstructed } = require('internal/streams/utils');

const kState = Symbol('kState');

Expand Down Expand Up @@ -251,20 +252,14 @@ ObjectDefineProperties(ReadableState.prototype, {


function ReadableState(options, stream, isDuplex) {
// Duplex streams are both readable and writable, but share
// the same options object.
// However, some cases require setting options to different
// values for the readable and the writable sides of the duplex stream.
// These options can be provided separately as readableXXX and writableXXX.
if (typeof isDuplex !== 'boolean')
isDuplex = stream instanceof Stream.Duplex;

// Bit map field to store ReadableState more effciently with 1 bit per field
// instead of a V8 slot per field.
this[kState] = kEmitClose | kAutoDestroy | kConstructed | kSync;

// Object stream flag. Used to make read(n) ignore n and to
// make all the buffer merging and length checks go away.
if (options && options.objectMode) this[kState] |= kObjectMode;
if (options && options.objectMode)
this[kState] |= kObjectMode;

if (isDuplex && options && options.readableObjectMode)
this[kState] |= kObjectMode;
Expand Down Expand Up @@ -310,16 +305,17 @@ function ReadableState(options, stream, isDuplex) {
}
}

ReadableState.prototype[kOnConstructed] = function onConstructed(stream) {
if ((this[kState] & kNeedReadable) !== 0) {
maybeReadMore(stream, this);
}
};

function Readable(options) {
if (!(this instanceof Readable))
return new Readable(options);

// Checking for a Stream.Duplex instance is faster here instead of inside
// the ReadableState constructor, at least with V8 6.5.
const isDuplex = this instanceof Stream.Duplex;

this._readableState = new ReadableState(options, this, isDuplex);
this._readableState = new ReadableState(options, this, false);

if (options) {
if (typeof options.read === 'function')
Expand All @@ -331,17 +327,17 @@ function Readable(options) {
if (typeof options.construct === 'function')
this._construct = options.construct;

if (options.signal && !isDuplex)
if (options.signal)
addAbortSignal(options.signal, this);
}

Stream.call(this, options);

destroyImpl.construct(this, () => {
if (this._readableState.needReadable) {
maybeReadMore(this, this._readableState);
}
});
if (this._construct != null) {
destroyImpl.construct(this, () => {
this._readableState[kOnConstructed](this);
});
}
}

Readable.prototype.destroy = destroyImpl.destroy;
Expand Down
4 changes: 4 additions & 0 deletions lib/internal/streams/utils.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ const {
SymbolAsyncIterator,
SymbolIterator,
SymbolFor,
Symbol,
} = primordials;

// We need to use SymbolFor to make these globally available
Expand All @@ -16,6 +17,8 @@ const kIsReadable = SymbolFor('nodejs.stream.readable');
const kIsWritable = SymbolFor('nodejs.stream.writable');
const kIsDisturbed = SymbolFor('nodejs.stream.disturbed');

const kOnConstructed = Symbol('kOnConstructed');

const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction');

Expand Down Expand Up @@ -303,6 +306,7 @@ function isErrored(stream) {
}

module.exports = {
kOnConstructed,
isDestroyed,
kIsDestroyed,
isDisturbed,
Expand Down
58 changes: 23 additions & 35 deletions lib/internal/streams/writable.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ const EE = require('events');
const Stream = require('internal/streams/legacy').Stream;
const { Buffer } = require('buffer');
const destroyImpl = require('internal/streams/destroy');
const { kOnConstructed } = require('internal/streams/utils');

const {
addAbortSignal,
Expand Down Expand Up @@ -290,20 +291,15 @@ ObjectDefineProperties(WritableState.prototype, {
});

function WritableState(options, stream, isDuplex) {
// Duplex streams are both readable and writable, but share
// the same options object.
// However, some cases require setting options to different
// values for the readable and the writable sides of the duplex stream,
// e.g. options.readableObjectMode vs. options.writableObjectMode, etc.
if (typeof isDuplex !== 'boolean')
isDuplex = stream instanceof Stream.Duplex;

// Bit map field to store WritableState more effciently with 1 bit per field
// instead of a V8 slot per field.
this[kState] = kSync | kConstructed | kEmitClose | kAutoDestroy;

if (options && options.objectMode) this[kState] |= kObjectMode;
if (isDuplex && options && options.writableObjectMode) this[kState] |= kObjectMode;
if (options && options.objectMode)
this[kState] |= kObjectMode;

if (isDuplex && options && options.writableObjectMode)
this[kState] |= kObjectMode;

// The point at which write() starts returning false
// Note: 0 is a valid value, means that we always return false if
Expand All @@ -323,7 +319,7 @@ function WritableState(options, stream, isDuplex) {
// Crypto is kind of old and crusty. Historically, its default string
// encoding is 'binary' so we have to make this configurable.
// Everything else in the universe uses 'utf8', though.
const defaultEncoding = options?.defaultEncoding;
const defaultEncoding = options ? options.defaultEncoding : null;
if (defaultEncoding == null || defaultEncoding === 'utf8' || defaultEncoding === 'utf-8') {
this[kState] |= kDefaultUTF8Encoding;
} else if (Buffer.isEncoding(defaultEncoding)) {
Expand Down Expand Up @@ -372,23 +368,21 @@ ObjectDefineProperty(WritableState.prototype, 'bufferedRequestCount', {
},
});

function Writable(options) {
// Writable ctor is applied to Duplexes, too.
// `realHasInstance` is necessary because using plain `instanceof`
// would return false, as no `_writableState` property is attached.

// Trying to use the custom `instanceof` for Writable here will also break the
// Node.js LazyTransform implementation, which has a non-trivial getter for
// `_writableState` that would lead to infinite recursion.
WritableState.prototype[kOnConstructed] = function onConstructed(stream) {
if ((this[kState] & kWriting) === 0) {
clearBuffer(stream, this);
}

// Checking for a Stream.Duplex instance is faster here instead of inside
// the WritableState constructor, at least with V8 6.5.
const isDuplex = (this instanceof Stream.Duplex);
if ((this[kState] & kEnding) !== 0) {
finishMaybe(stream, this);
}
};

if (!isDuplex && !FunctionPrototypeSymbolHasInstance(Writable, this))
function Writable(options) {
if (!(this instanceof Writable))
return new Writable(options);

this._writableState = new WritableState(options, this, isDuplex);
this._writableState = new WritableState(options, this, false);

if (options) {
if (typeof options.write === 'function')
Expand All @@ -412,17 +406,11 @@ function Writable(options) {

Stream.call(this, options);

destroyImpl.construct(this, () => {
const state = this._writableState;

if ((state[kState] & kWriting) === 0) {
clearBuffer(this, state);
}

if ((state[kState] & kEnding) !== 0) {
finishMaybe(this, state);
}
});
if (this._construct != null) {
destroyImpl.construct(this, () => {
this._writableState[kOnConstructed](this);
});
}
}

ObjectDefineProperty(Writable, SymbolHasInstance, {
Expand Down

0 comments on commit e433fa5

Please sign in to comment.