使用子进程的标准输出作为另一个子进程的标准输入时，数据有时不会传递给第二个子进程

Question

当使用一个子进程的标准输出作为另一个子进程的标准输入时，似乎有时数据不会传递给下一个子进程：

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

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

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

  pipeId++;

  task1.on('exit', launch);
};

launch();

有些文件是空的：

ls -lhS /tmp/body-pipeline-*

我还尝试通过访问task0.stdout._handle.fd 将文件描述符作为正整数传递，但问题仍然存在。

据我所知，这就是 shell 管道的工作方式：一个进程的标准输出的相同文件描述符被用作另一个进程的标准输入。我试图避免通过 NodeJS 进程传递所有数据，因为当子进程输出大量数据时，它会导致 CPU 负载过高。

更新：当管道同时用于标准输入和标准输出时，一切都按预期工作（在此处使用 cat 测试更长的文本）：

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

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

var launch = function(){
  var task0 = launchProcess('cat');
  var task1 = launchProcess('tee', ['/tmp/body-pipeline-' + pipeId]);

  task0.stdout.pipe(task1.stdin)

  task0.stdin.write(JSON.stringify(process.env).split(',').join('\n'))
  task0.stdin.end();

  pipeId++;

  task1.on('exit', launch);
};

launch();

Update2：当使用task0.stdout.pipe(task1.stdin) 时，脚本使用 50% CPU（相比之下，将 task0 的标准输出作为 task1 的标准输入时为 0%）：

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

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

var launch = function(){
  var task0 = launchProcess('yes', ['lala'], 'ignore', 'pipe', 'ignore');
  var task1 = launchProcess('tee', ['/tmp/body-pipeline-' + pipeId], 'pipe', 'ignore', 'ignore');
  // var task1 = launchProcess('tee', ['/tmp/body-pipeline-' + pipeId], task0.stdout, 'ignore', 'ignore');


  task0.stdout.pipe(task1.stdin);

  pipeId++;

  task1.on('exit', launch);
};

launch();

Update3：这更好地说明了我的问题。我试图在原始代码中简化它，但我认为它过于简化了。 Larry Turtis 为简化案例提供了一种解决方法，但不适用于我的情况：

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

Answer 1

如果您不等待第二个进程退出，您的原始代码可以正常工作。

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

  pipeId++;

  launch();
};

可能发生的是 task1 已完成，但 task0 未完成。我不是 100% 清楚为什么这很重要，但很明显。可能与在节点documentation 中我们注意到：

..when the 'exit' event is triggered, child process stdio streams might still be open.

确保两个任务都完成可以解决问题。

var spawn = require('child_process').spawn;
var q = require("q");
var pipeId = 0;

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

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

    var p0 = q.defer();
    var p1 = q.defer();
    task0.on('exit', p0.resolve);
    task1.on('exit',p1.resolve);

    q.all(p0, p1).then(launch)

    pipeId++;
};

launch();

Answer 2

这是一个已知的 NodeJS 问题：https://github.com/nodejs/node/issues/9413

TLDR;我的一位同事提出了一个解决此问题的好主意：

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

想法是在启动发送任务之前启动接收任务！