From c467ffe876b9335946b373b0fe4f076c77ae596f Mon Sep 17 00:00:00 2001
From: Thiago Santos <thiago.santos@maxmilhas.com.br>
Date: Mon, 20 Dec 2021 15:04:04 -0300
Subject: [PATCH 01/25] 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.
---
 lib/events.js                                 |  64 +++-------
 lib/internal/linkedMap.js                     | 110 +++++++++++++++++
 .../readline/eventsToAsyncIteratorFactory.js  | 107 ++++++++++++++++
 lib/internal/readline/interface.js            |  46 ++-----
 ...t-readline-async-iterators-backpressure.js |  23 ++--
 .../parallel/test-readline-async-iterators.js | 115 +++++++++++++++++-
 6 files changed, 372 insertions(+), 93 deletions(-)
 create mode 100644 lib/internal/linkedMap.js
 create mode 100644 lib/internal/readline/eventsToAsyncIteratorFactory.js

diff --git a/lib/events.js b/lib/events.js
index 7abf18f42c0064..b39d5629cacb30 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -27,6 +27,7 @@ const {
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
+  ArrayFrom,
   Boolean,
   Error,
   ErrorCaptureStackTrace,
@@ -71,6 +72,7 @@ const {
   },
   genericNodeError,
 } = require('internal/errors');
+const getLinkedMap = require('internal/linkedMap');
 
 const {
   validateAbortSignal,
@@ -509,30 +511,19 @@ EventEmitter.prototype.emit = function emit(type, ...args) {
   if (handler === undefined)
     return false;
 
-  if (typeof handler === 'function') {
-    const result = handler.apply(this, args);
+  const listeners = ArrayFrom(handler);
+  const len = handler.length;
+  for (let i = 0; i < len; ++i) {
+    const result = listeners[i].apply(this, args);
 
     // We check if result is undefined first because that
     // is the most common case so we do not pay any perf
-    // penalty
+    // penalty.
+    // This code is duplicated because extracting it away
+    // would make it non-inlineable.
     if (result !== undefined && result !== null) {
       addCatch(this, result, type, args);
     }
-  } else {
-    const len = handler.length;
-    const listeners = arrayClone(handler);
-    for (let i = 0; i < len; ++i) {
-      const result = listeners[i].apply(this, args);
-
-      // We check if result is undefined first because that
-      // is the most common case so we do not pay any perf
-      // penalty.
-      // This code is duplicated because extracting it away
-      // would make it non-inlineable.
-      if (result !== undefined && result !== null) {
-        addCatch(this, result, type, args);
-      }
-    }
   }
 
   return true;
@@ -565,19 +556,13 @@ function _addListener(target, type, listener, prepend) {
 
   if (existing === undefined) {
     // Optimize the case of one listener. Don't need the extra array object.
-    events[type] = listener;
+    existing = events[type] = getLinkedMap().push(listener);
     ++target._eventsCount;
+  } else if (prepend) {
+    existing.unshift(listener);
   } else {
-    if (typeof existing === 'function') {
-      // Adding the second element, need to change to array.
-      existing = events[type] =
-        prepend ? [listener, existing] : [existing, listener];
-      // If we've already got an array, just append.
-    } else if (prepend) {
-      existing.unshift(listener);
-    } else {
-      existing.push(listener);
-    }
+    existing.push(listener);
+  }
 
     // Check for listener leak
     m = _getMaxListeners(target);
@@ -682,11 +667,8 @@ EventEmitter.prototype.removeListener =
       const list = events[type];
       if (list === undefined)
         return this;
-
-      if (list === listener || list.listener === listener) {
-        if (--this._eventsCount === 0)
-          this._events = ObjectCreate(null);
-        else {
+      if (list?.remove(listener)) {
+        if (list.length === 0) {
           delete events[type];
           if (events.removeListener)
             this.emit('removeListener', type, list.listener || listener);
@@ -838,19 +820,7 @@ EventEmitter.prototype.listenerCount = listenerCount;
  * @returns {number}
  */
 function listenerCount(type) {
-  const events = this._events;
-
-  if (events !== undefined) {
-    const evlistener = events[type];
-
-    if (typeof evlistener === 'function') {
-      return 1;
-    } else if (evlistener !== undefined) {
-      return evlistener.length;
-    }
-  }
-
-  return 0;
+  return this._events?.[type]?.length || 0;
 }
 
 /**
diff --git a/lib/internal/linkedMap.js b/lib/internal/linkedMap.js
new file mode 100644
index 00000000000000..14e9f699578336
--- /dev/null
+++ b/lib/internal/linkedMap.js
@@ -0,0 +1,110 @@
+const {
+  Map,
+  SymbolIterator,
+} = primordials;
+
+function push(root, value) {
+  const node = { value };
+  if (root.last) {
+    node.previous = root.last;
+    root.last.next = root.last = node;
+  } else {
+    root.last = root.first = node;
+  }
+  root.length++;
+
+  return node;
+}
+
+function unshift(value) {
+  const node = { value };
+  if (root.first) {
+    node.next = root.first;
+    root.first.previous = root.first = node;
+  } else {
+    root.last = root.first = node;
+  }
+  root.length++;
+
+  return node;
+}
+
+function shift(root) {
+  if (root.first) {
+    const { value } = root.first;
+    root.first = root.first.next;
+    root.length--;
+    return value;
+  }
+}
+
+
+function getLinkedMap() {
+  const map = new Map();
+  function addToMap(key, node, operation) {
+    let refs = map.get(key);
+    if (!refs) {
+      map.set(key, refs = { length: 0 });
+    }
+      operation(refs, node);
+  }
+  const root = { length: 0 };
+
+  return {
+    get length() {
+      return root.length;
+    },
+    push(value) {
+      const node = push(root, value);
+      addToMap(value, node, push);
+
+      return this;
+    },
+    unshift(value) {
+      const node = unshift(root, value);
+      addToMap(key, node, unshift);
+
+      return this;
+    },
+    remove(value) {
+      const refs = map.get(value);
+      if (refs) {
+        const result = shift(refs);
+        if (result.previous)
+          result.previous.next = result.next;
+        if (result.next)
+          result.next.previous = result.previous;
+        if (refs.length === 0) {
+          map.delete(value);
+        }
+        root.length--;
+        return 1;
+      }
+      return 0;
+    },
+    [SymbolIterator]() {
+      let node = root.first;
+
+      const iterator = {
+        next() {
+          if (!node) {
+            return { done: true };
+          }
+          const result = {
+            done: false,
+            value: node.value,
+          };
+          node = node.next;
+          return result;
+        },
+        [SymbolIterator]() {
+          return iterator;
+        }
+      };
+
+      return iterator;
+    }
+  }
+}
+
+module.exports = getLinkedMap;
diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
new file mode 100644
index 00000000000000..ec165d27ac4744
--- /dev/null
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -0,0 +1,107 @@
+'use strict';
+const {
+  Promise,
+  SymbolAsyncIterator,
+  ArrayPrototypeConcat,
+} = primordials;
+const FixedQueue = require('internal/fixed_queue');
+
+const PAUSE_THRESHOLD = 1024;
+const RESUME_THRESHOLD = 1;
+const ITEM_EVENTS = ['data'];
+const CLOSE_EVENTS = ['close', 'end'];
+const ERROR_EVENTS = ['error'];
+
+
+function waitNext(emitter, next, events) {
+  return new Promise((resolve, reject) => {
+    const resolveNext = () => {
+      for (let i = 0; i < events.length; i++)
+        emitter.off(events[i], resolveNext);
+      try {
+        resolve(next());
+      } catch (promiseError) {
+        reject(promiseError);
+      }
+    };
+    for (let i = 0; i < events.length; i++)
+      emitter.once(events[i], resolveNext);
+  });
+}
+
+module.exports = function eventsToAsyncIteratorFactory(readable, {
+  pauseThreshold = PAUSE_THRESHOLD,
+  resumeThreshold = RESUME_THRESHOLD,
+  closeEvents = CLOSE_EVENTS,
+  itemEvents = ITEM_EVENTS,
+  errorEvents = ERROR_EVENTS,
+}) {
+  const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents);
+  const highWaterMark = RESUME_THRESHOLD;
+
+  const queue = new FixedQueue();
+  let done = false;
+  let error;
+  let queueSize = 0;
+  let paused = false;
+  const onError = (value) => {
+    turn('off');
+    error = value;
+  };
+  const onClose = () => {
+    turn('off');
+    done = true;
+  };
+  const onItem = (value) => {
+    queue.push(value);
+    queueSize++;
+    if (queueSize >= pauseThreshold) {
+      paused = true;
+      readable.pause();
+    }
+  };
+  function turn(onOff) {
+    for (let i = 0; i < closeEvents.length; i++)
+      readable[onOff](closeEvents[i], onClose);
+    for (let i = 0; i < itemEvents.length; i++)
+      readable[onOff](itemEvents[i], onItem);
+    for (let i = 0; i < itemEvents.length; i++)
+      readable[onOff](errorEvents[i], onError);
+  }
+
+  turn('on');
+
+  function next() {
+    if (!queue.isEmpty()) {
+      const value = queue.shift();
+      queueSize--;
+      if (queueSize < resumeThreshold) {
+        paused = false;
+        readable.resume();
+      }
+      return {
+        done: false,
+        value,
+      };
+    }
+    if (error) {
+      throw error;
+    }
+    if (done) {
+      return { done };
+    }
+    return waitNext(readable, next, events);
+  }
+
+  return {
+    next,
+    highWaterMark,
+    get isPaused() {
+      return paused;
+    },
+    get queueSize() {
+      return queueSize;
+    },
+    [SymbolAsyncIterator]() { return result; },
+  };
+};
diff --git a/lib/internal/readline/interface.js b/lib/internal/readline/interface.js
index 694f4462ceea7a..6f842a3c7ca51b 100644
--- a/lib/internal/readline/interface.js
+++ b/lib/internal/readline/interface.js
@@ -70,9 +70,6 @@ const {
 
 const { StringDecoder } = require('string_decoder');
 
-// Lazy load Readable for startup performance.
-let Readable;
-
 const kHistorySize = 30;
 const kMaxUndoRedoStackSize = 2048;
 const kMincrlfDelay = 100;
@@ -1345,41 +1342,16 @@ class Interface extends InterfaceConstructor {
    * @returns {InterfaceAsyncIterator}
    */
   [SymbolAsyncIterator]() {
-    if (this[kLineObjectStream] === undefined) {
-      if (Readable === undefined) {
-        Readable = require('stream').Readable;
-      }
-      const readable = new Readable({
-        objectMode: true,
-        read: () => {
-          this.resume();
-        },
-        destroy: (err, cb) => {
-          this.off('line', lineListener);
-          this.off('close', closeListener);
-          this.close();
-          cb(err);
-        },
-      });
-      const lineListener = (input) => {
-        if (!readable.push(input)) {
-          // TODO(rexagod): drain to resume flow
-          this.pause();
-        }
-      };
-      const closeListener = () => {
-        readable.push(null);
-      };
-      const errorListener = (err) => {
-        readable.destroy(err);
-      };
-      this.on('error', errorListener);
-      this.on('line', lineListener);
-      this.on('close', closeListener);
-      this[kLineObjectStream] = readable;
+    if (!this[kLineObjectStream]) {
+      this[kLineObjectStream] = require(
+        'internal/readline/eventsToAsyncIteratorFactory'
+      )(
+        this, {
+          itemEvents: ['line'],
+          closeEvents: ['close']
+        });
     }
-
-    return this[kLineObjectStream][SymbolAsyncIterator]();
+    return this[kLineObjectStream];
   }
 }
 
diff --git a/test/parallel/test-readline-async-iterators-backpressure.js b/test/parallel/test-readline-async-iterators-backpressure.js
index 2ca124dde5b890..fddaeb368593e4 100644
--- a/test/parallel/test-readline-async-iterators-backpressure.js
+++ b/test/parallel/test-readline-async-iterators-backpressure.js
@@ -6,11 +6,17 @@ const { Readable } = require('stream');
 const readline = require('readline');
 
 const CONTENT = 'content';
-const TOTAL_LINES = 18;
+const LINES_PER_PUSH = 2051;
+const REPETITIONS = 3;
 
 (async () => {
   const readable = new Readable({ read() {} });
-  readable.push(`${CONTENT}\n`.repeat(TOTAL_LINES));
+  let salt = 0;
+  for (let i = 0; i < REPETITIONS; i++) {
+    readable.push(`${CONTENT}\n`.repeat(LINES_PER_PUSH + i));
+    salt += i;
+  }
+  const TOTAL_LINES = LINES_PER_PUSH * REPETITIONS + salt;
 
   const rli = readline.createInterface({
     input: readable,
@@ -18,7 +24,7 @@ const TOTAL_LINES = 18;
   });
 
   const it = rli[Symbol.asyncIterator]();
-  const highWaterMark = it.stream.readableHighWaterMark;
+  const highWaterMark = it.highWaterMark;
 
   // For this test to work, we have to queue up more than the number of
   // highWaterMark items in rli. Make sure that is the case.
@@ -26,13 +32,15 @@ const TOTAL_LINES = 18;
 
   let iterations = 0;
   let readableEnded = false;
+  let notPaused = 0;
   for await (const line of it) {
     assert.strictEqual(readableEnded, false);
-
     assert.strictEqual(line, CONTENT);
-
-    const expectedPaused = TOTAL_LINES - iterations > highWaterMark;
-    assert.strictEqual(readable.isPaused(), expectedPaused);
+    assert.ok(it.queueSize <= TOTAL_LINES);
+    assert.strictEqual(readable.isPaused(), it.queueSize >= 1);
+    if (!readable.isPaused()) {
+      notPaused++;
+    }
 
     iterations += 1;
 
@@ -45,4 +53,5 @@ const TOTAL_LINES = 18;
   }
 
   assert.strictEqual(iterations, TOTAL_LINES);
+  assert.strictEqual(notPaused, REPETITIONS);
 })().then(common.mustCall());
diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
index 2aa557a3363486..71d36ca41d49f9 100644
--- a/test/parallel/test-readline-async-iterators.js
+++ b/test/parallel/test-readline-async-iterators.js
@@ -4,6 +4,7 @@ const common = require('../common');
 const fs = require('fs');
 const { join } = require('path');
 const readline = require('readline');
+const { Readable } = require('stream');
 const assert = require('assert');
 
 const tmpdir = require('../common/tmpdir');
@@ -63,7 +64,6 @@ async function testMutual() {
       // This outer loop should only iterate once.
       assert.strictEqual(iterated, false);
       iterated = true;
-
       iteratedLines.push(k);
       for await (const l of rli) {
         iteratedLines.push(l);
@@ -74,4 +74,115 @@ async function testMutual() {
   }
 }
 
-testSimple().then(testMutual).then(common.mustCall());
+async function testPerformance() {
+  const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
+Dui accumsan sit amet nulla facilisi morbi tempus iaculis urna.
+Eget dolor morbi non arcu risus quis varius quam quisque.
+Lacus viverra vitae congue eu consequat ac felis donec.
+Amet porttitor eget dolor morbi non arcu.
+Velit ut tortor pretium viverra suspendisse.
+Mauris nunc congue nisi vitae suscipit tellus.
+Amet nisl suscipit adipiscing bibendum est ultricies integer.
+Sit amet dictum sit amet justo donec enim diam.
+Condimentum mattis pellentesque id nibh tortor id aliquet lectus proin.
+Diam in arcu cursus euismod quis viverra nibh.
+`;
+
+  const REPETITIONS = 10000;
+  const SAMPLE = 100;
+  const THRESHOLD = 81;
+
+  function getLoremIpsumStream() {
+    const readable = Readable({
+      objectMode: true,
+    });
+    let i = 0;
+    readable._read = () => readable.push(
+      i++ >= REPETITIONS ? null : loremIpsum
+    );
+    return readable;
+  }
+
+  function oldWay() {
+    const readable = new Readable({
+      objectMode: true,
+      read: () => {
+        this.resume();
+      },
+      destroy: (err, cb) => {
+        this.off('line', lineListener);
+        this.off('close', closeListener);
+        this.close();
+        cb(err);
+      },
+    });
+    const lineListener = (input) => {
+      if (!readable.push(input)) {
+        // TODO(rexagod): drain to resume flow
+        this.pause();
+      }
+    };
+    const closeListener = () => {
+      readable.push(null);
+    };
+    const errorListener = (err) => {
+      readable.destroy(err);
+    };
+    this.on('error', errorListener);
+    this.on('line', lineListener);
+    this.on('close', closeListener);
+    return readable[Symbol.asyncIterator]();
+  }
+
+  function getAvg(mean, x, n) {
+    return (mean * n + x) / (n + 1);
+  }
+
+  let totalTimeOldWay = 0;
+  let totalTimeNewWay = 0;
+  let totalCharsOldWay = 0;
+  let totalCharsNewWay = 0;
+  const linesOldWay = [];
+  const linesNewWay = [];
+
+  for (let time = 0; time < SAMPLE; time++) {
+    const rlOldWay = readline.createInterface({
+      input: getLoremIpsumStream(),
+    });
+    let currenttotalTimeOldWay = Date.now();
+    for await (const line of oldWay.call(rlOldWay)) {
+      totalCharsOldWay += line.length;
+      if (time === 0) {
+        linesOldWay.push(line);
+      }
+    }
+    currenttotalTimeOldWay = Date.now() - currenttotalTimeOldWay;
+    totalTimeOldWay = getAvg(totalTimeOldWay, currenttotalTimeOldWay, SAMPLE);
+
+    const rlNewWay = readline.createInterface({
+      input: getLoremIpsumStream(),
+    });
+    let currentTotalTimeNewWay = Date.now();
+    for await (const line of rlNewWay) {
+      totalCharsNewWay += line.length;
+      if (time === 0) {
+        linesNewWay.push(line);
+      }
+    }
+    currentTotalTimeNewWay = Date.now() - currentTotalTimeNewWay;
+    totalTimeNewWay = getAvg(totalTimeNewWay, currentTotalTimeNewWay, SAMPLE);
+  }
+
+  assert.strictEqual(totalCharsOldWay, totalCharsNewWay);
+  assert.strictEqual(linesOldWay.length, linesNewWay.length);
+  linesOldWay.forEach((line, index) =>
+    assert.strictEqual(line, linesNewWay[index])
+  );
+  const percentage = totalTimeNewWay * 100 / totalTimeOldWay;
+  assert.ok(percentage <= THRESHOLD, `Failed: ${totalTimeNewWay} isn't lesser than ${THRESHOLD}% of ${totalTimeOldWay}. Actual percentage: ${percentage.toFixed(2)}%`);
+}
+
+testSimple()
+  .then(testMutual)
+  .then(testPerformance)
+  .then(common.mustCall());

From 05db8ffb2e5bbcf0a48aee9e394e83852b3ca897 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 22 Dec 2021 16:52:24 -0300
Subject: [PATCH 02/25] lib: undoing accidental changes

Some Changes I just sent wasn't related to the PR in question,
so I'm undoing then (in events.js and the addition of linkedMap.js)
Also, fixed the [SymbolAsyncIterator] function returned by
eventsToAsyncIteratorFactory, that was referencing a no longer
available variable
---
 lib/events.js                                 |  63 +++++++---
 lib/internal/linkedMap.js                     | 110 ------------------
 .../readline/eventsToAsyncIteratorFactory.js  |   2 +-
 3 files changed, 48 insertions(+), 127 deletions(-)
 delete mode 100644 lib/internal/linkedMap.js

diff --git a/lib/events.js b/lib/events.js
index b39d5629cacb30..11f369c52cebd3 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -72,7 +72,6 @@ const {
   },
   genericNodeError,
 } = require('internal/errors');
-const getLinkedMap = require('internal/linkedMap');
 
 const {
   validateAbortSignal,
@@ -511,19 +510,30 @@ EventEmitter.prototype.emit = function emit(type, ...args) {
   if (handler === undefined)
     return false;
 
-  const listeners = ArrayFrom(handler);
-  const len = handler.length;
-  for (let i = 0; i < len; ++i) {
-    const result = listeners[i].apply(this, args);
+  if (typeof handler === 'function') {
+    const result = handler.apply(this, args);
 
     // We check if result is undefined first because that
     // is the most common case so we do not pay any perf
-    // penalty.
-    // This code is duplicated because extracting it away
-    // would make it non-inlineable.
+    // penalty
     if (result !== undefined && result !== null) {
       addCatch(this, result, type, args);
     }
+  } else {
+    const len = handler.length;
+    const listeners = arrayClone(handler);
+    for (let i = 0; i < len; ++i) {
+      const result = listeners[i].apply(this, args);
+
+      // We check if result is undefined first because that
+      // is the most common case so we do not pay any perf
+      // penalty.
+      // This code is duplicated because extracting it away
+      // would make it non-inlineable.
+      if (result !== undefined && result !== null) {
+        addCatch(this, result, type, args);
+      }
+    }
   }
 
   return true;
@@ -556,13 +566,19 @@ function _addListener(target, type, listener, prepend) {
 
   if (existing === undefined) {
     // Optimize the case of one listener. Don't need the extra array object.
-    existing = events[type] = getLinkedMap().push(listener);
+    events[type] = listener;
     ++target._eventsCount;
-  } else if (prepend) {
-    existing.unshift(listener);
   } else {
-    existing.push(listener);
-  }
+    if (typeof existing === 'function') {
+      // Adding the second element, need to change to array.
+      existing = events[type] =
+        prepend ? [listener, existing] : [existing, listener];
+      // If we've already got an array, just append.
+    } else if (prepend) {
+      existing.unshift(listener);
+    } else {
+      existing.push(listener);
+    }
 
     // Check for listener leak
     m = _getMaxListeners(target);
@@ -667,8 +683,11 @@ EventEmitter.prototype.removeListener =
       const list = events[type];
       if (list === undefined)
         return this;
-      if (list?.remove(listener)) {
-        if (list.length === 0) {
+
+      if (list === listener || list.listener === listener) {
+        if (--this._eventsCount === 0)
+          this._events = ObjectCreate(null);
+        else {
           delete events[type];
           if (events.removeListener)
             this.emit('removeListener', type, list.listener || listener);
@@ -820,7 +839,19 @@ EventEmitter.prototype.listenerCount = listenerCount;
  * @returns {number}
  */
 function listenerCount(type) {
-  return this._events?.[type]?.length || 0;
+  const events = this._events;
+
+  if (events !== undefined) {
+    const evlistener = events[type];
+
+    if (typeof evlistener === 'function') {
+      return 1;
+    } else if (evlistener !== undefined) {
+      return evlistener.length;
+    }
+  }
+
+  return 0;
 }
 
 /**
diff --git a/lib/internal/linkedMap.js b/lib/internal/linkedMap.js
deleted file mode 100644
index 14e9f699578336..00000000000000
--- a/lib/internal/linkedMap.js
+++ /dev/null
@@ -1,110 +0,0 @@
-const {
-  Map,
-  SymbolIterator,
-} = primordials;
-
-function push(root, value) {
-  const node = { value };
-  if (root.last) {
-    node.previous = root.last;
-    root.last.next = root.last = node;
-  } else {
-    root.last = root.first = node;
-  }
-  root.length++;
-
-  return node;
-}
-
-function unshift(value) {
-  const node = { value };
-  if (root.first) {
-    node.next = root.first;
-    root.first.previous = root.first = node;
-  } else {
-    root.last = root.first = node;
-  }
-  root.length++;
-
-  return node;
-}
-
-function shift(root) {
-  if (root.first) {
-    const { value } = root.first;
-    root.first = root.first.next;
-    root.length--;
-    return value;
-  }
-}
-
-
-function getLinkedMap() {
-  const map = new Map();
-  function addToMap(key, node, operation) {
-    let refs = map.get(key);
-    if (!refs) {
-      map.set(key, refs = { length: 0 });
-    }
-      operation(refs, node);
-  }
-  const root = { length: 0 };
-
-  return {
-    get length() {
-      return root.length;
-    },
-    push(value) {
-      const node = push(root, value);
-      addToMap(value, node, push);
-
-      return this;
-    },
-    unshift(value) {
-      const node = unshift(root, value);
-      addToMap(key, node, unshift);
-
-      return this;
-    },
-    remove(value) {
-      const refs = map.get(value);
-      if (refs) {
-        const result = shift(refs);
-        if (result.previous)
-          result.previous.next = result.next;
-        if (result.next)
-          result.next.previous = result.previous;
-        if (refs.length === 0) {
-          map.delete(value);
-        }
-        root.length--;
-        return 1;
-      }
-      return 0;
-    },
-    [SymbolIterator]() {
-      let node = root.first;
-
-      const iterator = {
-        next() {
-          if (!node) {
-            return { done: true };
-          }
-          const result = {
-            done: false,
-            value: node.value,
-          };
-          node = node.next;
-          return result;
-        },
-        [SymbolIterator]() {
-          return iterator;
-        }
-      };
-
-      return iterator;
-    }
-  }
-}
-
-module.exports = getLinkedMap;
diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index ec165d27ac4744..ec245e36e70400 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -102,6 +102,6 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     get queueSize() {
       return queueSize;
     },
-    [SymbolAsyncIterator]() { return result; },
+    [SymbolAsyncIterator]() { return this; },
   };
 };

From 3ecee3cdeaa1437b6d42e835c0bc04cd0928a80b Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Thu, 23 Dec 2021 15:39:22 -0300
Subject: [PATCH 03/25] lib: changing once to on in waitNext

After seeing how once method works on EventEmitter,
it was clear that it doesn't make sense to use it
into the waitNext method, as the callback
will already remove the events
---
 lib/internal/readline/eventsToAsyncIteratorFactory.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index ec245e36e70400..bc97f92d213d9f 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -25,7 +25,7 @@ function waitNext(emitter, next, events) {
       }
     };
     for (let i = 0; i < events.length; i++)
-      emitter.once(events[i], resolveNext);
+      emitter.on(events[i], resolveNext);
   });
 }
 

From 83919d2e88583723941ff021058f8fbd19ea9adc Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Tue, 28 Dec 2021 22:44:39 -0300
Subject: [PATCH 04/25] lib: adding benchmarks for readline iterable

---
 benchmark/readline/readline-iterable.js  | 46 ++++++++++++++++++++++++
 test/benchmark/test-bechmark-readline.js |  7 ++++
 2 files changed, 53 insertions(+)
 create mode 100644 benchmark/readline/readline-iterable.js
 create mode 100644 test/benchmark/test-bechmark-readline.js

diff --git a/benchmark/readline/readline-iterable.js b/benchmark/readline/readline-iterable.js
new file mode 100644
index 00000000000000..dad243c20ab779
--- /dev/null
+++ b/benchmark/readline/readline-iterable.js
@@ -0,0 +1,46 @@
+'use strict';
+const common = require('../common.js');
+const readline = require('readline');
+const { Readable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  n: [1e6],
+});
+
+const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
+Dui accumsan sit amet nulla facilisi morbi tempus iaculis urna.
+Eget dolor morbi non arcu risus quis varius quam quisque.
+Lacus viverra vitae congue eu consequat ac felis donec.
+Amet porttitor eget dolor morbi non arcu.
+Velit ut tortor pretium viverra suspendisse.
+Mauris nunc congue nisi vitae suscipit tellus.
+Amet nisl suscipit adipiscing bibendum est ultricies integer.
+Sit amet dictum sit amet justo donec enim diam.
+Condimentum mattis pellentesque id nibh tortor id aliquet lectus proin.
+Diam in arcu cursus euismod quis viverra nibh.
+Rest of line`;
+
+function getLoremIpsumStream(repetitions) {
+  const readable = Readable({
+    objectMode: true,
+  });
+  let i = 0;
+  readable._read = () => readable.push(
+    i++ >= repetitions ? null : loremIpsum
+  );
+  return readable;
+}
+
+async function main({ n }) {
+  bench.start();
+  let lineCount = 0;
+
+  const iterable = readline.createInterface({
+    input: getLoremIpsumStream(n),
+  });
+
+  for await (const line of iterable) {
+    lineCount++;
+  }
+  bench.end(n);
+}
diff --git a/test/benchmark/test-bechmark-readline.js b/test/benchmark/test-bechmark-readline.js
new file mode 100644
index 00000000000000..eaa9fd08ed487e
--- /dev/null
+++ b/test/benchmark/test-bechmark-readline.js
@@ -0,0 +1,7 @@
+'use strict';
+
+require('../common');
+
+const runBenchmark = require('../common/benchmark');
+
+runBenchmark('readline', { NODEJS_BENCHMARK_ZERO_ALLOWED: 1 });

From 03ac26d5cb8d0aa9df95df48f3b5c3d654107562 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Tue, 28 Dec 2021 22:56:49 -0300
Subject: [PATCH 05/25] lib: adding different lines multiplier

Adding different lines multipliers is good for this bench
because we can see how well the change performs in
different scales. That way we can find out if there some
curve with some point where the benefits starts or ends
---
 benchmark/readline/readline-iterable.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/benchmark/readline/readline-iterable.js b/benchmark/readline/readline-iterable.js
index dad243c20ab779..664cde2079049b 100644
--- a/benchmark/readline/readline-iterable.js
+++ b/benchmark/readline/readline-iterable.js
@@ -4,7 +4,7 @@ const readline = require('readline');
 const { Readable } = require('stream');
 
 const bench = common.createBenchmark(main, {
-  n: [1e6],
+  n: [1e1, 1e2, 1e3, 1e4, 1e5, 1e6],
 });
 
 const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.

From 870b5cf6d085ccb8a1ce4e807947855af355b2d5 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Tue, 28 Dec 2021 23:11:28 -0300
Subject: [PATCH 06/25] lib: fixing lint on benchmark

---
 benchmark/readline/readline-iterable.js | 4 ++--
 lib/events.js                           | 1 -
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/benchmark/readline/readline-iterable.js b/benchmark/readline/readline-iterable.js
index 664cde2079049b..5c35987d997b96 100644
--- a/benchmark/readline/readline-iterable.js
+++ b/benchmark/readline/readline-iterable.js
@@ -39,8 +39,8 @@ async function main({ n }) {
     input: getLoremIpsumStream(n),
   });
 
-  for await (const line of iterable) {
+  for await (const {} of iterable) {
     lineCount++;
   }
-  bench.end(n);
+  bench.end(lineCount);
 }
diff --git a/lib/events.js b/lib/events.js
index 11f369c52cebd3..7abf18f42c0064 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -27,7 +27,6 @@ const {
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
-  ArrayFrom,
   Boolean,
   Error,
   ErrorCaptureStackTrace,

From ae3325c269ca1322051b49e99e959107a2ce0361 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 29 Dec 2021 12:58:00 -0300
Subject: [PATCH 07/25] lib: adding test to validate slow stream

---
 .../parallel/test-readline-async-iterators.js | 41 +++++++++++++++++++
 1 file changed, 41 insertions(+)

diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
index 71d36ca41d49f9..3bbcb810acae5f 100644
--- a/test/parallel/test-readline-async-iterators.js
+++ b/test/parallel/test-readline-async-iterators.js
@@ -182,7 +182,48 @@ Diam in arcu cursus euismod quis viverra nibh.
   assert.ok(percentage <= THRESHOLD, `Failed: ${totalTimeNewWay} isn't lesser than ${THRESHOLD}% of ${totalTimeOldWay}. Actual percentage: ${percentage.toFixed(2)}%`);
 }
 
+async function testSlowStreamForLeaks() {
+  const message = 'a\nb\nc\n';
+  const DELAY = 1;
+  const REPETITIONS = 100;
+  const warningCallback = common.mustNotCall();
+  process.on('warning', warningCallback);
+
+  function getStream() {
+    const readable = Readable({
+      objectMode: true,
+    });
+    readable._read = () => {};
+    let i = REPETITIONS;
+    function schedule() {
+      setTimeout(() => {
+        i--;
+        if (i < 0) {
+          readable.push(null);
+        } else {
+          readable.push(message);
+          schedule();
+        }
+      }, DELAY);
+    }
+    schedule();
+    return readable;
+  }
+  const iterable = readline.createInterface({
+    input: getStream(),
+  });
+
+  let lines = 0;
+  for await (const {} of iterable) {
+    lines++;
+  }
+
+  assert.strictEqual(lines, 3 * REPETITIONS);
+  process.off('warning', warningCallback);
+}
+
 testSimple()
   .then(testMutual)
   .then(testPerformance)
+  .then(testSlowStreamForLeaks)
   .then(common.mustCall());

From bb6f6ab280084bf8d24342ae4fd55f0060f9e15a Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Thu, 30 Dec 2021 13:29:07 -0300
Subject: [PATCH 08/25] lib: adding return and throw implementation to
 eventsToAsyncIteratorFactory

---
 .../readline/eventsToAsyncIteratorFactory.js  | 68 +++++++++++++------
 1 file changed, 47 insertions(+), 21 deletions(-)

diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index bc97f92d213d9f..59f10fce3708b3 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -1,9 +1,15 @@
 'use strict';
 const {
+  Error,
   Promise,
   SymbolAsyncIterator,
   ArrayPrototypeConcat,
 } = primordials;
+const {
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+  },
+} = require('internal/errors');
 const FixedQueue = require('internal/fixed_queue');
 
 const PAUSE_THRESHOLD = 1024;
@@ -12,23 +18,6 @@ const ITEM_EVENTS = ['data'];
 const CLOSE_EVENTS = ['close', 'end'];
 const ERROR_EVENTS = ['error'];
 
-
-function waitNext(emitter, next, events) {
-  return new Promise((resolve, reject) => {
-    const resolveNext = () => {
-      for (let i = 0; i < events.length; i++)
-        emitter.off(events[i], resolveNext);
-      try {
-        resolve(next());
-      } catch (promiseError) {
-        reject(promiseError);
-      }
-    };
-    for (let i = 0; i < events.length; i++)
-      emitter.on(events[i], resolveNext);
-  });
-}
-
 module.exports = function eventsToAsyncIteratorFactory(readable, {
   pauseThreshold = PAUSE_THRESHOLD,
   resumeThreshold = RESUME_THRESHOLD,
@@ -38,6 +27,17 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
 }) {
   const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents);
   const highWaterMark = RESUME_THRESHOLD;
+  let onFunction;
+  let offFunction;
+  if (typeof readable.on === 'function') {
+    onFunction = 'on';
+    offFunction = 'off';
+  } else if (typeof readable.addEventListener === 'function') {
+    onFunction = 'addEventListener';
+    offFunction = 'removeEventListener';
+  } else {
+    throw new ERR_INVALID_ARG_TYPE('readable', 'Readable', readable);
+  }
 
   const queue = new FixedQueue();
   let done = false;
@@ -45,11 +45,11 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
   let queueSize = 0;
   let paused = false;
   const onError = (value) => {
-    turn('off');
+    turn(offFunction);
     error = value;
   };
   const onClose = () => {
-    turn('off');
+    turn(offFunction);
     done = true;
   };
   const onItem = (value) => {
@@ -69,7 +69,7 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
       readable[onOff](errorEvents[i], onError);
   }
 
-  turn('on');
+  turn(onFunction);
 
   function next() {
     if (!queue.isEmpty()) {
@@ -90,7 +90,19 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     if (done) {
       return { done };
     }
-    return waitNext(readable, next, events);
+    return new Promise((resolve, reject) => {
+      const resolveNext = () => {
+        for (let i = 0; i < events.length; i++)
+          readable[offFunction](events[i], resolveNext);
+        try {
+          resolve(next());
+        } catch (promiseError) {
+          reject(promiseError);
+        }
+      };
+      for (let i = 0; i < events.length; i++)
+        readable[onFunction](events[i], resolveNext);
+    });
   }
 
   return {
@@ -102,6 +114,20 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     get queueSize() {
       return queueSize;
     },
+    return() {
+      turn(offFunction);
+      done = true;
+
+      return { done };
+    },
+    throw(err) {
+      if (!err || !(err instanceof Error)) {
+        throw new ERR_INVALID_ARG_TYPE('err',
+                                       'Error', err);
+      }
+      error = err;
+      turn(offFunction);
+    },
     [SymbolAsyncIterator]() { return this; },
   };
 };

From 221e23984bb13e861cb1ba103671d03a1f80b945 Mon Sep 17 00:00:00 2001
From: Thiago Santos <thiago.santos@maxmilhas.com.br>
Date: Mon, 20 Dec 2021 15:04:04 -0300
Subject: [PATCH 09/25] 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.
---
 lib/events.js                                 |  64 +++-------
 lib/internal/linkedMap.js                     | 110 ++++++++++++++++++
 .../readline/eventsToAsyncIteratorFactory.js  |  70 ++++-------
 .../parallel/test-readline-async-iterators.js |  42 +------
 4 files changed, 150 insertions(+), 136 deletions(-)
 create mode 100644 lib/internal/linkedMap.js

diff --git a/lib/events.js b/lib/events.js
index 7abf18f42c0064..b39d5629cacb30 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -27,6 +27,7 @@ const {
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
+  ArrayFrom,
   Boolean,
   Error,
   ErrorCaptureStackTrace,
@@ -71,6 +72,7 @@ const {
   },
   genericNodeError,
 } = require('internal/errors');
+const getLinkedMap = require('internal/linkedMap');
 
 const {
   validateAbortSignal,
@@ -509,30 +511,19 @@ EventEmitter.prototype.emit = function emit(type, ...args) {
   if (handler === undefined)
     return false;
 
-  if (typeof handler === 'function') {
-    const result = handler.apply(this, args);
+  const listeners = ArrayFrom(handler);
+  const len = handler.length;
+  for (let i = 0; i < len; ++i) {
+    const result = listeners[i].apply(this, args);
 
     // We check if result is undefined first because that
     // is the most common case so we do not pay any perf
-    // penalty
+    // penalty.
+    // This code is duplicated because extracting it away
+    // would make it non-inlineable.
     if (result !== undefined && result !== null) {
       addCatch(this, result, type, args);
     }
-  } else {
-    const len = handler.length;
-    const listeners = arrayClone(handler);
-    for (let i = 0; i < len; ++i) {
-      const result = listeners[i].apply(this, args);
-
-      // We check if result is undefined first because that
-      // is the most common case so we do not pay any perf
-      // penalty.
-      // This code is duplicated because extracting it away
-      // would make it non-inlineable.
-      if (result !== undefined && result !== null) {
-        addCatch(this, result, type, args);
-      }
-    }
   }
 
   return true;
@@ -565,19 +556,13 @@ function _addListener(target, type, listener, prepend) {
 
   if (existing === undefined) {
     // Optimize the case of one listener. Don't need the extra array object.
-    events[type] = listener;
+    existing = events[type] = getLinkedMap().push(listener);
     ++target._eventsCount;
+  } else if (prepend) {
+    existing.unshift(listener);
   } else {
-    if (typeof existing === 'function') {
-      // Adding the second element, need to change to array.
-      existing = events[type] =
-        prepend ? [listener, existing] : [existing, listener];
-      // If we've already got an array, just append.
-    } else if (prepend) {
-      existing.unshift(listener);
-    } else {
-      existing.push(listener);
-    }
+    existing.push(listener);
+  }
 
     // Check for listener leak
     m = _getMaxListeners(target);
@@ -682,11 +667,8 @@ EventEmitter.prototype.removeListener =
       const list = events[type];
       if (list === undefined)
         return this;
-
-      if (list === listener || list.listener === listener) {
-        if (--this._eventsCount === 0)
-          this._events = ObjectCreate(null);
-        else {
+      if (list?.remove(listener)) {
+        if (list.length === 0) {
           delete events[type];
           if (events.removeListener)
             this.emit('removeListener', type, list.listener || listener);
@@ -838,19 +820,7 @@ EventEmitter.prototype.listenerCount = listenerCount;
  * @returns {number}
  */
 function listenerCount(type) {
-  const events = this._events;
-
-  if (events !== undefined) {
-    const evlistener = events[type];
-
-    if (typeof evlistener === 'function') {
-      return 1;
-    } else if (evlistener !== undefined) {
-      return evlistener.length;
-    }
-  }
-
-  return 0;
+  return this._events?.[type]?.length || 0;
 }
 
 /**
diff --git a/lib/internal/linkedMap.js b/lib/internal/linkedMap.js
new file mode 100644
index 00000000000000..14e9f699578336
--- /dev/null
+++ b/lib/internal/linkedMap.js
@@ -0,0 +1,110 @@
+const {
+  Map,
+  SymbolIterator,
+} = primordials;
+
+function push(root, value) {
+  const node = { value };
+  if (root.last) {
+    node.previous = root.last;
+    root.last.next = root.last = node;
+  } else {
+    root.last = root.first = node;
+  }
+  root.length++;
+
+  return node;
+}
+
+function unshift(value) {
+  const node = { value };
+  if (root.first) {
+    node.next = root.first;
+    root.first.previous = root.first = node;
+  } else {
+    root.last = root.first = node;
+  }
+  root.length++;
+
+  return node;
+}
+
+function shift(root) {
+  if (root.first) {
+    const { value } = root.first;
+    root.first = root.first.next;
+    root.length--;
+    return value;
+  }
+}
+
+
+function getLinkedMap() {
+  const map = new Map();
+  function addToMap(key, node, operation) {
+    let refs = map.get(key);
+    if (!refs) {
+      map.set(key, refs = { length: 0 });
+    }
+      operation(refs, node);
+  }
+  const root = { length: 0 };
+
+  return {
+    get length() {
+      return root.length;
+    },
+    push(value) {
+      const node = push(root, value);
+      addToMap(value, node, push);
+
+      return this;
+    },
+    unshift(value) {
+      const node = unshift(root, value);
+      addToMap(key, node, unshift);
+
+      return this;
+    },
+    remove(value) {
+      const refs = map.get(value);
+      if (refs) {
+        const result = shift(refs);
+        if (result.previous)
+          result.previous.next = result.next;
+        if (result.next)
+          result.next.previous = result.previous;
+        if (refs.length === 0) {
+          map.delete(value);
+        }
+        root.length--;
+        return 1;
+      }
+      return 0;
+    },
+    [SymbolIterator]() {
+      let node = root.first;
+
+      const iterator = {
+        next() {
+          if (!node) {
+            return { done: true };
+          }
+          const result = {
+            done: false,
+            value: node.value,
+          };
+          node = node.next;
+          return result;
+        },
+        [SymbolIterator]() {
+          return iterator;
+        }
+      };
+
+      return iterator;
+    }
+  }
+}
+
+module.exports = getLinkedMap;
diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index 59f10fce3708b3..ec165d27ac4744 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -1,15 +1,9 @@
 'use strict';
 const {
-  Error,
   Promise,
   SymbolAsyncIterator,
   ArrayPrototypeConcat,
 } = primordials;
-const {
-  codes: {
-    ERR_INVALID_ARG_TYPE,
-  },
-} = require('internal/errors');
 const FixedQueue = require('internal/fixed_queue');
 
 const PAUSE_THRESHOLD = 1024;
@@ -18,6 +12,23 @@ const ITEM_EVENTS = ['data'];
 const CLOSE_EVENTS = ['close', 'end'];
 const ERROR_EVENTS = ['error'];
 
+
+function waitNext(emitter, next, events) {
+  return new Promise((resolve, reject) => {
+    const resolveNext = () => {
+      for (let i = 0; i < events.length; i++)
+        emitter.off(events[i], resolveNext);
+      try {
+        resolve(next());
+      } catch (promiseError) {
+        reject(promiseError);
+      }
+    };
+    for (let i = 0; i < events.length; i++)
+      emitter.once(events[i], resolveNext);
+  });
+}
+
 module.exports = function eventsToAsyncIteratorFactory(readable, {
   pauseThreshold = PAUSE_THRESHOLD,
   resumeThreshold = RESUME_THRESHOLD,
@@ -27,17 +38,6 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
 }) {
   const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents);
   const highWaterMark = RESUME_THRESHOLD;
-  let onFunction;
-  let offFunction;
-  if (typeof readable.on === 'function') {
-    onFunction = 'on';
-    offFunction = 'off';
-  } else if (typeof readable.addEventListener === 'function') {
-    onFunction = 'addEventListener';
-    offFunction = 'removeEventListener';
-  } else {
-    throw new ERR_INVALID_ARG_TYPE('readable', 'Readable', readable);
-  }
 
   const queue = new FixedQueue();
   let done = false;
@@ -45,11 +45,11 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
   let queueSize = 0;
   let paused = false;
   const onError = (value) => {
-    turn(offFunction);
+    turn('off');
     error = value;
   };
   const onClose = () => {
-    turn(offFunction);
+    turn('off');
     done = true;
   };
   const onItem = (value) => {
@@ -69,7 +69,7 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
       readable[onOff](errorEvents[i], onError);
   }
 
-  turn(onFunction);
+  turn('on');
 
   function next() {
     if (!queue.isEmpty()) {
@@ -90,19 +90,7 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     if (done) {
       return { done };
     }
-    return new Promise((resolve, reject) => {
-      const resolveNext = () => {
-        for (let i = 0; i < events.length; i++)
-          readable[offFunction](events[i], resolveNext);
-        try {
-          resolve(next());
-        } catch (promiseError) {
-          reject(promiseError);
-        }
-      };
-      for (let i = 0; i < events.length; i++)
-        readable[onFunction](events[i], resolveNext);
-    });
+    return waitNext(readable, next, events);
   }
 
   return {
@@ -114,20 +102,6 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     get queueSize() {
       return queueSize;
     },
-    return() {
-      turn(offFunction);
-      done = true;
-
-      return { done };
-    },
-    throw(err) {
-      if (!err || !(err instanceof Error)) {
-        throw new ERR_INVALID_ARG_TYPE('err',
-                                       'Error', err);
-      }
-      error = err;
-      turn(offFunction);
-    },
-    [SymbolAsyncIterator]() { return this; },
+    [SymbolAsyncIterator]() { return result; },
   };
 };
diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
index 3bbcb810acae5f..21bea1f5d594cf 100644
--- a/test/parallel/test-readline-async-iterators.js
+++ b/test/parallel/test-readline-async-iterators.js
@@ -5,6 +5,7 @@ const fs = require('fs');
 const { join } = require('path');
 const readline = require('readline');
 const { Readable } = require('stream');
+const { Readable } = require('stream');
 const assert = require('assert');
 
 const tmpdir = require('../common/tmpdir');
@@ -182,48 +183,7 @@ Diam in arcu cursus euismod quis viverra nibh.
   assert.ok(percentage <= THRESHOLD, `Failed: ${totalTimeNewWay} isn't lesser than ${THRESHOLD}% of ${totalTimeOldWay}. Actual percentage: ${percentage.toFixed(2)}%`);
 }
 
-async function testSlowStreamForLeaks() {
-  const message = 'a\nb\nc\n';
-  const DELAY = 1;
-  const REPETITIONS = 100;
-  const warningCallback = common.mustNotCall();
-  process.on('warning', warningCallback);
-
-  function getStream() {
-    const readable = Readable({
-      objectMode: true,
-    });
-    readable._read = () => {};
-    let i = REPETITIONS;
-    function schedule() {
-      setTimeout(() => {
-        i--;
-        if (i < 0) {
-          readable.push(null);
-        } else {
-          readable.push(message);
-          schedule();
-        }
-      }, DELAY);
-    }
-    schedule();
-    return readable;
-  }
-  const iterable = readline.createInterface({
-    input: getStream(),
-  });
-
-  let lines = 0;
-  for await (const {} of iterable) {
-    lines++;
-  }
-
-  assert.strictEqual(lines, 3 * REPETITIONS);
-  process.off('warning', warningCallback);
-}
-
 testSimple()
   .then(testMutual)
   .then(testPerformance)
-  .then(testSlowStreamForLeaks)
   .then(common.mustCall());

From b5361443a31028c3ec2534ece72ec40a20f90c69 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 22 Dec 2021 16:52:24 -0300
Subject: [PATCH 10/25] lib: undoing accidental changes

Some Changes I just sent wasn't related to the PR in question,
so I'm undoing then (in events.js and the addition of linkedMap.js)
Also, fixed the [SymbolAsyncIterator] function returned by
eventsToAsyncIteratorFactory, that was referencing a no longer
available variable
---
 lib/events.js                                 |  63 +++++++---
 lib/internal/linkedMap.js                     | 110 ------------------
 .../readline/eventsToAsyncIteratorFactory.js  |   2 +-
 3 files changed, 48 insertions(+), 127 deletions(-)
 delete mode 100644 lib/internal/linkedMap.js

diff --git a/lib/events.js b/lib/events.js
index b39d5629cacb30..11f369c52cebd3 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -72,7 +72,6 @@ const {
   },
   genericNodeError,
 } = require('internal/errors');
-const getLinkedMap = require('internal/linkedMap');
 
 const {
   validateAbortSignal,
@@ -511,19 +510,30 @@ EventEmitter.prototype.emit = function emit(type, ...args) {
   if (handler === undefined)
     return false;
 
-  const listeners = ArrayFrom(handler);
-  const len = handler.length;
-  for (let i = 0; i < len; ++i) {
-    const result = listeners[i].apply(this, args);
+  if (typeof handler === 'function') {
+    const result = handler.apply(this, args);
 
     // We check if result is undefined first because that
     // is the most common case so we do not pay any perf
-    // penalty.
-    // This code is duplicated because extracting it away
-    // would make it non-inlineable.
+    // penalty
     if (result !== undefined && result !== null) {
       addCatch(this, result, type, args);
     }
+  } else {
+    const len = handler.length;
+    const listeners = arrayClone(handler);
+    for (let i = 0; i < len; ++i) {
+      const result = listeners[i].apply(this, args);
+
+      // We check if result is undefined first because that
+      // is the most common case so we do not pay any perf
+      // penalty.
+      // This code is duplicated because extracting it away
+      // would make it non-inlineable.
+      if (result !== undefined && result !== null) {
+        addCatch(this, result, type, args);
+      }
+    }
   }
 
   return true;
@@ -556,13 +566,19 @@ function _addListener(target, type, listener, prepend) {
 
   if (existing === undefined) {
     // Optimize the case of one listener. Don't need the extra array object.
-    existing = events[type] = getLinkedMap().push(listener);
+    events[type] = listener;
     ++target._eventsCount;
-  } else if (prepend) {
-    existing.unshift(listener);
   } else {
-    existing.push(listener);
-  }
+    if (typeof existing === 'function') {
+      // Adding the second element, need to change to array.
+      existing = events[type] =
+        prepend ? [listener, existing] : [existing, listener];
+      // If we've already got an array, just append.
+    } else if (prepend) {
+      existing.unshift(listener);
+    } else {
+      existing.push(listener);
+    }
 
     // Check for listener leak
     m = _getMaxListeners(target);
@@ -667,8 +683,11 @@ EventEmitter.prototype.removeListener =
       const list = events[type];
       if (list === undefined)
         return this;
-      if (list?.remove(listener)) {
-        if (list.length === 0) {
+
+      if (list === listener || list.listener === listener) {
+        if (--this._eventsCount === 0)
+          this._events = ObjectCreate(null);
+        else {
           delete events[type];
           if (events.removeListener)
             this.emit('removeListener', type, list.listener || listener);
@@ -820,7 +839,19 @@ EventEmitter.prototype.listenerCount = listenerCount;
  * @returns {number}
  */
 function listenerCount(type) {
-  return this._events?.[type]?.length || 0;
+  const events = this._events;
+
+  if (events !== undefined) {
+    const evlistener = events[type];
+
+    if (typeof evlistener === 'function') {
+      return 1;
+    } else if (evlistener !== undefined) {
+      return evlistener.length;
+    }
+  }
+
+  return 0;
 }
 
 /**
diff --git a/lib/internal/linkedMap.js b/lib/internal/linkedMap.js
deleted file mode 100644
index 14e9f699578336..00000000000000
--- a/lib/internal/linkedMap.js
+++ /dev/null
@@ -1,110 +0,0 @@
-const {
-  Map,
-  SymbolIterator,
-} = primordials;
-
-function push(root, value) {
-  const node = { value };
-  if (root.last) {
-    node.previous = root.last;
-    root.last.next = root.last = node;
-  } else {
-    root.last = root.first = node;
-  }
-  root.length++;
-
-  return node;
-}
-
-function unshift(value) {
-  const node = { value };
-  if (root.first) {
-    node.next = root.first;
-    root.first.previous = root.first = node;
-  } else {
-    root.last = root.first = node;
-  }
-  root.length++;
-
-  return node;
-}
-
-function shift(root) {
-  if (root.first) {
-    const { value } = root.first;
-    root.first = root.first.next;
-    root.length--;
-    return value;
-  }
-}
-
-
-function getLinkedMap() {
-  const map = new Map();
-  function addToMap(key, node, operation) {
-    let refs = map.get(key);
-    if (!refs) {
-      map.set(key, refs = { length: 0 });
-    }
-      operation(refs, node);
-  }
-  const root = { length: 0 };
-
-  return {
-    get length() {
-      return root.length;
-    },
-    push(value) {
-      const node = push(root, value);
-      addToMap(value, node, push);
-
-      return this;
-    },
-    unshift(value) {
-      const node = unshift(root, value);
-      addToMap(key, node, unshift);
-
-      return this;
-    },
-    remove(value) {
-      const refs = map.get(value);
-      if (refs) {
-        const result = shift(refs);
-        if (result.previous)
-          result.previous.next = result.next;
-        if (result.next)
-          result.next.previous = result.previous;
-        if (refs.length === 0) {
-          map.delete(value);
-        }
-        root.length--;
-        return 1;
-      }
-      return 0;
-    },
-    [SymbolIterator]() {
-      let node = root.first;
-
-      const iterator = {
-        next() {
-          if (!node) {
-            return { done: true };
-          }
-          const result = {
-            done: false,
-            value: node.value,
-          };
-          node = node.next;
-          return result;
-        },
-        [SymbolIterator]() {
-          return iterator;
-        }
-      };
-
-      return iterator;
-    }
-  }
-}
-
-module.exports = getLinkedMap;
diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index ec165d27ac4744..ec245e36e70400 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -102,6 +102,6 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     get queueSize() {
       return queueSize;
     },
-    [SymbolAsyncIterator]() { return result; },
+    [SymbolAsyncIterator]() { return this; },
   };
 };

From 31efb2d89322e138fbb07d52c7765a65fde32db8 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Thu, 23 Dec 2021 15:39:22 -0300
Subject: [PATCH 11/25] lib: changing once to on in waitNext

After seeing how once method works on EventEmitter,
it was clear that it doesn't make sense to use it
into the waitNext method, as the callback
will already remove the events
---
 lib/internal/readline/eventsToAsyncIteratorFactory.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index ec245e36e70400..bc97f92d213d9f 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -25,7 +25,7 @@ function waitNext(emitter, next, events) {
       }
     };
     for (let i = 0; i < events.length; i++)
-      emitter.once(events[i], resolveNext);
+      emitter.on(events[i], resolveNext);
   });
 }
 

From 0e04a1801ca4e7708d2887b610ee98b41169d2cb Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Tue, 28 Dec 2021 23:11:28 -0300
Subject: [PATCH 12/25] lib: fixing lint on benchmark

---
 lib/events.js | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/events.js b/lib/events.js
index 11f369c52cebd3..7abf18f42c0064 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -27,7 +27,6 @@ const {
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
-  ArrayFrom,
   Boolean,
   Error,
   ErrorCaptureStackTrace,

From 7a8c3fa9fc207a3db2db39a6a30df141606027a4 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 29 Dec 2021 12:58:00 -0300
Subject: [PATCH 13/25] lib: adding test to validate slow stream

---
 .../parallel/test-readline-async-iterators.js | 41 +++++++++++++++++++
 1 file changed, 41 insertions(+)

diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
index 21bea1f5d594cf..a59b412b4178fb 100644
--- a/test/parallel/test-readline-async-iterators.js
+++ b/test/parallel/test-readline-async-iterators.js
@@ -183,7 +183,48 @@ Diam in arcu cursus euismod quis viverra nibh.
   assert.ok(percentage <= THRESHOLD, `Failed: ${totalTimeNewWay} isn't lesser than ${THRESHOLD}% of ${totalTimeOldWay}. Actual percentage: ${percentage.toFixed(2)}%`);
 }
 
+async function testSlowStreamForLeaks() {
+  const message = 'a\nb\nc\n';
+  const DELAY = 1;
+  const REPETITIONS = 100;
+  const warningCallback = common.mustNotCall();
+  process.on('warning', warningCallback);
+
+  function getStream() {
+    const readable = Readable({
+      objectMode: true,
+    });
+    readable._read = () => {};
+    let i = REPETITIONS;
+    function schedule() {
+      setTimeout(() => {
+        i--;
+        if (i < 0) {
+          readable.push(null);
+        } else {
+          readable.push(message);
+          schedule();
+        }
+      }, DELAY);
+    }
+    schedule();
+    return readable;
+  }
+  const iterable = readline.createInterface({
+    input: getStream(),
+  });
+
+  let lines = 0;
+  for await (const {} of iterable) {
+    lines++;
+  }
+
+  assert.strictEqual(lines, 3 * REPETITIONS);
+  process.off('warning', warningCallback);
+}
+
 testSimple()
   .then(testMutual)
   .then(testPerformance)
+  .then(testSlowStreamForLeaks)
   .then(common.mustCall());

From 156837bc428043ae102c812254f92f7a952e6adf Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Thu, 30 Dec 2021 13:29:07 -0300
Subject: [PATCH 14/25] lib: adding return and throw implementation to
 eventsToAsyncIteratorFactory

---
 .../readline/eventsToAsyncIteratorFactory.js  | 68 +++++++++++++------
 1 file changed, 47 insertions(+), 21 deletions(-)

diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index bc97f92d213d9f..59f10fce3708b3 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -1,9 +1,15 @@
 'use strict';
 const {
+  Error,
   Promise,
   SymbolAsyncIterator,
   ArrayPrototypeConcat,
 } = primordials;
+const {
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+  },
+} = require('internal/errors');
 const FixedQueue = require('internal/fixed_queue');
 
 const PAUSE_THRESHOLD = 1024;
@@ -12,23 +18,6 @@ const ITEM_EVENTS = ['data'];
 const CLOSE_EVENTS = ['close', 'end'];
 const ERROR_EVENTS = ['error'];
 
-
-function waitNext(emitter, next, events) {
-  return new Promise((resolve, reject) => {
-    const resolveNext = () => {
-      for (let i = 0; i < events.length; i++)
-        emitter.off(events[i], resolveNext);
-      try {
-        resolve(next());
-      } catch (promiseError) {
-        reject(promiseError);
-      }
-    };
-    for (let i = 0; i < events.length; i++)
-      emitter.on(events[i], resolveNext);
-  });
-}
-
 module.exports = function eventsToAsyncIteratorFactory(readable, {
   pauseThreshold = PAUSE_THRESHOLD,
   resumeThreshold = RESUME_THRESHOLD,
@@ -38,6 +27,17 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
 }) {
   const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents);
   const highWaterMark = RESUME_THRESHOLD;
+  let onFunction;
+  let offFunction;
+  if (typeof readable.on === 'function') {
+    onFunction = 'on';
+    offFunction = 'off';
+  } else if (typeof readable.addEventListener === 'function') {
+    onFunction = 'addEventListener';
+    offFunction = 'removeEventListener';
+  } else {
+    throw new ERR_INVALID_ARG_TYPE('readable', 'Readable', readable);
+  }
 
   const queue = new FixedQueue();
   let done = false;
@@ -45,11 +45,11 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
   let queueSize = 0;
   let paused = false;
   const onError = (value) => {
-    turn('off');
+    turn(offFunction);
     error = value;
   };
   const onClose = () => {
-    turn('off');
+    turn(offFunction);
     done = true;
   };
   const onItem = (value) => {
@@ -69,7 +69,7 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
       readable[onOff](errorEvents[i], onError);
   }
 
-  turn('on');
+  turn(onFunction);
 
   function next() {
     if (!queue.isEmpty()) {
@@ -90,7 +90,19 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     if (done) {
       return { done };
     }
-    return waitNext(readable, next, events);
+    return new Promise((resolve, reject) => {
+      const resolveNext = () => {
+        for (let i = 0; i < events.length; i++)
+          readable[offFunction](events[i], resolveNext);
+        try {
+          resolve(next());
+        } catch (promiseError) {
+          reject(promiseError);
+        }
+      };
+      for (let i = 0; i < events.length; i++)
+        readable[onFunction](events[i], resolveNext);
+    });
   }
 
   return {
@@ -102,6 +114,20 @@ module.exports = function eventsToAsyncIteratorFactory(readable, {
     get queueSize() {
       return queueSize;
     },
+    return() {
+      turn(offFunction);
+      done = true;
+
+      return { done };
+    },
+    throw(err) {
+      if (!err || !(err instanceof Error)) {
+        throw new ERR_INVALID_ARG_TYPE('err',
+                                       'Error', err);
+      }
+      error = err;
+      turn(offFunction);
+    },
     [SymbolAsyncIterator]() { return this; },
   };
 };

From 87da1a2f9d8558c269d32dcb29c0405e29d827d3 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Tue, 1 Feb 2022 21:19:07 -0300
Subject: [PATCH 15/25] lib: keep ASCII order on primordials import

---
 lib/internal/readline/eventsToAsyncIteratorFactory.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index 59f10fce3708b3..3ef047209a8373 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -1,9 +1,9 @@
 'use strict';
 const {
+  ArrayPrototypeConcat,
   Error,
   Promise,
   SymbolAsyncIterator,
-  ArrayPrototypeConcat,
 } = primordials;
 const {
   codes: {

From e7f358af8187e6ba14df039aef1ab15f636b1cf4 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Tue, 1 Feb 2022 21:21:06 -0300
Subject: [PATCH 16/25] lib: changing standard for internal constants names

---
 .../readline/eventsToAsyncIteratorFactory.js  | 22 +++++++++----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
index 3ef047209a8373..5ce19bcd37c904 100644
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ b/lib/internal/readline/eventsToAsyncIteratorFactory.js
@@ -12,21 +12,21 @@ const {
 } = require('internal/errors');
 const FixedQueue = require('internal/fixed_queue');
 
-const PAUSE_THRESHOLD = 1024;
-const RESUME_THRESHOLD = 1;
-const ITEM_EVENTS = ['data'];
-const CLOSE_EVENTS = ['close', 'end'];
-const ERROR_EVENTS = ['error'];
+const kPauseThreshold = 1024;
+const kResumeThreshold = 1;
+const kItemEvents = ['data'];
+const kCloseEvents = ['close', 'end'];
+const kErrorEvents = ['error'];
 
 module.exports = function eventsToAsyncIteratorFactory(readable, {
-  pauseThreshold = PAUSE_THRESHOLD,
-  resumeThreshold = RESUME_THRESHOLD,
-  closeEvents = CLOSE_EVENTS,
-  itemEvents = ITEM_EVENTS,
-  errorEvents = ERROR_EVENTS,
+  pauseThreshold = kPauseThreshold,
+  resumeThreshold = kResumeThreshold,
+  closeEvents = kCloseEvents,
+  itemEvents = kItemEvents,
+  errorEvents = kErrorEvents,
 }) {
   const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents);
-  const highWaterMark = RESUME_THRESHOLD;
+  const highWaterMark = kResumeThreshold;
   let onFunction;
   let offFunction;
   if (typeof readable.on === 'function') {

From d5917c96ac4525b887dd682be920cd02c084b225 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Thu, 3 Feb 2022 22:41:43 -0300
Subject: [PATCH 17/25] lib: unifying on and eventsToAsyncIteratorFactory
 functions

---
 lib/events.js                                 | 124 +++++++++++-----
 .../readline/eventsToAsyncIteratorFactory.js  | 133 ------------------
 lib/internal/readline/interface.js            |  11 +-
 3 files changed, 96 insertions(+), 172 deletions(-)
 delete mode 100644 lib/internal/readline/eventsToAsyncIteratorFactory.js

diff --git a/lib/events.js b/lib/events.js
index 7abf18f42c0064..6d05c63c4a989b 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -23,7 +23,6 @@
 
 const {
   ArrayPrototypeJoin,
-  ArrayPrototypeShift,
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
@@ -33,6 +32,7 @@ const {
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   NumberIsNaN,
+  NumberPOSITIVE_INFINITY,
   ObjectCreate,
   ObjectDefineProperty,
   ObjectDefineProperties,
@@ -84,7 +84,7 @@ const kErrorMonitor = Symbol('events.errorMonitor');
 const kMaxEventTargetListeners = Symbol('events.maxEventTargetListeners');
 const kMaxEventTargetListenersWarned =
   Symbol('events.maxEventTargetListenersWarned');
-
+let FixedQueue;
 let EventEmitterAsyncResource;
 // The EventEmitterAsyncResource has to be initialized lazily because event.js
 // is loaded so early in the bootstrap process, before async_hooks is available.
@@ -999,7 +999,13 @@ 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[];
+ *    pauseThreshold?: number,
+ *    resumeThreshold?: number,
+ *    firstEventParam: boolean
+ *   }} [options]
  * @returns {AsyncIterator}
  */
 function on(emitter, event, options) {
@@ -1008,16 +1014,44 @@ function on(emitter, event, options) {
   if (signal?.aborted)
     throw new AbortError(undefined, { cause: signal?.reason });
 
-  const unconsumedEvents = [];
-  const unconsumedPromises = [];
+  if (!FixedQueue) FixedQueue = require('internal/fixed_queue');
+
+  const unconsumedEvents = new FixedQueue();
+  const unconsumedPromises = new FixedQueue();
   let error = null;
   let finished = false;
+  const close = options?.close;
+  const firstEventParam = options?.firstEventParam;
+  let pauseThreshold = options?.pauseThreshold;
+  let resumeThreshold = options?.resumeThreshold;
+  let paused = false;
+  let size = 0;
+  if (pauseThreshold === undefined) {
+    pauseThreshold = NumberPOSITIVE_INFINITY;
+  }
+  if (resumeThreshold === undefined) {
+    resumeThreshold = 1;
+  }
 
   const iterator = ObjectSetPrototypeOf({
+    get highWaterMark() {
+      return pauseThreshold;
+    },
+    get isPaused() {
+      return paused;
+    },
+    get queueSize() {
+      return size;
+    },
     next() {
       // First, we consume all unread events
-      const value = unconsumedEvents.shift();
-      if (value) {
+      if (size) {
+        const value = unconsumedEvents.shift();
+        size--;
+        if (paused && size < resumeThreshold) {
+          emitter.resume();
+          paused = false;
+        }
         return PromiseResolve(createIterResult(value, false));
       }
 
@@ -1033,6 +1067,7 @@ function on(emitter, event, options) {
 
       // If the iterator is finished, resolve to done
       if (finished) {
+        removeListeners();
         return PromiseResolve(createIterResult(undefined, true));
       }
 
@@ -1043,24 +1078,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) {
@@ -1069,8 +1087,7 @@ function on(emitter, event, options) {
                                        'Error', err);
       }
       error = err;
-      eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
-      eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
+      removeListeners();
     },
 
     [SymbolAsyncIterator]() {
@@ -1078,10 +1095,15 @@ function on(emitter, event, options) {
     }
   }, AsyncIteratorPrototype);
 
-  eventTargetAgnosticAddListener(emitter, event, eventHandler);
+  eventTargetAgnosticAddListener(emitter, event, firstEventParam ? eventHandlerFirstParam : eventHandler);
   if (event !== 'error' && typeof emitter.on === 'function') {
     emitter.on('error', errorHandler);
   }
+  if (close && close.length) {
+    for (let i = 0; i <= close.length; i++) {
+      eventTargetAgnosticAddListener(emitter, close[i], closeHandler);
+    }
+  }
 
   if (signal) {
     eventTargetAgnosticAddListener(
@@ -1093,23 +1115,59 @@ function on(emitter, event, options) {
 
   return iterator;
 
+  function closeHandler() {
+    removeListeners();
+    finished = true;
+
+    while (!unconsumedPromises.isEmpty()) {
+      unconsumedPromises.shift().resolve(createIterResult(undefined, true));
+    }
+
+    return PromiseResolve(createIterResult(undefined, true));
+  }
+
+  function removeListeners() {
+    eventTargetAgnosticRemoveListener(emitter, event, firstEventParam ? eventHandlerFirstParam : eventHandler);
+    eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
+    if (close && close.length) {
+      for (let i = 0; i <= close.length; i++) {
+        eventTargetAgnosticRemoveListener(emitter, close[i], closeHandler);
+      }
+    }
+    if (signal) {
+      eventTargetAgnosticRemoveListener(
+        signal,
+        'abort',
+        abortListener,
+        { once: true });
+    }
+  }
+
   function abortListener() {
     errorHandler(new AbortError(undefined, { cause: signal?.reason }));
   }
 
   function eventHandler(...args) {
-    const promise = ArrayPrototypeShift(unconsumedPromises);
-    if (promise) {
-      promise.resolve(createIterResult(args, false));
+    return eventHandlerFirstParam(args);
+  }
+
+  function eventHandlerFirstParam(arg) {
+    if (unconsumedPromises.isEmpty()) {
+      size++;
+      if (size > pauseThreshold) {
+        paused = true;
+        emitter.pause();
+      }
+      unconsumedEvents.push(arg);
     } else {
-      unconsumedEvents.push(args);
+      unconsumedPromises.shift().resolve(createIterResult(arg, false));
     }
   }
 
   function errorHandler(err) {
     finished = true;
 
-    const toError = ArrayPrototypeShift(unconsumedPromises);
+    const toError = unconsumedPromises.shift();
 
     if (toError) {
       toError.reject(err);
diff --git a/lib/internal/readline/eventsToAsyncIteratorFactory.js b/lib/internal/readline/eventsToAsyncIteratorFactory.js
deleted file mode 100644
index 5ce19bcd37c904..00000000000000
--- a/lib/internal/readline/eventsToAsyncIteratorFactory.js
+++ /dev/null
@@ -1,133 +0,0 @@
-'use strict';
-const {
-  ArrayPrototypeConcat,
-  Error,
-  Promise,
-  SymbolAsyncIterator,
-} = primordials;
-const {
-  codes: {
-    ERR_INVALID_ARG_TYPE,
-  },
-} = require('internal/errors');
-const FixedQueue = require('internal/fixed_queue');
-
-const kPauseThreshold = 1024;
-const kResumeThreshold = 1;
-const kItemEvents = ['data'];
-const kCloseEvents = ['close', 'end'];
-const kErrorEvents = ['error'];
-
-module.exports = function eventsToAsyncIteratorFactory(readable, {
-  pauseThreshold = kPauseThreshold,
-  resumeThreshold = kResumeThreshold,
-  closeEvents = kCloseEvents,
-  itemEvents = kItemEvents,
-  errorEvents = kErrorEvents,
-}) {
-  const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents);
-  const highWaterMark = kResumeThreshold;
-  let onFunction;
-  let offFunction;
-  if (typeof readable.on === 'function') {
-    onFunction = 'on';
-    offFunction = 'off';
-  } else if (typeof readable.addEventListener === 'function') {
-    onFunction = 'addEventListener';
-    offFunction = 'removeEventListener';
-  } else {
-    throw new ERR_INVALID_ARG_TYPE('readable', 'Readable', readable);
-  }
-
-  const queue = new FixedQueue();
-  let done = false;
-  let error;
-  let queueSize = 0;
-  let paused = false;
-  const onError = (value) => {
-    turn(offFunction);
-    error = value;
-  };
-  const onClose = () => {
-    turn(offFunction);
-    done = true;
-  };
-  const onItem = (value) => {
-    queue.push(value);
-    queueSize++;
-    if (queueSize >= pauseThreshold) {
-      paused = true;
-      readable.pause();
-    }
-  };
-  function turn(onOff) {
-    for (let i = 0; i < closeEvents.length; i++)
-      readable[onOff](closeEvents[i], onClose);
-    for (let i = 0; i < itemEvents.length; i++)
-      readable[onOff](itemEvents[i], onItem);
-    for (let i = 0; i < itemEvents.length; i++)
-      readable[onOff](errorEvents[i], onError);
-  }
-
-  turn(onFunction);
-
-  function next() {
-    if (!queue.isEmpty()) {
-      const value = queue.shift();
-      queueSize--;
-      if (queueSize < resumeThreshold) {
-        paused = false;
-        readable.resume();
-      }
-      return {
-        done: false,
-        value,
-      };
-    }
-    if (error) {
-      throw error;
-    }
-    if (done) {
-      return { done };
-    }
-    return new Promise((resolve, reject) => {
-      const resolveNext = () => {
-        for (let i = 0; i < events.length; i++)
-          readable[offFunction](events[i], resolveNext);
-        try {
-          resolve(next());
-        } catch (promiseError) {
-          reject(promiseError);
-        }
-      };
-      for (let i = 0; i < events.length; i++)
-        readable[onFunction](events[i], resolveNext);
-    });
-  }
-
-  return {
-    next,
-    highWaterMark,
-    get isPaused() {
-      return paused;
-    },
-    get queueSize() {
-      return queueSize;
-    },
-    return() {
-      turn(offFunction);
-      done = true;
-
-      return { done };
-    },
-    throw(err) {
-      if (!err || !(err instanceof Error)) {
-        throw new ERR_INVALID_ARG_TYPE('err',
-                                       'Error', err);
-      }
-      error = err;
-      turn(offFunction);
-    },
-    [SymbolAsyncIterator]() { return this; },
-  };
-};
diff --git a/lib/internal/readline/interface.js b/lib/internal/readline/interface.js
index 6f842a3c7ca51b..4e7583b7749a96 100644
--- a/lib/internal/readline/interface.js
+++ b/lib/internal/readline/interface.js
@@ -1343,12 +1343,11 @@ class Interface extends InterfaceConstructor {
    */
   [SymbolAsyncIterator]() {
     if (!this[kLineObjectStream]) {
-      this[kLineObjectStream] = require(
-        'internal/readline/eventsToAsyncIteratorFactory'
-      )(
-        this, {
-          itemEvents: ['line'],
-          closeEvents: ['close']
+      this[kLineObjectStream] = EventEmitter.on(
+        this, 'line', {
+          close: ['close'],
+          pauseThreshold: 1024,
+          firstEventParam: true,
         });
     }
     return this[kLineObjectStream];

From 0143cdfc50cda7fe05efd50780eb04091744f747 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Fri, 4 Feb 2022 09:20:15 -0300
Subject: [PATCH 18/25] lib: making firstEventParam interna only

---
 lib/events.js                      |  7 ++++---
 lib/internal/events/symbols.js     | 11 +++++++++++
 lib/internal/readline/interface.js |  4 +++-
 3 files changed, 18 insertions(+), 4 deletions(-)
 create mode 100644 lib/internal/events/symbols.js

diff --git a/lib/events.js b/lib/events.js
index 6d05c63c4a989b..5e29990d22a432 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -85,6 +85,7 @@ const kMaxEventTargetListeners = Symbol('events.maxEventTargetListeners');
 const kMaxEventTargetListenersWarned =
   Symbol('events.maxEventTargetListenersWarned');
 let FixedQueue;
+let kFirstEventParam;
 let EventEmitterAsyncResource;
 // The EventEmitterAsyncResource has to be initialized lazily because event.js
 // is loaded so early in the bootstrap process, before async_hooks is available.
@@ -1003,8 +1004,7 @@ function eventTargetAgnosticAddListener(emitter, name, listener, flags) {
  *    signal: AbortSignal;
  *    close?: string[];
  *    pauseThreshold?: number,
- *    resumeThreshold?: number,
- *    firstEventParam: boolean
+ *    resumeThreshold?: number
  *   }} [options]
  * @returns {AsyncIterator}
  */
@@ -1015,13 +1015,14 @@ function on(emitter, event, options) {
     throw new AbortError(undefined, { cause: signal?.reason });
 
   if (!FixedQueue) FixedQueue = require('internal/fixed_queue');
+  if (!kFirstEventParam) kFirstEventParam = require('internal/events/symbols').kFirstEventParam;
 
   const unconsumedEvents = new FixedQueue();
   const unconsumedPromises = new FixedQueue();
   let error = null;
   let finished = false;
   const close = options?.close;
-  const firstEventParam = options?.firstEventParam;
+  const firstEventParam = options?.[kFirstEventParam];
   let pauseThreshold = options?.pauseThreshold;
   let resumeThreshold = options?.resumeThreshold;
   let paused = false;
diff --git a/lib/internal/events/symbols.js b/lib/internal/events/symbols.js
new file mode 100644
index 00000000000000..92ce58484aabc9
--- /dev/null
+++ b/lib/internal/events/symbols.js
@@ -0,0 +1,11 @@
+'use strict';
+
+const {
+  Symbol
+} = primordials;
+
+const kFirstEventParam = Symbol('nodejs.kFirstEventParam');
+
+module.exports = {
+  kFirstEventParam,
+};
diff --git a/lib/internal/readline/interface.js b/lib/internal/readline/interface.js
index 4e7583b7749a96..0d9972fd7e76d0 100644
--- a/lib/internal/readline/interface.js
+++ b/lib/internal/readline/interface.js
@@ -62,6 +62,7 @@ const {
   kSubstringSearch,
 } = require('internal/readline/utils');
 let emitKeypressEvents;
+let kFirstEventParam;
 const {
   clearScreenDown,
   cursorTo,
@@ -1343,11 +1344,12 @@ class Interface extends InterfaceConstructor {
    */
   [SymbolAsyncIterator]() {
     if (!this[kLineObjectStream]) {
+      if (!kFirstEventParam) kFirstEventParam = require('internal/events/symbols').kFirstEventParam;
       this[kLineObjectStream] = EventEmitter.on(
         this, 'line', {
           close: ['close'],
           pauseThreshold: 1024,
-          firstEventParam: true,
+          [kFirstEventParam]: true,
         });
     }
     return this[kLineObjectStream];

From 917c51b105d97098501fa658558002480bf153f9 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Sat, 5 Feb 2022 14:44:33 -0300
Subject: [PATCH 19/25] lib: refactoring function on

---
 lib/events.js                                 | 184 +++++++++---------
 lib/internal/readline/interface.js            |   2 +-
 ...t-readline-async-iterators-backpressure.js |   9 +-
 3 files changed, 96 insertions(+), 99 deletions(-)

diff --git a/lib/events.js b/lib/events.js
index 5e29990d22a432..1e2d1e0926d0c1 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -32,7 +32,7 @@ const {
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   NumberIsNaN,
-  NumberPOSITIVE_INFINITY,
+  NumberMAX_SAFE_INTEGER,
   ObjectCreate,
   ObjectDefineProperty,
   ObjectDefineProperties,
@@ -47,7 +47,6 @@ const {
   StringPrototypeSplit,
   Symbol,
   SymbolFor,
-  SymbolAsyncIterator,
 } = primordials;
 const kRejection = SymbolFor('nodejs.rejection');
 
@@ -59,6 +58,8 @@ const {
 } = require('internal/util/inspect');
 
 let spliceOne;
+let FixedQueue;
+let kFirstEventParam;
 
 const {
   AbortError,
@@ -73,6 +74,7 @@ const {
 } = require('internal/errors');
 
 const {
+  validateInteger,
   validateAbortSignal,
   validateBoolean,
   validateFunction,
@@ -84,8 +86,8 @@ const kErrorMonitor = Symbol('events.errorMonitor');
 const kMaxEventTargetListeners = Symbol('events.maxEventTargetListeners');
 const kMaxEventTargetListenersWarned =
   Symbol('events.maxEventTargetListenersWarned');
-let FixedQueue;
-let kFirstEventParam;
+const kWatermarkData = SymbolFor('nodejs.watermarkData');
+
 let EventEmitterAsyncResource;
 // The EventEmitterAsyncResource has to be initialized lazily because event.js
 // is loaded so early in the bootstrap process, before async_hooks is available.
@@ -1003,57 +1005,42 @@ function eventTargetAgnosticAddListener(emitter, name, listener, flags) {
  * @param {{
  *    signal: AbortSignal;
  *    close?: string[];
- *    pauseThreshold?: number,
- *    resumeThreshold?: number
+ *    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 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
   if (!FixedQueue) FixedQueue = require('internal/fixed_queue');
-  if (!kFirstEventParam) kFirstEventParam = require('internal/events/symbols').kFirstEventParam;
-
   const unconsumedEvents = new FixedQueue();
   const unconsumedPromises = new FixedQueue();
+  let paused = false;
   let error = null;
   let finished = false;
-  const close = options?.close;
-  const firstEventParam = options?.[kFirstEventParam];
-  let pauseThreshold = options?.pauseThreshold;
-  let resumeThreshold = options?.resumeThreshold;
-  let paused = false;
   let size = 0;
-  if (pauseThreshold === undefined) {
-    pauseThreshold = NumberPOSITIVE_INFINITY;
-  }
-  if (resumeThreshold === undefined) {
-    resumeThreshold = 1;
-  }
 
   const iterator = ObjectSetPrototypeOf({
-    get highWaterMark() {
-      return pauseThreshold;
-    },
-    get isPaused() {
-      return paused;
-    },
-    get queueSize() {
-      return size;
-    },
     next() {
       // First, we consume all unread events
       if (size) {
         const value = unconsumedEvents.shift();
         size--;
-        if (paused && size < resumeThreshold) {
+        if (paused && size < lowWatermark) {
           emitter.resume();
           paused = false;
         }
-        return PromiseResolve(createIterResult(value, false));
+        return value;
       }
 
       // Then we error, if an error happened
@@ -1067,10 +1054,7 @@ function on(emitter, event, options) {
       }
 
       // If the iterator is finished, resolve to done
-      if (finished) {
-        removeListeners();
-        return PromiseResolve(createIterResult(undefined, true));
-      }
+      if (finished) return closeHandler();
 
       // Wait until an event happens
       return new Promise(function(resolve, reject) {
@@ -1087,27 +1071,53 @@ function on(emitter, event, options) {
         throw new ERR_INVALID_ARG_TYPE('EventEmitter.AsyncIterator',
                                        'Error', err);
       }
-      error = err;
-      removeListeners();
+      errorHandler(err);
+    },
+    [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;
+      }
     },
-
-    [SymbolAsyncIterator]() {
-      return this;
-    }
   }, AsyncIteratorPrototype);
 
-  eventTargetAgnosticAddListener(emitter, event, firstEventParam ? eventHandlerFirstParam : eventHandler);
+  // Adding event handlers
+  const { addEventListener, removeAll } = listenersController();
+  if (!kFirstEventParam) 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);
   }
-  if (close && close.length) {
-    for (let i = 0; i <= close.length; i++) {
-      eventTargetAgnosticAddListener(emitter, close[i], closeHandler);
+  const closeEvents = options?.close;
+  if (closeEvents && closeEvents.length) {
+    for (let i = 0; i < closeEvents.length; i++) {
+      addEventListener(emitter, closeEvents[i], closeHandler);
     }
   }
-
   if (signal) {
-    eventTargetAgnosticAddListener(
+    addEventListener(
       signal,
       'abort',
       abortListener,
@@ -1116,67 +1126,53 @@ function on(emitter, event, options) {
 
   return iterator;
 
-  function closeHandler() {
-    removeListeners();
-    finished = true;
-
-    while (!unconsumedPromises.isEmpty()) {
-      unconsumedPromises.shift().resolve(createIterResult(undefined, true));
-    }
-
-    return PromiseResolve(createIterResult(undefined, true));
-  }
-
-  function removeListeners() {
-    eventTargetAgnosticRemoveListener(emitter, event, firstEventParam ? eventHandlerFirstParam : eventHandler);
-    eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
-    if (close && close.length) {
-      for (let i = 0; i <= close.length; i++) {
-        eventTargetAgnosticRemoveListener(emitter, close[i], closeHandler);
-      }
-    }
-    if (signal) {
-      eventTargetAgnosticRemoveListener(
-        signal,
-        'abort',
-        abortListener,
-        { once: true });
-    }
-  }
-
   function abortListener() {
     errorHandler(new AbortError(undefined, { cause: signal?.reason }));
   }
 
-  function eventHandler(...args) {
-    return eventHandlerFirstParam(args);
-  }
-
-  function eventHandlerFirstParam(arg) {
+  function eventHandler(value) {
+    const arg = createIterResult(value, false);
     if (unconsumedPromises.isEmpty()) {
       size++;
-      if (size > pauseThreshold) {
+      if (!paused && size > highWatermark) {
         paused = true;
         emitter.pause();
       }
       unconsumedEvents.push(arg);
-    } else {
-      unconsumedPromises.shift().resolve(createIterResult(arg, false));
-    }
+    } else unconsumedPromises.shift().resolve(arg);
   }
 
   function errorHandler(err) {
-    finished = true;
+    if (unconsumedPromises.isEmpty()) error = err;
+    else unconsumedPromises.shift().reject(err);
 
-    const toError = unconsumedPromises.shift();
+    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);
+      listeners.push([emitter, event, handler, flags]);
+    },
+    removeAll() {
+      while (listeners.length > 0) {
+        eventTargetAgnosticRemoveListener(...listeners.pop());
+      }
+    }
+  };
+}
diff --git a/lib/internal/readline/interface.js b/lib/internal/readline/interface.js
index 0d9972fd7e76d0..5779b0ff4bd672 100644
--- a/lib/internal/readline/interface.js
+++ b/lib/internal/readline/interface.js
@@ -1348,7 +1348,7 @@ class Interface extends InterfaceConstructor {
       this[kLineObjectStream] = EventEmitter.on(
         this, 'line', {
           close: ['close'],
-          pauseThreshold: 1024,
+          highWatermark: 1024,
           [kFirstEventParam]: true,
         });
     }
diff --git a/test/parallel/test-readline-async-iterators-backpressure.js b/test/parallel/test-readline-async-iterators-backpressure.js
index fddaeb368593e4..e8f443f4383346 100644
--- a/test/parallel/test-readline-async-iterators-backpressure.js
+++ b/test/parallel/test-readline-async-iterators-backpressure.js
@@ -24,11 +24,12 @@ const REPETITIONS = 3;
   });
 
   const it = rli[Symbol.asyncIterator]();
-  const highWaterMark = it.highWaterMark;
+  const watermarkData = it[Symbol.for('nodejs.watermarkData')];
+  const highWaterMark = watermarkData.high;
 
   // For this test to work, we have to queue up more than the number of
   // highWaterMark items in rli. Make sure that is the case.
-  assert(TOTAL_LINES > highWaterMark);
+  assert(TOTAL_LINES > highWaterMark, `TOTAL_LINES (${TOTAL_LINES}) isn't greater than highWaterMark (${highWaterMark})`);
 
   let iterations = 0;
   let readableEnded = false;
@@ -36,8 +37,8 @@ const REPETITIONS = 3;
   for await (const line of it) {
     assert.strictEqual(readableEnded, false);
     assert.strictEqual(line, CONTENT);
-    assert.ok(it.queueSize <= TOTAL_LINES);
-    assert.strictEqual(readable.isPaused(), it.queueSize >= 1);
+    assert.ok(watermarkData.size <= TOTAL_LINES);
+    assert.strictEqual(readable.isPaused(), watermarkData.size >= 1);
     if (!readable.isPaused()) {
       notPaused++;
     }

From 4e8be801bc3fd01dab44019a69037cda5e7c6c4e Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 14 Sep 2022 17:11:55 -0300
Subject: [PATCH 20/25] lint: fixing linting problems after merge

---
 benchmark/readline/readline-iterable.js        | 3 ++-
 test/parallel/test-readline-async-iterators.js | 4 ++--
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/benchmark/readline/readline-iterable.js b/benchmark/readline/readline-iterable.js
index 5c35987d997b96..6dc73b7669962a 100644
--- a/benchmark/readline/readline-iterable.js
+++ b/benchmark/readline/readline-iterable.js
@@ -39,7 +39,8 @@ async function main({ n }) {
     input: getLoremIpsumStream(n),
   });
 
-  for await (const {} of iterable) {
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of iterable) {
     lineCount++;
   }
   bench.end(lineCount);
diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
index a59b412b4178fb..33ab159ab7f8eb 100644
--- a/test/parallel/test-readline-async-iterators.js
+++ b/test/parallel/test-readline-async-iterators.js
@@ -5,7 +5,6 @@ const fs = require('fs');
 const { join } = require('path');
 const readline = require('readline');
 const { Readable } = require('stream');
-const { Readable } = require('stream');
 const assert = require('assert');
 
 const tmpdir = require('../common/tmpdir');
@@ -215,7 +214,8 @@ async function testSlowStreamForLeaks() {
   });
 
   let lines = 0;
-  for await (const {} of iterable) {
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of iterable) {
     lines++;
   }
 

From 610a2758f851c0a6d7d361755a14ef2be523adee Mon Sep 17 00:00:00 2001
From: Thiago Oliveira Santos <tos.oliveira@gmail.com>
Date: Wed, 14 Sep 2022 19:35:51 -0300
Subject: [PATCH 21/25] refactor: Apply suggestions from code review

Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com>
---
 benchmark/readline/readline-iterable.js       |  3 ++-
 lib/events.js                                 | 22 ++++++++++---------
 lib/internal/events/symbols.js                |  2 +-
 lib/internal/readline/interface.js            |  4 ++--
 .../parallel/test-readline-async-iterators.js |  3 ++-
 5 files changed, 19 insertions(+), 15 deletions(-)

diff --git a/benchmark/readline/readline-iterable.js b/benchmark/readline/readline-iterable.js
index 6dc73b7669962a..e706c7f2f1506e 100644
--- a/benchmark/readline/readline-iterable.js
+++ b/benchmark/readline/readline-iterable.js
@@ -7,7 +7,8 @@ const bench = common.createBenchmark(main, {
   n: [1e1, 1e2, 1e3, 1e4, 1e5, 1e6],
 });
 
-const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
+const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed
+do eiusmod tempor incididunt ut labore et dolore magna aliqua.
 Dui accumsan sit amet nulla facilisi morbi tempus iaculis urna.
 Eget dolor morbi non arcu risus quis varius quam quisque.
 Lacus viverra vitae congue eu consequat ac felis donec.
diff --git a/lib/events.js b/lib/events.js
index 1e2d1e0926d0c1..e5aee007c64630 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -23,6 +23,8 @@
 
 const {
   ArrayPrototypeJoin,
+  ArrayPrototypePop,
+  ArrayPrototypeShift,
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
@@ -1016,13 +1018,13 @@ function on(emitter, event, options = {}) {
   validateAbortSignal(signal, 'options.signal');
   if (signal?.aborted)
     throw new AbortError(undefined, { cause: signal?.reason });
-  const highWatermark = options.highWatermark || NumberMAX_SAFE_INTEGER;
+  const highWatermark = options.highWatermark ?? NumberMAX_SAFE_INTEGER;
   validateInteger(highWatermark, 'options.highWatermark', 1);
-  const lowWatermark = options.lowWatermark || 1;
+  const lowWatermark = options.lowWatermark ?? 1;
   validateInteger(lowWatermark, 'options.lowWatermark', 1);
 
   // Preparing controlling queues and variables
-  if (!FixedQueue) FixedQueue = require('internal/fixed_queue');
+  FixedQueue ??= require('internal/fixed_queue');
   const unconsumedEvents = new FixedQueue();
   const unconsumedPromises = new FixedQueue();
   let paused = false;
@@ -1103,7 +1105,7 @@ function on(emitter, event, options = {}) {
 
   // Adding event handlers
   const { addEventListener, removeAll } = listenersController();
-  if (!kFirstEventParam) kFirstEventParam = require('internal/events/symbols').kFirstEventParam;
+  kFirstEventParam ??= require('internal/events/symbols').kFirstEventParam;
   addEventListener(emitter, event, options[kFirstEventParam] ? eventHandler : function(...args) {
     return eventHandler(args);
   });
@@ -1111,13 +1113,13 @@ function on(emitter, event, options = {}) {
     addEventListener(emitter, 'error', errorHandler);
   }
   const closeEvents = options?.close;
-  if (closeEvents && closeEvents.length) {
+  if (closeEvents?.length) {
     for (let i = 0; i < closeEvents.length; i++) {
       addEventListener(emitter, closeEvents[i], closeHandler);
     }
   }
   if (signal) {
-    addEventListener(
+    eventTargetAgnosticAddListener(
       signal,
       'abort',
       abortListener,
@@ -1139,12 +1141,12 @@ function on(emitter, event, options = {}) {
         emitter.pause();
       }
       unconsumedEvents.push(arg);
-    } else unconsumedPromises.shift().resolve(arg);
+    } else ArrayPrototypeShift(unconsumedPromises).resolve(arg);
   }
 
   function errorHandler(err) {
     if (unconsumedPromises.isEmpty()) error = err;
-    else unconsumedPromises.shift().reject(err);
+    else ArrayPrototypeShift(unconsumedPromises).reject(err);
 
     closeHandler();
   }
@@ -1154,7 +1156,7 @@ function on(emitter, event, options = {}) {
     finished = true;
     const doneResult = createIterResult(undefined, true);
     while (!unconsumedPromises.isEmpty()) {
-      unconsumedPromises.shift().resolve(doneResult);
+      ArrayPrototypeShift(unconsumedPromises).resolve(doneResult);
     }
 
     return PromiseResolve(doneResult);
@@ -1171,7 +1173,7 @@ function listenersController() {
     },
     removeAll() {
       while (listeners.length > 0) {
-        eventTargetAgnosticRemoveListener(...listeners.pop());
+        ReflectApply(eventTargetAgnosticRemoveListener, undefined, ArrayPrototypePop(listeners));
       }
     }
   };
diff --git a/lib/internal/events/symbols.js b/lib/internal/events/symbols.js
index 92ce58484aabc9..b1b89ddb8f0a4d 100644
--- a/lib/internal/events/symbols.js
+++ b/lib/internal/events/symbols.js
@@ -1,7 +1,7 @@
 'use strict';
 
 const {
-  Symbol
+  Symbol,
 } = primordials;
 
 const kFirstEventParam = Symbol('nodejs.kFirstEventParam');
diff --git a/lib/internal/readline/interface.js b/lib/internal/readline/interface.js
index 5779b0ff4bd672..5de6a8fe03da13 100644
--- a/lib/internal/readline/interface.js
+++ b/lib/internal/readline/interface.js
@@ -1343,8 +1343,8 @@ class Interface extends InterfaceConstructor {
    * @returns {InterfaceAsyncIterator}
    */
   [SymbolAsyncIterator]() {
-    if (!this[kLineObjectStream]) {
-      if (!kFirstEventParam) kFirstEventParam = require('internal/events/symbols').kFirstEventParam;
+    if (this[kLineObjectStream] === undefined) {
+      kFirstEventParam ??= require('internal/events/symbols').kFirstEventParam;
       this[kLineObjectStream] = EventEmitter.on(
         this, 'line', {
           close: ['close'],
diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
index 33ab159ab7f8eb..57c74f69d587d8 100644
--- a/test/parallel/test-readline-async-iterators.js
+++ b/test/parallel/test-readline-async-iterators.js
@@ -75,7 +75,8 @@ async function testMutual() {
 }
 
 async function testPerformance() {
-  const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
+  const loremIpsum = `Lorem ipsum dolor sit amet, consectetur adipiscing elit,
+sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
 Dui accumsan sit amet nulla facilisi morbi tempus iaculis urna.
 Eget dolor morbi non arcu risus quis varius quam quisque.
 Lacus viverra vitae congue eu consequat ac felis donec.

From e1dd9d3965e72f434ffc272b06d6fff025cf78b8 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 14 Sep 2022 20:03:58 -0300
Subject: [PATCH 22/25] refactor: readding wrongly removed SymAsyncIterator
 definition

---
 lib/events.js | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/lib/events.js b/lib/events.js
index e5aee007c64630..b8c13ea9b7c61f 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -49,6 +49,7 @@ const {
   StringPrototypeSplit,
   Symbol,
   SymbolFor,
+  SymbolAsyncIterator,
 } = primordials;
 const kRejection = SymbolFor('nodejs.rejection');
 
@@ -1075,6 +1076,9 @@ function on(emitter, event, options = {}) {
       }
       errorHandler(err);
     },
+    [SymbolAsyncIterator]() {
+      return this;
+    },
     [kWatermarkData]: {
       /**
        * The current queue size

From 26d066bdfa5f2d52cc5a69cd3cb5bb407f0d55b0 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Wed, 14 Sep 2022 20:07:27 -0300
Subject: [PATCH 23/25] refactor: using ArrayPrototypePush instead of push

---
 lib/events.js | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/lib/events.js b/lib/events.js
index b8c13ea9b7c61f..54244a192a4303 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -24,6 +24,7 @@
 const {
   ArrayPrototypeJoin,
   ArrayPrototypePop,
+  ArrayPrototypePush,
   ArrayPrototypeShift,
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
@@ -1173,7 +1174,7 @@ function listenersController() {
   return {
     addEventListener(emitter, event, handler, flags) {
       eventTargetAgnosticAddListener(emitter, event, handler, flags);
-      listeners.push([emitter, event, handler, flags]);
+      ArrayPrototypePush(listeners, [emitter, event, handler, flags]);
     },
     removeAll() {
       while (listeners.length > 0) {

From 835bb5f7b476f614278eb8158b8e9f4456cc7254 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Thu, 15 Sep 2022 07:48:27 -0300
Subject: [PATCH 24/25] lib: fixing FixedQueue.shift calls mistakenly using
 ArrayPrototypeShift

---
 lib/events.js | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/lib/events.js b/lib/events.js
index 54244a192a4303..bd946e3298fad1 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -25,7 +25,6 @@ const {
   ArrayPrototypeJoin,
   ArrayPrototypePop,
   ArrayPrototypePush,
-  ArrayPrototypeShift,
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
@@ -1146,12 +1145,12 @@ function on(emitter, event, options = {}) {
         emitter.pause();
       }
       unconsumedEvents.push(arg);
-    } else ArrayPrototypeShift(unconsumedPromises).resolve(arg);
+    } else unconsumedPromises.shift().resolve(arg);
   }
 
   function errorHandler(err) {
     if (unconsumedPromises.isEmpty()) error = err;
-    else ArrayPrototypeShift(unconsumedPromises).reject(err);
+    else unconsumedPromises.shift().reject(err);
 
     closeHandler();
   }
@@ -1161,7 +1160,7 @@ function on(emitter, event, options = {}) {
     finished = true;
     const doneResult = createIterResult(undefined, true);
     while (!unconsumedPromises.isEmpty()) {
-      ArrayPrototypeShift(unconsumedPromises).resolve(doneResult);
+      unconsumedPromises.shift().resolve(doneResult);
     }
 
     return PromiseResolve(doneResult);

From e83baeabb292c8056c14f8477a6271c5c8477649 Mon Sep 17 00:00:00 2001
From: Thiago Santos <tos.oliveira@gmail.com>
Date: Fri, 16 Sep 2022 20:57:47 -0300
Subject: [PATCH 25/25] lib: fixing faulting PromiseResolve call

---
 lib/events.js | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/lib/events.js b/lib/events.js
index bd946e3298fad1..e05a3bc6b8b168 100644
--- a/lib/events.js
+++ b/lib/events.js
@@ -1043,7 +1043,7 @@ function on(emitter, event, options = {}) {
           emitter.resume();
           paused = false;
         }
-        return value;
+        return PromiseResolve(createIterResult(value, false));
       }
 
       // Then we error, if an error happened
@@ -1137,15 +1137,14 @@ function on(emitter, event, options = {}) {
   }
 
   function eventHandler(value) {
-    const arg = createIterResult(value, false);
     if (unconsumedPromises.isEmpty()) {
       size++;
       if (!paused && size > highWatermark) {
         paused = true;
         emitter.pause();
       }
-      unconsumedEvents.push(arg);
-    } else unconsumedPromises.shift().resolve(arg);
+      unconsumedEvents.push(value);
+    } else unconsumedPromises.shift().resolve(createIterResult(value, false));
   }
 
   function errorHandler(err) {