No 'readable' events after unshift-ing more than highWaterMark bytes

[paleolitico]

In the following test, if toUnshift is less than 16_1024, the program writes messages without end, as pretended.
But if toUnshift is greater than or equal to 16_1024, the program stops after the first 'readable' event.

This program is the simplest I could write to exhibit this strange behaviour.
It occurred to me in a real program that accepts several protocols in the same port; these protocols can be optionally compressed and encrypted with ssl.
The program explores the first bytes to determine the protocol, unshifts the read data and jumps to the found protocol.

var net = require('net');

var toUnshift = 16*1024-1;

var buf = new Buffer(65536);
for (var i = 0; i < buf.length; i++) buf[i] = i;

var pr;
var server = net.createServer();
server.on('connection', function(c) {
  console.log('New connection');
  c.on('end', function() {console.log('Connection ended');});
  pr = processReadable.bind(null, c);
  c.on('readable', pr);
});
server.listen(12345, function() {console.log("Server bound");});

sendTo("localhost", 12345);

var first = true;
function processReadable(c) {
  console.info("Readable");
  var data = c.read();
  console.info("Read: " + data.length);
  if (first) {
    first = false;
    c.removeListener('readable', pr);
    c.unshift(buf.slice(0, toUnshift));
    c.on('readable', processReadable.bind(null,c));
    var delayedRead = function() {
      var data = c.read();
      console.info("Delayed read: " + data.length);
    };
    process.nextTick(delayedRead);
  }
}

function sendTo(host, port) {
  var s = net.connect(port, host);
  s.on('drain', startSender);
  startSender();

  function startSender() {process.nextTick(doSend);}

  function doSend() {
    while (s.write(buf));
  }
}

[rmg]

@pepagos what is the purpose of removing and re-adding the readable listener?

[paleolitico]

It is explained in the issue.
The code is a simplified scheme of a real code.
In the real code, a readable listener is removed and another different one is added.
But the bug is not related to adding the same function (in fact, for v8 is a different one because of the .bind).

[rmg]

@pepagos sorry, I've lost all context from when I asked that question. I have a local branch where I was working on a test and fix and wanted to confirm the use-case, but the state of where I was is now a mystery to me.

[jasnell]

@pepagos ... is this still an issue for you? can you test the latest v0.12.x and io.js?

[paleolitico]

Tested with node-v0.12.5 and iojs-v2.3.1.
The bug is still there.
Because I know the bug, my code avoids it.
But I think a bug will always be an issue for someone.

[jasnell]

Possibly related to #14604. Calling unshift does not reset the reading state.

[plafer]

After investigating, I found that the problem actually was related to removing and adding a listener to readable, unshifting in-between. Basically, removing a readable listener doesn't update readableListening state - it remains true. Unshifting then puts data back into the stream, and emits a readable... but no one is there to catch it. Then, when you re-attach a readable listener, the stream code goes right through and doesn't check to see if it should emit a readable event (or call a read(0)) because it still thinks a readable has been listening all along.

You didn't get any error when toUnshift < 16 * 1024 (the highWaterMark) because that would make the internal push call from the socket to the stream return true; the socket then wouldn't stop reading from the event loop. You would then add your readable listener back again, and move on. On the other hand, when toUnshift >= 16 * 1024, push would return false, the socket would stop reading from the event loop, and adding the readable listener back again would not trigger a _read... causing the program to hang.

I submitted a PR which solves this issue.
Note: in the PR, the setImmediate(emitReadable.bind(null, this)); is necessary to support cases where the "flow" would be managed by the user program, such as in test-stream3-pause-then-read.js