Readable.push calling _read

[MrHacky]

In the documentation for stream.Readable, 'API for Stream Implementors', the documentation states that push should continue to be called until it returns false. However when calling push from an async callback each push call will call _read, leading to unexpected behaviour.

Example:

var stream = require('stream');
var util = require('util');

util.inherits(TestStream, stream.Readable);
function TestStream(options) {
    stream.Readable.call(this, options);
}

TestStream.prototype._read = function(n) {
    var mthis = this;
    setTimeout(function() {
        for (var i = 0; i < 5; ++i)
            if (!mthis.push('x'))
                break;
    }, 2000);
};

var s = new TestStream();
s.pipe(process.stdout);

I expect this to output up to 5 'x' characters every 2 seconds. However, because each push call also does a _read call, after 2 seconds 5 callbacks will be registered and after 2 more seconds 25 'x' characters will be output.

I think push should not call _read like this, or at least return false after it has done so.

[brendanashworth]

I think you're mistaken about the way your readable stream is working. Because you're piping it into a writable stream (process.stdout), it will suck all the data it can out of your stream as long as the highWaterMark of your writable stream is not hit. This means that it will call ._read repeatedly - not because .push('x') triggers more reads.

For example, replacing 5 with n will read out the requested amount of data, 16384 bytes.

[MrHacky]

Replacing '5' with 'n' will still have each push call do a _read call. If you pipe the output to a file you will see it will grow much faster than 16k every 2 seconds.

I understand _read will keep getting called becuase i am doing a 'infinite' pipe. But the docs say _read will not be called until i call push. The docs also say i should keep calling push until it returns false. From this i expect _read should be called after push returned false. It doesn't make sense to start requesting more data by calling _read while at the same time indicating i should call push again by returning true from push.

Not this is exactly how it works when not using async calls, but the non-direct push call somehow behaves differently...

[brendanashworth]

the docs say _read will not be called until i call push

Could you point me to where it says this? I don't see this anywhere.

The docs also say i should keep calling push until it returns false.

Yes, but this is with regards to the highWaterMark, which isn't really relevant to this use case.

Is this an actual issue for you? If your stream needs to load data, you just shouldn't push anything when there isn't data available.

[MrHacky]

Could you point me to where it says this? I don't see this anywhere.

Quoted from https://nodejs.org/api/stream.html#stream_readable_read_size_1
Note: once the _read() method is called, it will not be called again until the push method is called.

Is this an actual issue for you? If your stream needs to load data, you just shouldn't push anything when there isn't data available.

Yeah, i'm trying to wrap a source of data into a readable stream, a less abstract example:

.prototype._read = function(n) {
    this.myread();
}
.prototype.myread() {
    while (/*source has data available*/ data = this.source.getdata())
        if (!this.push(data))
            return; /* highWaterMark is reached; Stream.Readable will call _read again when it needs more data*/

    /* source ran out of data but Stream.Readable wants more (highWaterMark was not reached) 
       , attach an event to continue to push data when it arrives
    */
    var mthis = this;
    this.source.once('dataready', function() { mthis.myread(); });
}

The problem happens when the dataready event triggers an async call to myread and thus push. push starts to behave differently and each async call to push calls _read which triggers more push calls and/or event callback registrations. In this case the only observable effect is everything gets slower and slower as more and more callbacks are registered and called.

Also, what is wrong with the code in my first post? Why shouldn't it be a stream that produces 5 x-es every 2 seconds?

[Trott]

/cc @nodejs/streams Is this a bug? Or a misunderstanding? Maybe a documentation update is in order?

[chrisdickinson]

.push can trigger ._read (via maybeReadMore and here) in order to fill the readable stream's buffer to hwm, which might be what's going on here.

[mself]

I'm having the same problem. When _read() is called, I fetch data using an async API. I then start pushing the results using push(). But as soon as I push the first object to the stream, the stream.on('data') gets called (consuming the object), which then causes _read() to get called again even though I'm still in the middle of pushing the previous results. This results in the API getting called again even though I still have data to push from the last call.

[mcollina]

The current behaviour guarantees the maximum performance and throughput.
The reason Readable behaves in this way it is to start all the I/O necessary to fill the hwm before any other processing. Changing this behaviour will slow things down.

[MrHacky]

I'm sure my examples failed to communicate the problem properly, but defending buggy behavior with performance claims seems weird to me...

It's fine to call _read when new data is actually needed. But to do this inside the push() call doesn't make sense as the push return value is already going to indicate whether more data is needed to the caller.

In other words, if I'm calling push, and it's return value is indicating I should keep doing so, I don't need/want _read being called asking for even more data (I'm already push()-ing it as fast as I can)

The thing is, that it behaves exactly like this the way I want as long as you call push directly from _read, but once you do so indirectly through an async callback it behaves completely different. (this is at the very least a documentation issue as this difference is never mentioned anywhere, but IMHO it just doesn't make any sense...)

[mcollina]

I am not "defending buggy behavior", I am just explaining why things are done in a certain way. Let me know if you need some more discussion on why it is faster this way.

Even though I might even agree with you that we should change this behavior, there is a massive risk of breakage in the ecosystem. Anyway, I do no think this a road to pursue, but I am very happy in reviewing any proposed changes in streams.

Improving the docs is certainly something we need! Feel free to send a PR for that!

[mself]

One thing that might help is if there was a way to push() a set of objects in a single call. Right now, after you push the first item (from an async function) there will be an immediate call to _read() which is what can cause issues.

Instead, I was thinking there could be a pushObjects() API that would let you push in all of the new objects at once. This call would also generate a call to _read(), but only after the call to pushObjects() is done. I think this would avoid the race condition and allow for simpler code.

The alternative I ended up having to build was to keep my own buffer of objects. After the async call adds more objects to this array, it then shift()s the first one, push()es it, and then returns. After this, _read() will be called, and it push()es the rest of the objects synchronously from the buffer.

This worked fine, but was somewhat complicated. It also requires an additional buffer of objects, which seems silly since the point of the stream is to hold the buffer.

I think this change might be possible in a backwards-compatible way.

[mself]

I found a good solution for this problem, which is to pause the stream while pushing in a set of objects that come from an async API call. Here is the code that I ended up with:

MyStream.prototype._read = function() {
	const self = this;

	// Get a set of objects from an API.
	this.fetchObjects(function(err, objects) {
		// Pause the stream to avoid race conditions while pushing in the new objects.
		// Without this, _read() would be called again from inside each push(),
		// resulting in multiple parallel calls to fetchObjects().
		const wasPaused = self.isPaused();
		self.pause();

		// Push all of the objects into the stream.
		objects.forEach(function(object) {
			self.push(object);
		});

		if (!wasPaused) {
			// This will deliver the objects and trigger the next call to _read() once they have been consumed.
			self.resume();
		}
	});
};

[mself]

As I mentioned above, it would be great if the Readable class included a pushObjects() method that essentially encapsulated this. It wouldn't necessarily really need to pause the stream, but it would need to ensure that no calls to _read() are made until all of the objects have been pushed into the stream's buffer. Then the code would be:

MyStream.prototype._read = function() {
	const self = this;

	// Get a set of objects from an API.
	this.fetchObjects(function(err, objects) {
		// Push all of the objects into the stream.
		self.pushObjects(objects);
	});
};

Much simpler and less error-prone.