stream: optimize creation

Refs: nodejs/performance#79 PR-URL: #50337 Reviewed-By: Yagiz Nizipli <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Moshe Atlow <[email protected]> Reviewed-By: Vinícius Lourenço Claro Cardoso <[email protected]> Reviewed-By: Luigi Pinca <[email protected]>

Author: ronag

lib/internal/streams/duplex.js

@@ -35,9 +35,17 @@ const{
 
 module.exports=Duplex;
 
+constStream=require('internal/streams/legacy').Stream;
 constReadable=require('internal/streams/readable');
 constWritable=require('internal/streams/writable');
 
+const{
+ addAbortSignal,
+}=require('internal/streams/add-abort-signal');
+
+constdestroyImpl=require('internal/streams/destroy');
+const{ kOnConstructed }=require('internal/streams/utils');
+
 ObjectSetPrototypeOf(Duplex.prototype,Readable.prototype);
 ObjectSetPrototypeOf(Duplex,Readable);
 
@@ -55,8 +63,8 @@ function Duplex(options){
 if(!(thisinstanceofDuplex))
 returnnewDuplex(options);
 
-Readable.call(this,options);
-Writable.call(this,options);
+this._readableState=newReadable.ReadableState(options,this,true);
+this._writableState=newWritable.WritableState(options,this,true);
 
 if(options){
 this.allowHalfOpen=options.allowHalfOpen!==false;
@@ -73,9 +81,39 @@ function Duplex(options){
 this._writableState.ended=true;
 this._writableState.finished=true;
 }
+
+if(typeofoptions.read==='function')
+this._read=options.read;
+
+if(typeofoptions.write==='function')
+this._write=options.write;
+
+if(typeofoptions.writev==='function')
+this._writev=options.writev;
+
+if(typeofoptions.destroy==='function')
+this._destroy=options.destroy;
+
+if(typeofoptions.final==='function')
+this._final=options.final;
+
+if(typeofoptions.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,{

lib/internal/streams/readable.js

@@ -71,6 +71,7 @@ const{
  AbortError,
 }=require('internal/errors');
 const{ validateObject }=require('internal/validators');
+const{ kOnConstructed }=require('internal/streams/utils');
 
 constkState=Symbol('kState');
 
@@ -251,20 +252,14 @@ ObjectDefineProperties(ReadableState.prototype,{
 
 
 functionReadableState(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(typeofisDuplex!=='boolean')
-isDuplex=streaminstanceofStream.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;
@@ -310,16 +305,17 @@ function ReadableState(options, stream, isDuplex){
 }
 }
 
+ReadableState.prototype[kOnConstructed]=functiononConstructed(stream){
+if((this[kState]&kNeedReadable)!==0){
+maybeReadMore(stream,this);
+}
+};
 
 functionReadable(options){
 if(!(thisinstanceofReadable))
 returnnewReadable(options);
 
-// Checking for a Stream.Duplex instance is faster here instead of inside
-// the ReadableState constructor, at least with V8 6.5.
-constisDuplex=thisinstanceofStream.Duplex;
-
-this._readableState=newReadableState(options,this,isDuplex);
+this._readableState=newReadableState(options,this,false);
 
 if(options){
 if(typeofoptions.read==='function')
@@ -331,17 +327,17 @@ function Readable(options){
 if(typeofoptions.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;

lib/internal/streams/utils.js

@@ -4,6 +4,7 @@ const{
  SymbolAsyncIterator,
  SymbolIterator,
  SymbolFor,
+ Symbol,
 }=primordials;
 
 // We need to use SymbolFor to make these globally available
@@ -16,6 +17,8 @@ const kIsReadable = SymbolFor('nodejs.stream.readable');
 constkIsWritable=SymbolFor('nodejs.stream.writable');
 constkIsDisturbed=SymbolFor('nodejs.stream.disturbed');
 
+constkOnConstructed=Symbol('kOnConstructed');
+
 constkIsClosedPromise=SymbolFor('nodejs.webstream.isClosedPromise');
 constkControllerErrorFunction=SymbolFor('nodejs.webstream.controllerErrorFunction');
 
@@ -303,6 +306,7 @@ function isErrored(stream){
 }
 
 module.exports={
+ kOnConstructed,
  isDestroyed,
  kIsDestroyed,
  isDisturbed,

lib/internal/streams/writable.js

@@ -44,6 +44,7 @@ const EE = require('events');
 constStream=require('internal/streams/legacy').Stream;
 const{ Buffer }=require('buffer');
 constdestroyImpl=require('internal/streams/destroy');
+const{ kOnConstructed }=require('internal/streams/utils');
 
 const{
  addAbortSignal,
@@ -290,20 +291,15 @@ ObjectDefineProperties(WritableState.prototype,{
 });
 
 functionWritableState(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(typeofisDuplex!=='boolean')
-isDuplex=streaminstanceofStream.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
@@ -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.
-constdefaultEncoding=options?.defaultEncoding;
+constdefaultEncoding=options ? options.defaultEncoding : null;
 if(defaultEncoding==null||defaultEncoding==='utf8'||defaultEncoding==='utf-8'){
 this[kState]|=kDefaultUTF8Encoding;
 }elseif(Buffer.isEncoding(defaultEncoding)){
@@ -372,23 +368,21 @@ ObjectDefineProperty(WritableState.prototype, 'bufferedRequestCount',{
 },
 });
 
-functionWritable(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]=functiononConstructed(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.
-constisDuplex=(thisinstanceofStream.Duplex);
+if((this[kState]&kEnding)!==0){
+finishMaybe(stream,this);
+}
+};
 
-if(!isDuplex&&!FunctionPrototypeSymbolHasInstance(Writable,this))
+functionWritable(options){
+if(!(thisinstanceofWritable))
 returnnewWritable(options);
 
-this._writableState=newWritableState(options,this,isDuplex);
+this._writableState=newWritableState(options,this,false);
 
 if(options){
 if(typeofoptions.write==='function')
@@ -412,17 +406,11 @@ function Writable(options){
 
 Stream.call(this,options);
 
-destroyImpl.construct(this,()=>{
-conststate=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,{