events: try to improve stream creation perf

nodejs · May 12, 2024 · e787353 · e787353
1 parent b59447c
commit e787353
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 18 deletions.
diff --git a/lib/internal/events/event_emitter.js b/lib/internal/events/event_emitter.js
@@ -32,6 +32,7 @@ const {
   kMaxEventTargetListeners,
   kMaxEventTargetListenersWarned,
   kImpl,
+  kPreInitiated,
   kIsFastPath,
   kSwitchToSlowPath,
   kInitialEvents,
@@ -222,6 +223,13 @@ ObjectDefineProperties(EventEmitter.prototype, {
     configurable: false,
     writable: true,
   },
+  [kPreInitiated]: {
+    __proto__: null,
+    value: false,
+    enumerable: false,
+    configurable: false,
+    writable: true,
+  },
   [kIsFastPath]: {
     __proto__: null,
     value: undefined,
@@ -240,7 +248,10 @@ ObjectDefineProperties(EventEmitter.prototype, {
     __proto__: null,
     enumerable: true,
     get: function() {
-      return this[kImpl]?._events;
+      if(this[kImpl] === undefined) {
+        return undefined;
+      }
+      return this[kImpl]._events;
     },
     set: function(arg) {
       // If using the _events setter move to slow path to avoid bugs with incorrect shape or functions
@@ -347,6 +358,23 @@ EventEmitter.setMaxListeners =
 EventEmitter.init = function(opts) {
   let thisPrototype;
 
+  this._maxListeners = this._maxListeners || undefined;
+
+
+  if (opts?.captureRejections) {
+    validateBoolean(opts.captureRejections, 'options.captureRejections');
+    this[kCapture] = Boolean(opts.captureRejections);
+  } else {
+    // Assigning the kCapture property directly saves an expensive
+    // prototype lookup in a very sensitive hot path.
+    this[kCapture] = EventEmitter.prototype[kCapture];
+  }
+
+  if(this[kPreInitiated] === true) {
+    this[kPreInitiated] = false;
+    return;
+  }
+
   if (this[kImpl] === undefined || (thisPrototype = ObjectGetPrototypeOf(this))[kImpl] === this[kImpl]) {
     this[kImpl] = new FastEventEmitter(this);
     this[kIsFastPath] = true;
@@ -368,18 +396,6 @@ EventEmitter.init = function(opts) {
     impl._eventsCount = 0;
     impl[kShapeMode] = false;
   }
-
-  this._maxListeners = this._maxListeners || undefined;
-
-
-  if (opts?.captureRejections) {
-    validateBoolean(opts.captureRejections, 'options.captureRejections');
-    this[kCapture] = Boolean(opts.captureRejections);
-  } else {
-    // Assigning the kCapture property directly saves an expensive
-    // prototype lookup in a very sensitive hot path.
-    this[kCapture] = EventEmitter.prototype[kCapture];
-  }
 };
 
 EventEmitter.prototype[kSwitchToSlowPath] = function() {

diff --git a/lib/internal/events/symbols.js b/lib/internal/events/symbols.js
@@ -17,6 +17,7 @@ const kMaxEventTargetListenersWarned =
 const kWatermarkData = SymbolFor('nodejs.watermarkData');
 
 const kImpl = Symbol('kImpl');
+const kPreInitiated = Symbol('kPreInitiated');
 const kIsFastPath = Symbol('kIsFastPath');
 const kSwitchToSlowPath = Symbol('kSwitchToSlowPath');
 const kRejection = SymbolFor('nodejs.rejection');
@@ -31,6 +32,7 @@ module.exports = {
   kMaxEventTargetListenersWarned,
   kWatermarkData,
   kImpl,
+  kPreInitiated,
   kIsFastPath,
   kSwitchToSlowPath,
   kRejection,

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
@@ -43,7 +43,7 @@ Readable.ReadableState = ReadableState;
 
 const EE = require('events');
 const { Stream, prependListener } = require('internal/streams/legacy');
-const { kInitialEvents } = require('internal/events/symbols');
+const { kInitialEvents, kImpl, kShapeMode, kIsFastPath, kPreInitiated} = require('internal/events/symbols');
 const { Buffer } = require('buffer');
 
 const {
@@ -92,6 +92,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
+const {FastEventEmitter} = require("internal/events/fast_event_emitter");
 
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
@@ -320,9 +321,9 @@ function Readable(options) {
   if (!(this instanceof Readable))
     return new Readable(options);
 
-  const thisEvents = this._events;
-  if (thisEvents === undefined || thisEvents === null) {
-    this[kInitialEvents] ??= {
+  if (this[kImpl] === undefined && this._events === undefined) {
+    this[kIsFastPath] = true;
+    this[kImpl] = new FastEventEmitter(this, this[kInitialEvents] !== undefined ? this[kInitialEvents] :{
       close: undefined,
       error: undefined,
       data: undefined,
@@ -335,7 +336,10 @@ function Readable(options) {
       // unpipe: undefined,
       // [destroyImpl.kConstruct]: undefined,
       // [destroyImpl.kDestroy]: undefined,
-    };
+    });
+    this[kImpl][kShapeMode] = true;
+  } else {
+    this[kPreInitiated] = false;
   }
 
   this._readableState = new ReadableState(options, this, false);
@@ -363,6 +367,7 @@ function Readable(options) {
   }
 }
 
+Readable.prototype[kPreInitiated] = true;
 Readable.prototype.destroy = destroyImpl.destroy;
 Readable.prototype._undestroy = destroyImpl.undestroy;
 Readable.prototype._destroy = function(err, cb) {