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/internal/readline/eventsToAsyncIteratorFactory.js

@@ -0,0 +1,114 @@
+'use strict';
+const {
+  Promise,
+  SymbolAsyncIterator,
+} = primordials;
+
+const PAUSE_THRESHOLD = 1024;
+const RESUME_THRESHOLD = 1;
+const ITEM_EVENTS = ['data'];
+const CLOSE_EVENTS = ['close', 'end'];
+const ERROR_EVENTS = ['error'];
+
+let FixedQueue;
+let setImmediate;
+function waitNext(emitter, next, events) {
+  return new Promise((resolve, reject) => {
+    const resolveNext = () => {
+      events.forEach((event) => emitter.off(event, resolveNext));
+      setImmediate(() => {
+        try {
+          resolve(next());
+        } catch (promiseError) {
+          reject(promiseError);
+        }
+      });
+    };
+    events.forEach((event) => emitter.once(event, resolveNext));
+  });
+}
+
+function turnFactory(readable, callback, onOff) {
+  return (event) => readable[onOff](event, callback);
+}
+
+module.exports = function eventsToAsyncIteratorFactory(readable, {
+  pauseThreshold = PAUSE_THRESHOLD,
+  resumeThreshold = RESUME_THRESHOLD,
+  closeEvents = CLOSE_EVENTS,
+  itemEvents = ITEM_EVENTS,
+  errorEvents = ERROR_EVENTS
+}) {
+  const events = [...itemEvents, ...errorEvents, ...closeEvents];
+  const highWaterMark = RESUME_THRESHOLD;
+  return function asyncIterator() {
+    if (!FixedQueue) {
+      FixedQueue = require('internal/fixed_queue');
+      setImmediate = require('timers').setImmediate;
+    }
+    const queue = new FixedQueue();
+    let done = false;
+    let error;
+    let queueSize = 0;
+    let paused = false;
+    const onError = (value) => {
+      turnOff();
+      error = value;
+    };
+    const onClose = () => {
+      turnOff();
+      done = true;
+    };
+    const onItem = (value) => {
+      queue.push(value);
+      queueSize++;
+      if (queueSize >= pauseThreshold) {
+        paused = true;
+        readable.pause();
+      }
+    };
+    const turnOff = () => {
+      closeEvents.forEach(turnFactory(readable, onClose, 'off'));
+      itemEvents.forEach(turnFactory(readable, onItem, 'off'));
+      errorEvents.forEach(turnFactory(readable, onError, 'off'));
+    };
+    itemEvents.forEach(turnFactory(readable, onItem, 'on'));
+    errorEvents.forEach(turnFactory(readable, onError, 'on'));
+    closeEvents.forEach(turnFactory(readable, onClose, '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);
+    }
+
+    const result = {
+      next,
+      highWaterMark,
+      get isPaused() {
+        return paused;
+      },
+      get queueSize() {
+        return queueSize;
+      }
+    };
+    result[SymbolAsyncIterator] = () => result;
+    return result;
+  };
+};

lib/internal/readline/interface.js

@@ -64,9 +64,6 @@ const {
 
 const { StringDecoder } = require('string_decoder');
 
-// Lazy load Readable for startup performance.
-let Readable;
-
 const kHistorySize = 30;
 const kMincrlfDelay = 100;
 // \r\n, \n, or \r followed by something other than \n
@@ -1185,41 +1182,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];
   }
 }
 

test/parallel/test-readline-async-iterators-backpressure.js

@@ -6,33 +6,41 @@ 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,
     crlfDelay: Infinity
   });
 
   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.
   assert(TOTAL_LINES > highWaterMark);
 
   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());

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());