child_process: close pipe ends that are re-piped

lib/internal/child_process.js

@@ -384,6 +384,12 @@ ChildProcess.prototype.spawn = function(options) {
       continue;
     }
 
+    // stream is already cloned and piped, so close
+    if (stream.type === 'wrap') {
+      stream.handle.close();
+      continue;
+    }
+
     if (stream.handle) {
       // when i === 0 - we're dealing with stdin
       // (which is the only one writable pipe)

test/parallel/test-child-process-pipe-dataflow.js

@@ -0,0 +1,51 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const path = require('path');
+const fs = require('fs');
+const spawn = require('child_process').spawn;
+const tmpdir = require('../common/tmpdir');
+
+let cat, grep, wc;
+
+const KB = 1024;
+const MB = KB * KB;
+
+
+// Make sure process chaining allows desired data flow:
+// check cat <file> | grep 'x' | wc -c === 1MB
+// This helps to make sure no data is lost between pipes.
+
+{
+  tmpdir.refresh();
+  const file = path.resolve(tmpdir.path, 'data.txt');
+  const buf = Buffer.alloc(MB).fill('x');
+
+  // Most OS commands that deal with data, attach special
+  // meanings to new line - for example, line buffering.
+  // So cut the buffer into lines at some points, forcing
+  // data flow to be split in the stream.
+  for (let i = 0; i < KB; i++)
+    buf[i * KB] = 10;
+  fs.writeFileSync(file, buf.toString());
+
+  cat = spawn('cat', [file]);
+  grep = spawn('grep', ['x'], { stdio: [cat.stdout, 'pipe', 'pipe'] });
+  wc = spawn('wc', ['-c'], { stdio: [grep.stdout, 'pipe', 'pipe'] });
+
+  wc.stdout.on('data', common.mustCall(function(data) {
+    assert.strictEqual(data.toString().trim(), MB.toString());
+  }));
+
+  cat.on('exit', common.mustCall(function(code) {
+    assert.strictEqual(code, 0);
+  }));
+
+  grep.on('exit', common.mustCall(function(code) {
+    assert.strictEqual(code, 0);
+  }));
+
+  wc.on('exit', common.mustCall(function(code) {
+    assert.strictEqual(code, 0);
+  }));
+}