Skip to content

Commit cd2b9f5

Browse files
committed
stream: Avoid nextTick warning filling read buffer
In the function that pre-emptively fills the Readable queue, it relies on a recursion through: stream.push(chunk) -> maybeReadMore(stream, state) -> if (not reading more and < hwm) stream.read(0) -> stream._read() -> stream.push(chunk) -> repeat. Since this was only calling read() a single time, and then relying on a future nextTick to collect more data, it ends up causing a nextTick recursion error (and potentially a RangeError, even) if you have a very high highWaterMark, and are getting very small chunks pushed synchronously in _read (as happens with TLS, or many simple test streams). This change implements a new approach, so that read(0) is called repeatedly as long as it is effective (that is, the length keeps increasing), and thus quickly fills up the buffer for streams such as these, without any stacks overflowing.
1 parent 738347b commit cd2b9f5

2 files changed

Lines changed: 22 additions & 5 deletions

File tree

lib/_stream_readable.js

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -386,17 +386,23 @@ function maybeReadMore(stream, state) {
386386
if (!state.readingMore) {
387387
state.readingMore = true;
388388
process.nextTick(function() {
389-
state.readingMore = false;
390389
maybeReadMore_(stream, state);
391390
});
392391
}
393392
}
394393

395394
function maybeReadMore_(stream, state) {
396-
if (!state.reading && !state.flowing && !state.ended &&
397-
state.length < state.highWaterMark) {
395+
var len = state.length;
396+
while (!state.reading && !state.flowing && !state.ended &&
397+
state.length < state.highWaterMark) {
398398
stream.read(0);
399+
if (len === state.length)
400+
// didn't get any data, stop spinning.
401+
break;
402+
else
403+
len = state.length;
399404
}
405+
state.readingMore = false;
400406
}
401407

402408
// abstract method. to be overridden in specific implementation classes.

test/simple/test-stream2-unpipe-leak.js

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ var common = require('../common.js');
2424
var assert = require('assert');
2525
var stream = require('stream');
2626

27+
var chunk = new Buffer('hallo');
28+
2729
var util = require('util');
2830

2931
function TestWriter() {
@@ -37,13 +39,15 @@ TestWriter.prototype._write = function(buffer, encoding, callback) {
3739

3840
var dest = new TestWriter();
3941

42+
// Set this high so that we'd trigger a nextTick warning
43+
// and/or RangeError if we do maybeReadMore wrong.
4044
function TestReader() {
41-
stream.Readable.call(this);
45+
stream.Readable.call(this, { highWaterMark: 0x10000 });
4246
}
4347
util.inherits(TestReader, stream.Readable);
4448

4549
TestReader.prototype._read = function(size) {
46-
this.push(new Buffer('hallo'));
50+
this.push(chunk);
4751
};
4852

4953
var src = new TestReader();
@@ -61,3 +65,10 @@ assert.equal(dest.listeners('drain').length, 0);
6165
assert.equal(dest.listeners('error').length, 0);
6266
assert.equal(dest.listeners('close').length, 0);
6367
assert.equal(dest.listeners('finish').length, 0);
68+
69+
console.error(src._readableState);
70+
process.on('exit', function() {
71+
assert(src._readableState.length >= src._readableState.highWaterMark);
72+
src._readableState.buffer.length = 0;
73+
console.error(src._readableState);
74+
});

0 commit comments

Comments
 (0)