Readable streams with highWaterMark: 0 terminate early

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.

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 _read pushes data synchronously is taken care of by the while-loop in flow(stream). As long as _read() pushes data, the stream.read() will return non-null result and that while loop continues.

If _read pushes data asynchronously, it calls Readable.push at 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 calls stream.read(), but only if state.length < state.highWaterMark.

As far as I can tell, maybeReadMore is 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 the data event to emit.

I guess the correct fix would be to calculate a bufferLimit to account for the first item being consumed immediately:

let bufferLimit = state.highWaterMark;
if (state.flowing)
    bufferLimit += 1;

Or alternatively match the addChunk if-check and add a similar condition to the while-loop in maybeReadMore_:

  while (!state.reading && !state.ended &&
         (state.length < state.highWaterMark ||
         state.flowing && state.length === 0 && !state.sync)) {

(Not sure whether the state.sync is required)

Forcing highWaterMark > 0 is a really ugly workaround for this for the cases where the stream is coming from a library and it isn't possible to change the highWaterMark in the constructor.

someStream._readableState.highWaterMark = 1;

... and I'm now looking into fixing this. Should have a PR up later today.

lib: ensure readable stream flows to end

Fixed in 37a5e01

lib: ensure readable stream flows to end

Navigation Menu

Repository navigation

Readable streams with highWaterMark: 0 terminate early #24915

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Readable streams with highWaterMark: 0 terminate early #24915

Description

Activity

Rantanen commented on Dec 9, 2018

Rantanen commented on Dec 9, 2018

Rantanen commented on Dec 9, 2018

Trott commented on Dec 14, 2018

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions