當使用一個子進程的標準輸出作爲另一個子進程的標準輸入時，數據有時不會傳遞給第二個子進程

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();