Optimize child_process IPC for large data

benchmark/child_process/child-process-read-ipc.js

@@ -1,14 +1,20 @@
 'use strict';
 if (process.argv[2] === 'child') {
   const len = +process.argv[3];
-  const msg = `"${'.'.repeat(len)}"`;
-  while (true) {
-    process.send(msg);
-  }
+  const msg = '.'.repeat(len);
+  const send = () => {
+    while (process.send(msg));
+    // Wait: backlog of unsent messages exceeds threshold
+    setImmediate(send);
+  };
+  send();
 } else {
   const common = require('../common.js');
   const bench = common.createBenchmark(main, {
-    len: [64, 256, 1024, 4096, 32768],
+    len: [
+      64, 256, 1024, 4096, 16384, 65536,
+      65536 << 4, 65536 << 8
+    ],
     dur: [5]
   });
   const spawn = require('child_process').spawn;
@@ -18,7 +24,7 @@ if (process.argv[2] === 'child') {
     const dur = +conf.dur;
     const len = +conf.len;
 
-    const options = { 'stdio': ['ignore', 'ignore', 'ignore', 'ipc'] };
+    const options = { 'stdio': ['ignore', 1, 2, 'ipc'] };
     const child = spawn(process.argv[0],
       [process.argv[1], 'child', len], options);
 
@@ -29,8 +35,7 @@ if (process.argv[2] === 'child') {
 
     setTimeout(function() {
       child.kill();
-      const gbits = (bytes * 8) / (1024 * 1024 * 1024);
-      bench.end(gbits);
+      bench.end(bytes);
     }, dur * 1000);
   }
 }

lib/internal/child_process.js

@@ -446,14 +446,20 @@ function setupChannel(target, channel) {
   channel.onread = function(nread, pool, recvHandle) {
     // TODO(bnoordhuis) Check that nread > 0.
     if (pool) {
-      jsonBuffer += decoder.write(pool);
-
-      var i, start = 0;
+      // Linebreak is used as a message end sign
+      var chunks = decoder.write(pool).split('\n');
+      var numCompleteChunks = chunks.length - 1;
+      // Last line does not have trailing linebreak
+      var incompleteChunk = chunks[numCompleteChunks];
+      if (numCompleteChunks === 0) {
+        jsonBuffer += incompleteChunk;
+        this.buffering = jsonBuffer.length !== 0;
+        return;
+      }
+      chunks[0] = jsonBuffer + chunks[0];
 
-      //Linebreak is used as a message end sign
-      while ((i = jsonBuffer.indexOf('\n', start)) >= 0) {
-        var json = jsonBuffer.slice(start, i);
-        var message = JSON.parse(json);
+      for (var i = 0; i < numCompleteChunks; i++) {
+        var message = JSON.parse(chunks[i]);
 
         // There will be at most one NODE_HANDLE message in every chunk we
         // read because SCM_RIGHTS messages don't get coalesced. Make sure
@@ -462,10 +468,8 @@ function setupChannel(target, channel) {
           handleMessage(target, message, recvHandle);
         else
           handleMessage(target, message, undefined);
-
-        start = i + 1;
       }
-      jsonBuffer = jsonBuffer.slice(start);
+      jsonBuffer = incompleteChunk;
       this.buffering = jsonBuffer.length !== 0;
 
     } else {
@@ -494,9 +498,10 @@ function setupChannel(target, channel) {
       var queue = target._handleQueue;
       target._handleQueue = null;
 
-      queue.forEach(function(args) {
+      for (var i = 0; i < queue.length; i++) {
+        var args = queue[i];
         target._send(args.message, args.handle, args.options, args.callback);
-      });
+      }
 
       // Process a pending disconnect (if any).
       if (!target.connected && target.channel && !target._handleQueue)