Repository navigation
Readable streams with highWaterMark: 0 terminate early #24915
Description
- Version: v10.10.0, v11.4.0
- Platform: Linux 4.4.0-45-generic deprecate domains #66-Ubuntu SMP Wed Oct 19 14:12:37 UTC 2016 x86_64 x86_64 x86_64 GNU/Linux
- Subsystem: Streams
#20503 fixed things with highWaterMark: 0 streams. As far as I could tell, this was merged into v10.9.0.
However the following code still fails in v10.10.0 and v11.4.0:
let util = require('util');
let Readable = require('stream').Readable;
function MyStream() {
Readable.call(this, { highWaterMark: 0 });
}
util.inherits(MyStream, Readable);
MyStream.prototype._read = function(n, fn) {
console.log(`_read`);
// Just keep emitting new data, but do that asynchronously.
process.nextTick( () => {
console.log('push');
this.push('a');
});
}
let s = new MyStream();
s.on('data', () => {
console.log('data');
});The MyStream implementation is supposed to continue emitting 'a' without terminating. However the output of that is:
_read
push
data
If I change the constructor into { highWaterMark: 1 } I get the expected infinite push/data cycle.
Activity
Rantanen commentedon Dec 9, 2018
The following is based on visual inspection on the code. I didn't actually debug through anything, so it could just as well be full of mistakes.
Looking further into this.
The case where
_readpushes data synchronously is taken care of by the while-loop inflow(stream). As long as_read()pushes data, thestream.read()will return non-null result and that while loop continues.If
_readpushes data asynchronously, it callsReadable.pushat some point. This results in the following call stack:Readable.push(chunk, encoding)calls...readableAddChunk(stream, chunk, ...)calls...addChunk(stream, state, ...)calls...maybeReadMore(stream, state)maybeReadMore_has a while-loop that callsstream.read(), but only ifstate.length < state.highWaterMark.As far as I can tell,
maybeReadMoreis used in both the flowing and paused modes. In paused mode that check seems correct: The stream should not pre-emptively request more data from the implementation but instead wait for calls to.read(). However in the flowing mode that check should take into account that it is responsible for providing data for thedataevent to emit.I guess the correct fix would be to calculate a
bufferLimitto account for the first item being consumed immediately:Or alternatively match the
addChunkif-check and add a similar condition to the while-loop inmaybeReadMore_:(Not sure whether the
state.syncis required)Rantanen commentedon Dec 9, 2018
Forcing
highWaterMark > 0is a really ugly workaround for this for the cases where the stream is coming from a library and it isn't possible to change thehighWaterMarkin the constructor.Rantanen commentedon Dec 9, 2018
... and I'm now looking into fixing this. Should have a PR up later today.
lib: ensure readable stream flows to end
Trott commentedon Dec 14, 2018
Fixed in 37a5e01
lib: ensure readable stream flows to end
lib: ensure readable stream flows to end
lib: ensure readable stream flows to end
lib: ensure readable stream flows to end
lib: ensure readable stream flows to end