Second child process doesn't receive data when using the stdout of first child process as stdin of second

[aalexgabi]

• Version: v0.12.9 - v7.0.0
• Platform: Linux ${edited hostname} 4.4.0-45-generic deprecate domains #66~14.04.1-Ubuntu SMP Wed Oct 19 15:05:38 UTC 2016 x86_64 x86_64 x86_64 GNU/Linux

Also posted here: http://stackoverflow.com/questions/40306385/missing-lines-when-using-the-stdout-of-a-child-process-as-stdin-of-another

When using the stdout of one child process as stdin for another, it seems that sometimes data is not passed to the next child:

var spawn = require('child_process').spawn;

var pipeId = 0;
var pipeSlots = 6;

var launchProcess = function(cmd, args, stdin, stdout){
  return spawn(cmd, args, {
    stdio: [stdin, stdout, 'ignore']
  });
};

var launch = function(){
  var task0 = launchProcess('echo', ['how\nare\nyou\ndear\nstranger'], 'ignore', 'pipe');
  var task1 = launchProcess('tee', ['/tmp/body-pipeline-' + pipeId], task0.stdout, 'ignore');

  task0.on('error', function(err){
    console.log('Error while processing task0:' + err.stack);
  });
  task1.on('error', function(err){
    console.log('Error while processing task1:' + err.stack);
  });

  pipeId++;
};

// Simulating message queue
setInterval(function(){
  // Simulating how many messages we get from the messaging queue
  var mqMessageCount = Math.floor(Math.random() * (pipeSlots + 1));

  for(var i = 0; i < mqMessageCount; i++){
    launch();
  }
}, 250); // For this test we assume that pipes finish under 250ms

Some files are empty:

ls -lhS /tmp/body-pipeline-*

FYI: task0.stdout.pipe(task1.stdin) solves the issue but the script uses 50% CPU (compared to 0% when passing stdout of task0 as stdin of task1) for the equivalent of yes | tee /tmp//tmp/body-pipeline-x

[bnoordhuis]

You are handing off task0.stdout to the second child process but because task0.stdout starts in flowing mode and since it also exists in the parent process, data can end up in either process.

It might be possible to work around when you create the first child's stdout pipe manually in paused mode but that's probably difficult to do without groping around in node internals.

Perhaps spawn() needs to be taught a pauseOnCreate option, similar to new net.Socket({ handle: handle, pauseOnCreate: true }). Note that it may not be the best possible fix, just the first one that comes to mind.

[addaleax]

/cc @targos fyi this seems to be the right answer to nodejs/help#324, too

[mscdex]

The problem has to do with task0's stdout stream buffering all of the output before task1 has a chance to start and hook up to task0's stdout. It's racy due to timing/scheduling of the child processes, that is why it does not happen all of the time.

EDIT: Oops, didn't see the comments before this. I was just about to say what @bnoordhuis said. It works if pauseOnCreate: true is passed (internally) and you manually set task0.stdout.readable = false in the userland code in order for flushStdio() to not call stream.resume() which starts buffering data on the next tick.

[aalexgabi]

Is there a workaround that doesn't involve task0.stdout.pipe(task1.stdin)? I tried task0.stdout.readable = false but I don't know how to pass pauseOnCreate: true.

@bnoordhuis How to manually create a pipe in non flowing mode?

[bnoordhuis]

Is there a workaround that doesn't involve task0.stdout.pipe(task1.stdin)?

Not at the moment, no. Not without monkeying around with node's internals.

[aalexgabi]

Is there a way to use task0.stdout.pipe(task1.stdin) without passing all the data through NodeJS VM? I'm thinking of something like:

  var task0 = launchProcess('echo', ['how\nare\nyou\ndear\nstranger'], 'ignore', 'pipe');
  var fakePipe = new net.Socket({fd: task0.stdout._handle.fd})
  var task1 = launchProcess('tee', ['/tmp/body-pipeline-' + pipeId], fakePipe, 'ignore');

How would you go about monkey patching node's internals? Temporarily replacing net.createSocket with a version forcing readable to false?

I'm using NodeJS for an export runner that launches pipelines of scripts to process lots of data and upload it at the end so I would really like to find a way of doing this without the runner having very high CPU usage.

[targos]

Thanks @addaleax, it seems to be the same issue indeed!

What I don't understand here is that it happens even when the two calls to spawn are made in the same tick. Is task0.stdout somehow receiving data synchronously?

[bnoordhuis]

The file descriptor of task0.stdout exists simultaneously in the parent and the child. Calling task0.stdout.destroy() or task0.stdout.pause() in the parent right after spawning task1 might work around the issue for now.

[addaleax]

Calling task0.stdout.destroy() or task0.stdout.pause() in the parent right after spawning task1 might work around the issue for now.

I haven’t looked at it in detail, so ignore me if this doesn’t make sense, but is there any reason why the second spawn call shouldn’t do that itself when receiving task0.stdout?

[targos]

Calling task0.stdout.destroy() or task0.stdout.pause() in the parent right after spawning task1 might work around the issue for now.

Just tried both and it works with destroy but not with pause.

[aalexgabi]

A colleague of mine had a great idea that fixed this:

  var task1 = launchProcess('tee', ['/tmp/body-pipeline-' + pipeId], 'pipe', 'ignore');
  var task0 = launchProcess('echo', ['how\nare\nyou\ndear\nstranger'], 'ignore', task1.stdin);

The idea is to launch the receiving task before launching the sending task !

[masaeedu]

@bnoordhuis Is there a way to create a process with a predefined WritableStream for stdout? I mean the API itself seemed racy when I started using it, because I don't want to start a process and then hook up something to receive stdout. Can't see how to provide a stdout receiver before starting the process with the existing streams API though.