Daten werden manchmal nicht an den zweiten untergeordneten Prozess übergeben, wenn der stdout eines untergeordneten Prozesses als Stdin eines anderen verwendet wird

Question

Wenn der stdout eines untergeordneten Prozesses als stdin für einen anderen verwendet wird, werden Daten manchmal nicht an das nächste untergeordnete Element übergeben:

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

Einige Dateien sind leer:

ls -lhS /tmp/body-pipeline-* 

ich versuchte auch, als eine positive ganze Zahl mit dem Dateideskriptor vorbei task0.stdout._handle.fd Zugriff auf und das Problem weiterhin besteht.

So funktioniert Shell-Pipes meines Wissens: Derselbe Dateideskriptor für die Standardausgabe eines Prozesses wird als Standard einer anderen verwendet. Ich versuche zu vermeiden, alle Daten über den NodeJS-Prozess zu übergeben, da es hohe CPU-Lasten verursacht, wenn die untergeordneten Prozesse viele Daten ausgeben.

aktualisieren: Bei Rohrleitungen für beiden stdin verwendet werden und stdout alles wie erwartet funktioniert (hier mit Katze mit längerem Text zu testen):

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: Wenn task0.stdout.pipe(task1.stdin) das Skript verwendet 50 % CPU (im Vergleich zu 0% bei der Übergabe von stdout Task0 als stdin von task1):

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: Das veranschaulicht mein Problem besser. Ich habe versucht, es im ursprünglichen Code zu vereinfachen, aber ich denke, es war zu vereinfacht. Larry Turtis bot eine Lösung für den vereinfachten Fall, aber das gilt nicht für die Mine:

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

Dies ist nun eine bekannte NodeJS Ausgabe: https://github.com/nodejs/node/issues/9413

TLDR; Ein Kollege von mir hatte eine großartige Idee, dass dieses Problem behoben:

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

Die Idee ist, den Empfang Aufgabe zu starten, bevor die Senden Aufgabe starten!

Answer 2

Ihr ursprünglicher Code funktioniert gut, wenn Sie nicht für den zweiten Prozess warten zu beenden.

Was wahrscheinlich passiert ist, dass Task1 erledigt ist, aber Task0 nicht. Es ist mir nicht 100% klar, warum das wichtig ist, aber es ist klar. Möglicherweise im Zusammenhang mit der Tatsache, dass in dem Knoten documentation stellen wir fest, dass:

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

beide Aufgaben Gewährleistung vollständig das Problem behebt.

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