Skip to content

Commit

Permalink
stream: ensure awaitDrain is increased once
Browse files Browse the repository at this point in the history
Guard against the call to write() inside pipe's ondata pushing more data
back onto the Readable, thus causing ondata to be called again.

This is fine but results in awaitDrain being increased more than once.
The problem with that is when the destination does drain, only a single
'drain' event is emitted, so awaitDrain in this case will never reach
zero and we end up with a permanently paused stream.

Fixes: #7278
PR-URL: #7292
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>
  • Loading branch information
davedoesdev authored and evanlucas committed Jun 16, 2016
1 parent ad1045c commit b0225e5
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 1 deletion.
9 changes: 8 additions & 1 deletion lib/_stream_readable.js
Original file line number Diff line number Diff line change
Expand Up @@ -549,11 +549,17 @@ 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);
if (false === ret) {
if (false === ret && !increasedAwaitDrain) {
// If the user unpiped during `dest.write()`, it is possible
// to get stuck in a permanently paused state if that write
// also returned false.
Expand All @@ -563,6 +569,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
!cleanedUp) {
debug('false write response, pause', src._readableState.awaitDrain);
src._readableState.awaitDrain++;
increasedAwaitDrain = true;
}
src.pause();
}
Expand Down
28 changes: 28 additions & 0 deletions test/parallel/test-stream-pipe-await-drain-push-while-write.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
'use strict';
const common = require('../common');
const stream = require('stream');

// A writable stream which pushes data onto the stream which pipes into it,
// but only the first time it's written to. Since it's not paused at this time,
// a second write will occur. If the pipe increases awaitDrain twice, we'll
// never get subsequent chunks because 'drain' is only emitted once.
const writable = new stream.Writable({
write: common.mustCall((chunk, encoding, cb) => {
if (chunk.length === 32 * 1024) { // first chunk
readable.push(new Buffer(33 * 1024)); // above hwm
}
cb();
}, 3)
});

// A readable stream which produces two buffers.
const bufs = [new Buffer(32 * 1024), new Buffer(33 * 1024)]; // above hwm
const readable = new stream.Readable({
read: function() {
while (bufs.length > 0) {
this.push(bufs.shift());
}
}
});

readable.pipe(writable);

0 comments on commit b0225e5

Please sign in to comment.