Hanging streams due to internal pause() without corresponding resume() during pipe()

[patrick-steele-idem]

After switching from Node.js v0.10.31 to v0.11.13 I started noticing that some streaming operations are hanging. I tracked it down and found that it appears to be due to the introduction of an internal and automatic src.pause() during piping without a corresponding src.resume(). The following src.pause() line is indicated below:

  src.on('data', ondata);
  function ondata(chunk) {
    debug('ondata');
    var ret = dest.write(chunk);
    if (false === ret) {
      debug('false write response, pause',
            src._readableState.awaitDrain);
      src._readableState.awaitDrain++;
      src.pause(); // <-- *** PROBLEMATIC LINE ***
    }
  }

Please see: _stream_readable.js:540

It makes perfect sense that a source stream would be paused if the destination stream says that its buffer is full during the write() call. However, what doesn't make sense is that there is no corresponding resume() when the destination stream is drained. This leaves the source stream in a permanent paused state. Instead of calling resume() when the destination stream is drained I see that only the flowing state is set to true:
_stream_readable.js:601

How is a source Readable stream supposed to know that it is supposed to start producing data again? I tried to workaround around this problem by calling resume() in my implementation of _read() but I then found that automatic pause() was being called after the last _read() call for some unknown reason.

This seems like a bug to me, but maybe I am missing something. In the meantime I had to disable pausing of the stream as a workaround.

The following commit by @isaacs appears to have introduced this problem: 0f8de5e

Based on my research so far I see a few options:

---

Option 1) Add a corresponding src.resume() when the destination stream is drained.

That is, replace state.flowing = true; with src.resume() at the following line:
_stream_readable.js:601

---

Option 2) Don't use pause() to control back-pressure

From the docs it appears that the return value of this.push() should be used to control back-pressure.

---

Option 3) Some how differentiate between between an internal pause() (done as a result of back-pressure) and an outside pause() (that may been done for some unknown outside reason).

---

I can work on isolating out my code to help you reproduce the problem, but it will take some time. In the mean time, maybe someone can provide some additional insights or suggested fixes?

Edit:

Further clarification: in my case I am creating a Readable stream that wraps one or more underlying Readable streams. When the wrapping stream is paused, the underlying wrapped streams are paused. When pause() is called on the wrapping stream, but there is no subsequent resume() then the wrapped stream remains permanently paused.

[vkurchatkin]

@patrick-steele-idem do you have some sort of a test case to demonstrate the issue?

[patrick-steele-idem]

I spent some more time on this issue and I seem to have also uncovered a significant performance regression, as well as some other bug related to streaming. Here's a program to illustrate the problem:

var inherits = require('util').inherits;
var Readable = require('stream').Readable;
var fs = require('fs');

process.on('exit', function(code) {
    console.log('\nExiting with code ', code);
});

// Helper Readable stream that just writes out numbers 0-9 up to the given max number of times
function MyReadable(max) {
    Readable.call(this, { encoding: 'utf8' });

    this.count = 0;
    this.max = max;
}

inherits(MyReadable, Readable);

MyReadable.prototype._read = function() {
    while (true) {
        if (this.count === this.max) {
            this.push('\n');
            this.push(null);
            break;
        }

        this.count++;

        if (this.push((this.count % 10).toString()) === false) {
            break;
        }
    }
};

// *** WORKING IMPLEMENTATION ***
function createCombinedStreamGood() {
    var combined = new Readable();

    var streams = [];
    var readStarted = false;

    var curStream;
    var i = -1;
    var len;
    var paused = false;

    function onData(chunk) {
        if (combined.push(chunk) === false) {
            // If backpressure was applied then
            // lets pause the underlying stream
            paused = true;
            curStream.pause();
            console.log('Pausing the underlying stream...');
        }
    }

    function onError(err) {
        combined.emit('error', err);
    }

    function next() {

        if (++i >= len) {
            // we're done
            combined.push(null);
        } else {
            curStream = streams[i];
            curStream.on('end', next);
            curStream.on('data', onData);
            curStream.on('error', onError);
        }
    }

    combined._read = function() {
        if (readStarted) {
            if (paused) {
                console.log('Resuming the underlying stream...');
                // If backpressure was applied then we previously
                // paused the underlying stream. Therefore, we need
                // to resume it now
                paused = false;
                curStream.resume();
            }
        } else {
            readStarted = true;

            len = streams.length;

            if (len === 0) {
                combined.push(null);
                return;
            } else {
                next();
            }
        }
    };

    combined.addStream = function(stream) {
        streams.push(stream);
    };

    return combined;
}


// *** BUGGY IMPLEMENTATION ***
function createCombinedStreamBad() {
    var combined = new Readable();
    var streams = [];
    var readStarted = false;
    var curStream;
    var i = -1;
    var len;

    function onData(chunk) {
        combined.push(chunk);
    }

    function onError(err) {
        combined.emit('error', err);
    }

    function next() {
        if (++i >= len) {
            // we're done
            combined.push(null);
        } else {
            curStream = streams[i];
            curStream.on('end', next);
            curStream.on('data', onData);
            curStream.on('error', onError);
        }
    }

    combined._read = function() {
        if (!readStarted) {
            readStarted = true;

            len = streams.length;

            if (len === 0) {
                combined.push(null);
                return;
            }

            next();
        }
    };


    combined.pause = function(stream) {
        // Pause the current underlying stream if this stream is paused
        if (curStream) {
            curStream.pause();
        }

        Readable.prototype.pause.call(this);
    };

    combined.resume = function(stream) {
        // Resume the current underlying stream if this stream is resumed
        if (curStream) {
            curStream.resume();
        }

        Readable.prototype.resume.call(this);
    };

    combined.addStream = function(stream) {
        streams.push(stream);
    };

    return combined;
}


function run(name, callback) {
    var startTime = Date.now();

    console.log('\nRunning (' + name + ')...');

    var combinedStream = name === 'good' ?
        createCombinedStreamGood() :
        createCombinedStreamBad();

    combinedStream.addStream(new MyReadable(10000));
    combinedStream.addStream(new MyReadable(10000));
    combinedStream.addStream(new MyReadable(10000));

    var outFile = 'out-' + name + '.txt';
    console.log('Writing to "' + outFile + '"...');

    combinedStream
        .on('end', function() {
            console.log('Combined stream ended for "' + name + '"');
        })
        .on('error', function(err) {
            console.error('ERROR: ', err);
        })
        .pipe(fs.createWriteStream(outFile, 'utf8'))
        .on('error', function(err) {
            console.error('ERROR: ', err);
        })
        .on('close', function() {
            var delta = Date.now() - startTime;
            console.log('Completed writing to "' + outFile + '" in ' + delta + 'ms!');
            callback();
        });
}

// Run the "good" version and then the "bad" version (one after another)
run('good', function() {
    run('bad', function() {
        console.log('\nAll done!');
    });
});

Running the above program on Node.js 0.10.31 produces the correct results for both the "good" and "bad" implementation. You should see the following output when running the program:

Running (good)...
Writing to "out-good.txt"...
Pausing the underlying stream...
Resuming the underlying stream...
Combined stream ended for "good"
Completed writing to "out-good.txt" in 29ms!

Running (bad)...
Writing to "out-bad.txt"...
Combined stream ended for "bad"
Completed writing to "out-bad.txt" in 7ms!

All done!

Exiting with code  0

Running the above program on Node.js v0.10.31 produces the correct results for the "good" implementation, but for some reason the program just exits before the stream is ever ended in the "bad" implementation. Not only that, the program runs much, much slower on Node.js v0.11.13 You should see the following output when running the program using Node.js v0.11.13:

Running (good)...
Writing to "out-good.txt"...
Combined stream ended for "good"
Completed writing to "out-good.txt" in 1573ms!

Running (bad)...
Writing to "out-bad.txt"...

Exiting with code  0

In addition, to the different output, also notice the huge time difference for the "good" run!: 1573ms versus 29ms

When running the program on Node.js v0.11.13 I noticed that combined.push(chunk) never returns false for the good version unless I really bump up the max number bytes for the producer to produce. Have the thresholds changed for back pressure in the latest Node.js?

[vkurchatkin]

@patrick-steele-idem the problem is that you are overwriting resume and pause, but you are not supposed to do so. Your code was relying on implementation detail. You should use return value of push to pause underlying stream and _read to resume.

What about performance, I think it's due general buffer performance degradation, see #7633. It's better on master, though not much.

[patrick-steele-idem]

I agree that pause() and resume() should not be overridden, and I think the "good" version is better for that reason. However, the Node.js code still feels wrong to me, and I believe that if the code is going to call src.pause() then I still think there should be a corresponding src.resume(). I can live with that issue and I have updated my code accordingly. Ignoring that issue for now...

How about the other issue? Why is the "bad" version exiting early and never fully writing to the output file? There is no close event being emitted and the program is mysteriously exiting. That seems like a symptom of a possibly bigger problem.

The performance problem is concerning, but hopefully it has been resolved on master.

[vkurchatkin]

@patrick-steele-idem I see no problem here. The stream has been paused and it's not going to be resumed, nothing is going to happen, so the program exits.

[patrick-steele-idem]

Thanks @vkurchatkin. Seems reasonable about the process exiting after giving it some additional thought.

I'm okay with closing this issue and hopefully no one else will think it is a good idea to override either the pause() or resume() method. All my tests are passing after making the necessary changes.

Thanks again.