stream: sync resume on drain

benchmark/streams/pipe-backpressure.js

@@ -0,0 +1,24 @@
+'use strict';
+
+const common = require('../common');
+const { Readable, Writable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  n: [5e6],
+});
+
+function main({ n }) {
+  const b = Buffer.alloc(1024);
+  const r = new Readable();
+  const w = new Writable();
+
+  let i = 0;
+
+  r._read = () => r.push(i++ === n ? null : b);
+  w._write = (data, enc, cb) => queueMicrotask(cb);
+
+  bench.start();
+
+  r.pipe(w);
+  w.on('finish', () => bench.end(n));
+}

lib/internal/streams/readable.js

@@ -1091,7 +1091,7 @@
 
     if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) &&
       (state[kState] & kDataListening) !== 0) {
-      src.resume();
+      resume(stream, state, dest.listenerCount('drain') <= 1);
     }
   };
 }
@@ -1232,7 +1232,10 @@
 // pause() and resume() are remnants of the legacy readable stream API
 // If the user uses them, then switch into old mode.
 Readable.prototype.resume = function() {
-  const state = this._readableState;
+  return resume(this, this._readableState, false);
+};
+
+function resume(stream, state, sync) {
   if ((state[kState] & kFlowing) === 0) {
     debug('resume');
     // We flow only if there is no one listening
@@ -1244,18 +1247,18 @@
     } else {
       state[kState] &= ~kFlowing;
     }
-    resume(this, state);
+    if ((state[kState] & kResumeScheduled) === 0) {
+      if (sync) {
+        resume_(stream, state, false);
+      } else {
+        state[kState] |= kResumeScheduled;
+        process.nextTick(resume_, stream, state);
+      }
+    }
   }
   state[kState] |= kHasPaused;
   state[kState] &= ~kPaused;
-  return this;
-};
-
-function resume(stream, state) {
-  if ((state[kState] & kResumeScheduled) === 0) {
-    state[kState] |= kResumeScheduled;
-    process.nextTick(resume_, stream, state);
-  }
+  return stream;
 }
 
 function resume_(stream, state) {