Skip to content

Commit a08a017

Browse files
davedoesdevMyles Borins
authored and
Myles Borins
committed
stream: ensure awaitDrain is increased once
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]>
1 parent b73ec46 commit a08a017

File tree

2 files changed

+36
-1
lines changed

2 files changed

+36
-1
lines changed

lib/_stream_readable.js

+8-1
Original file line numberDiff line numberDiff line change
@@ -530,11 +530,17 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
530530
ondrain();
531531
}
532532

533+
// If the user pushes more data while we're writing to dest then we'll end up
534+
// in ondata again. However, we only want to increase awaitDrain once because
535+
// dest will only emit one 'drain' event for the multiple writes.
536+
// => Introduce a guard on increasing awaitDrain.
537+
var increasedAwaitDrain = false;
533538
src.on('data', ondata);
534539
function ondata(chunk) {
535540
debug('ondata');
541+
increasedAwaitDrain = false;
536542
var ret = dest.write(chunk);
537-
if (false === ret) {
543+
if (false === ret && !increasedAwaitDrain) {
538544
// If the user unpiped during `dest.write()`, it is possible
539545
// to get stuck in a permanently paused state if that write
540546
// also returned false.
@@ -544,6 +550,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
544550
!cleanedUp) {
545551
debug('false write response, pause', src._readableState.awaitDrain);
546552
src._readableState.awaitDrain++;
553+
increasedAwaitDrain = true;
547554
}
548555
src.pause();
549556
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
'use strict';
2+
const common = require('../common');
3+
const stream = require('stream');
4+
5+
// A writable stream which pushes data onto the stream which pipes into it,
6+
// but only the first time it's written to. Since it's not paused at this time,
7+
// a second write will occur. If the pipe increases awaitDrain twice, we'll
8+
// never get subsequent chunks because 'drain' is only emitted once.
9+
const writable = new stream.Writable({
10+
write: common.mustCall((chunk, encoding, cb) => {
11+
if (chunk.length === 32 * 1024) { // first chunk
12+
readable.push(new Buffer(33 * 1024)); // above hwm
13+
}
14+
cb();
15+
}, 3)
16+
});
17+
18+
// A readable stream which produces two buffers.
19+
const bufs = [new Buffer(32 * 1024), new Buffer(33 * 1024)]; // above hwm
20+
const readable = new stream.Readable({
21+
read: function() {
22+
while (bufs.length > 0) {
23+
this.push(bufs.shift());
24+
}
25+
}
26+
});
27+
28+
readable.pipe(writable);

0 commit comments

Comments
 (0)