stream: always reset awaitDrain when emitting data

The complicated `awaitDrain` machinery can be made a bit
slimmer, and more correct, by just resetting the value
each time `stream.emit('data')` is called.

By resetting the value before emitting the data chunk, and
seeing whether any pipe destinations return `.write() === false`,
we always end up in a consistent state and don’t need to worry
about odd situations (like `dest.write(chunk)` emitting more data).

PR-URL: nodejs#18516
Fixes: nodejs#18484
Fixes: nodejs#18512
Refs: nodejs#18515
Reviewed-By: Anatoli Papirovski <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Minwoo Jung <[email protected]>
Reviewed-By: Ruben Bridgewater <[email protected]>

lib/_stream_readable.js

@@ -258,6 +258,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
 
 function addChunk(stream, state, chunk, addToFront) {
   if (state.flowing && state.length === 0 && !state.sync) {
+    state.awaitDrain = 0;
     stream.emit('data', chunk);
   } else {
     // update the buffer info.
@@ -456,6 +457,7 @@ Readable.prototype.read = function(n) {
     n = 0;
   } else {
     state.length -= n;
+    state.awaitDrain = 0;
   }
 
   if (state.length === 0) {
@@ -637,18 +639,12 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       ondrain();
   }
 
-  // If the user pushes more data while we're writing to dest then we'll end up
-  // in ondata again. However, we only want to increase awaitDrain once because
-  // dest will only emit one 'drain' event for the multiple writes.
-  // => Introduce a guard on increasing awaitDrain.
-  var increasedAwaitDrain = false;
   src.on('data', ondata);
   function ondata(chunk) {
     debug('ondata');
-    increasedAwaitDrain = false;
     var ret = dest.write(chunk);
     debug('dest.write', ret);
-    if (false === ret && !increasedAwaitDrain) {
+    if (ret === false) {
       // If the user unpiped during `dest.write()`, it is possible
       // to get stuck in a permanently paused state if that write
       // also returned false.
@@ -658,7 +654,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
           !cleanedUp) {
         debug('false write response, pause', state.awaitDrain);
         state.awaitDrain++;
-        increasedAwaitDrain = true;
       }
       src.pause();
     }
@@ -834,7 +829,6 @@ function resume_(stream, state) {
   }
 
   state.resumeScheduled = false;
-  state.awaitDrain = 0;
   stream.emit('resume');
   flow(stream);
   if (state.flowing && !state.reading)

test/parallel/test-stream-pipe-manual-resume.js

@@ -0,0 +1,35 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+
+function test(throwCodeInbetween) {
+  // Check that a pipe does not stall if .read() is called unexpectedly
+  // (i.e. the stream is not resumed by the pipe).
+
+  const n = 1000;
+  let counter = n;
+  const rs = stream.Readable({
+    objectMode: true,
+    read: common.mustCallAtLeast(() => {
+      if (--counter >= 0)
+        rs.push({ counter });
+      else
+        rs.push(null);
+    }, n)
+  });
+
+  const ws = stream.Writable({
+    objectMode: true,
+    write: common.mustCall((data, enc, cb) => {
+      setImmediate(cb);
+    }, n)
+  });
+
+  setImmediate(() => throwCodeInbetween(rs, ws));
+
+  rs.pipe(ws);
+}
+
+test((rs) => rs.read());
+test((rs) => rs.resume());
+test(() => 0);