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

Question

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

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管道是如何工作的：一个进程的stdout的同一个文件描述符被用作另一个进程的stdin。我试图避免通过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（相比于0％通过task0的标准输出作为任务1的标准输入时）：

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：这更好地说明了我的问题。我试图在原始代码中简化它，但我认为它太简化了。拉里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

现在，这是一个已知问题的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); 

的想法是启动发送任务之前推出的接收任务！

Answer 2

你原来的代码工作正常，如果你不等待第二个进程退出。

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不是。我不明白为什么这很重要，但很明显。可能与一个事实，即在节点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();