lib: performance improvement on readline async iterator

Using a direct approach to create the readline async iterator
allowed an iteration over 20 to 58% faster.

**BREAKING CHANGE**: With that change, the async iteterator
obtained from the readline interface doesn't have the
property "stream" any longer. This happened because it's no
longer created through a Readable, instead, the async
iterator is created directly from the events of the readline
interface instance, so, if anyone is using that property,
this change will break their code.
Also, the Readable added a backpressure control that is
fairly compensated by the use of FixedQueue + monitoring
its size. This control wasn't really precise with readline
before, though, because it only pauses the reading of the
original stream, but the lines generated from the last
message received from it was still emitted. For example:
if the readable was paused at 1000 messages but the last one
received generated 10k lines, but no further messages were
emitted again until the queue was lower than the readable
highWaterMark. A similar  behavior still happens with the
new implementation, but the highWaterMark used is fixed: 1024,
and the original stream is resumed again only after the queue
is cleared.

Before making that change, I created a package implementing
the same concept used here to validate it. You can find it
[here](https://github.com/Farenheith/faster-readline-iterator)
if this helps anyhow.

PR-URL: https://github.com/nodejs/node/pull/41276
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

1 files changed, 113 insertions, 53 deletions

diff --git a/lib/events.js b/lib/events.js
index 7abf18f42c..e05a3bc6b8 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -23,7 +23,8 @@
 
 const {
   ArrayPrototypeJoin,
-  ArrayPrototypeShift,
+  ArrayPrototypePop,
+  ArrayPrototypePush,
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
@@ -33,6 +34,7 @@ const {
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   NumberIsNaN,
+  NumberMAX_SAFE_INTEGER,
   ObjectCreate,
   ObjectDefineProperty,
   ObjectDefineProperties,
@@ -59,6 +61,8 @@ const {
 } = require('internal/util/inspect');
 
 let spliceOne;
+let FixedQueue;
+let kFirstEventParam;
 
 const {
   AbortError,
@@ -73,6 +77,7 @@ const {
 } = require('internal/errors');
 
 const {
+  validateInteger,
   validateAbortSignal,
   validateBoolean,
   validateFunction,
@@ -84,6 +89,7 @@ const kErrorMonitor = Symbol('events.errorMonitor');
 const kMaxEventTargetListeners = Symbol('events.maxEventTargetListeners');
 const kMaxEventTargetListenersWarned =
   Symbol('events.maxEventTargetListenersWarned');
+const kWatermarkData = SymbolFor('nodejs.watermarkData');
 
 let EventEmitterAsyncResource;
 // The EventEmitterAsyncResource has to be initialized lazily because event.js
@@ -999,25 +1005,44 @@ function eventTargetAgnosticAddListener(emitter, name, listener, flags) {
  * Returns an `AsyncIterator` that iterates `event` events.
  * @param {EventEmitter} emitter
  * @param {string | symbol} event
- * @param {{ signal: AbortSignal; }} [options]
+ * @param {{
+ *    signal: AbortSignal;
+ *    close?: string[];
+ *    highWatermark?: number,
+ *    lowWatermark?: number
+ *   }} [options]
  * @returns {AsyncIterator}
  */
-function on(emitter, event, options) {
-  const signal = options?.signal;
+function on(emitter, event, options = {}) {
+  // Parameters validation
+  const signal = options.signal;
   validateAbortSignal(signal, 'options.signal');
   if (signal?.aborted)
     throw new AbortError(undefined, { cause: signal?.reason });
-
-  const unconsumedEvents = [];
-  const unconsumedPromises = [];
+  const highWatermark = options.highWatermark ?? NumberMAX_SAFE_INTEGER;
+  validateInteger(highWatermark, 'options.highWatermark', 1);
+  const lowWatermark = options.lowWatermark ?? 1;
+  validateInteger(lowWatermark, 'options.lowWatermark', 1);
+
+  // Preparing controlling queues and variables
+  FixedQueue ??= require('internal/fixed_queue');
+  const unconsumedEvents = new FixedQueue();
+  const unconsumedPromises = new FixedQueue();
+  let paused = false;
   let error = null;
   let finished = false;
+  let size = 0;
 
   const iterator = ObjectSetPrototypeOf({
     next() {
       // First, we consume all unread events
-      const value = unconsumedEvents.shift();
-      if (value) {
+      if (size) {
+        const value = unconsumedEvents.shift();
+        size--;
+        if (paused && size < lowWatermark) {
+          emitter.resume();
+          paused = false;
+        }
         return PromiseResolve(createIterResult(value, false));
       }
 
@@ -1032,9 +1057,7 @@ function on(emitter, event, options) {
       }
 
       // If the iterator is finished, resolve to done
-      if (finished) {
-        return PromiseResolve(createIterResult(undefined, true));
-      }
+      if (finished) return closeHandler();
 
       // Wait until an event happens
       return new Promise(function(resolve, reject) {
@@ -1043,24 +1066,7 @@ function on(emitter, event, options) {
     },
 
     return() {
-      eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
-      eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
-
-      if (signal) {
-        eventTargetAgnosticRemoveListener(
-          signal,
-          'abort',
-          abortListener,
-          { once: true });
-      }
-
-      finished = true;
-
-      for (const promise of unconsumedPromises) {
-        promise.resolve(createIterResult(undefined, true));
-      }
-
-      return PromiseResolve(createIterResult(undefined, true));
+      return closeHandler();
     },
 
     throw(err) {
@@ -1068,21 +1074,54 @@ function on(emitter, event, options) {
         throw new ERR_INVALID_ARG_TYPE('EventEmitter.AsyncIterator',
                                        'Error', err);
       }
-      error = err;
-      eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
-      eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
+      errorHandler(err);
     },
-
     [SymbolAsyncIterator]() {
       return this;
-    }
+    },
+    [kWatermarkData]: {
+      /**
+       * The current queue size
+       */
+      get size() {
+        return size;
+      },
+      /**
+       * The low watermark. The emitter is resumed every time size is lower than it
+       */
+      get low() {
+        return lowWatermark;
+      },
+      /**
+       * The high watermark. The emitter is paused every time size is higher than it
+       */
+      get high() {
+        return highWatermark;
+      },
+      /**
+       * It checks wether the emitter is paused by the watermark controller or not
+       */
+      get isPaused() {
+        return paused;
+      }
+    },
   }, AsyncIteratorPrototype);
 
-  eventTargetAgnosticAddListener(emitter, event, eventHandler);
+  // Adding event handlers
+  const { addEventListener, removeAll } = listenersController();
+  kFirstEventParam ??= require('internal/events/symbols').kFirstEventParam;
+  addEventListener(emitter, event, options[kFirstEventParam] ? eventHandler : function(...args) {
+    return eventHandler(args);
+  });
   if (event !== 'error' && typeof emitter.on === 'function') {
-    emitter.on('error', errorHandler);
+    addEventListener(emitter, 'error', errorHandler);
+  }
+  const closeEvents = options?.close;
+  if (closeEvents?.length) {
+    for (let i = 0; i < closeEvents.length; i++) {
+      addEventListener(emitter, closeEvents[i], closeHandler);
+    }
   }
-
   if (signal) {
     eventTargetAgnosticAddListener(
       signal,
@@ -1097,27 +1136,48 @@ function on(emitter, event, options) {
     errorHandler(new AbortError(undefined, { cause: signal?.reason }));
   }
 
-  function eventHandler(...args) {
-    const promise = ArrayPrototypeShift(unconsumedPromises);
-    if (promise) {
-      promise.resolve(createIterResult(args, false));
-    } else {
-      unconsumedEvents.push(args);
-    }
+  function eventHandler(value) {
+    if (unconsumedPromises.isEmpty()) {
+      size++;
+      if (!paused && size > highWatermark) {
+        paused = true;
+        emitter.pause();
+      }
+      unconsumedEvents.push(value);
+    } else unconsumedPromises.shift().resolve(createIterResult(value, false));
   }
 
   function errorHandler(err) {
-    finished = true;
+    if (unconsumedPromises.isEmpty()) error = err;
+    else unconsumedPromises.shift().reject(err);
 
-    const toError = ArrayPrototypeShift(unconsumedPromises);
+    closeHandler();
+  }
 
-    if (toError) {
-      toError.reject(err);
-    } else {
-      // The next time we call next()
-      error = err;
+  function closeHandler() {
+    removeAll();
+    finished = true;
+    const doneResult = createIterResult(undefined, true);
+    while (!unconsumedPromises.isEmpty()) {
+      unconsumedPromises.shift().resolve(doneResult);
     }
 
-    iterator.return();
+    return PromiseResolve(doneResult);
   }
 }
+
+function listenersController() {
+  const listeners = [];
+
+  return {
+    addEventListener(emitter, event, handler, flags) {
+      eventTargetAgnosticAddListener(emitter, event, handler, flags);
+      ArrayPrototypePush(listeners, [emitter, event, handler, flags]);
+    },
+    removeAll() {
+      while (listeners.length > 0) {
+        ReflectApply(eventTargetAgnosticRemoveListener, undefined, ArrayPrototypePop(listeners));
+      }
+    }
+  };
+}