1berry/nodejs
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

stream: always reset awaitDrain when emitting data

Browse Source 
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: https://github.com/nodejs/node/pull/18516
Fixes: https://github.com/nodejs/node/issues/18484
Fixes: https://github.com/nodejs/node/issues/18512
Refs: https://github.com/nodejs/node/pull/18515
Reviewed-By: Anatoli Papirovski <apapirovski@mac.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Minwoo Jung <minwoo@nodesource.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
This commit is contained in:
Anna Henningsen 2018-02-02 01:41:35 +01:00
parent 610cac26f0
commit e7cb694a60
No known key found for this signature in database
GPG Key ID: 9C63F3A6CD2AD8F9
2 changed files with 38 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
lib/_stream_readable.js
View File

@ -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)

35
test/parallel/test-stream-pipe-manual-resume.js Normal file
View File

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