child_process,cluster: allow using V8 serialization API

Add an `serialization` option that allows child process IPC to
use the (typically more powerful) V8 serialization API.

Fixes: https://github.com/nodejs/node/issues/10965

PR-URL: https://github.com/nodejs/node/pull/30162
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Gus Caplan <me@gus.host>
Reviewed-By: David Carlier <devnexen@gmail.com>
Reviewed-By: Michaël Zasso <targos@protonmail.com>

1 files changed, 27 insertions, 32 deletions

diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js
index 22f7da92ce8..9e13650fa3d 100644
--- a/lib/internal/child_process.js
+++ b/lib/internal/child_process.js
@@ -1,6 +1,6 @@
 'use strict';
 
-const { JSON, Object } = primordials;
+const { Object } = primordials;
 
 const {
   errnoException,
@@ -55,8 +55,6 @@ const {
 
 const { SocketListSend, SocketListReceive } = SocketList;
 
-// Lazy loaded for startup performance.
-let StringDecoder;
 // Lazy loaded for startup performance and to allow monkey patching of
 // internalBinding('http_parser').HTTPParser.
 let freeParser;
@@ -343,6 +341,15 @@ ChildProcess.prototype.spawn = function(options) {
   const ipcFd = stdio.ipcFd;
   stdio = options.stdio = stdio.stdio;
 
+  if (options.serialization !== undefined &&
+      options.serialization !== 'json' &&
+      options.serialization !== 'advanced') {
+    throw new ERR_INVALID_OPT_VALUE('options.serialization',
+                                    options.serialization);
+  }
+
+  const serialization = options.serialization || 'json';
+
   if (ipc !== undefined) {
     // Let child process know about opened IPC channel
     if (options.envPairs === undefined)
@@ -353,7 +360,8 @@ ChildProcess.prototype.spawn = function(options) {
                                      options.envPairs);
     }
 
-    options.envPairs.push('NODE_CHANNEL_FD=' + ipcFd);
+    options.envPairs.push(`NODE_CHANNEL_FD=${ipcFd}`);
+    options.envPairs.push(`NODE_CHANNEL_SERIALIZATION_MODE=${serialization}`);
   }
 
   validateString(options.file, 'options.file');
@@ -446,7 +454,7 @@ ChildProcess.prototype.spawn = function(options) {
     this.stdio.push(stdio[i].socket === undefined ? null : stdio[i].socket);
 
   // Add .send() method and start listening for IPC data
-  if (ipc !== undefined) setupChannel(this, ipc);
+  if (ipc !== undefined) setupChannel(this, ipc, serialization);
 
   return err;
 };
@@ -516,7 +524,8 @@ class Control extends EventEmitter {
 const channelDeprecationMsg = '_channel is deprecated. ' +
                               'Use ChildProcess.channel instead.';
 
-function setupChannel(target, channel) {
+let serialization;
+function setupChannel(target, channel, serializationMode) {
   target.channel = channel;
 
   Object.defineProperty(target, '_channel', {
@@ -535,12 +544,16 @@ function setupChannel(target, channel) {
 
   const control = new Control(channel);
 
-  if (StringDecoder === undefined)
-    StringDecoder = require('string_decoder').StringDecoder;
-  const decoder = new StringDecoder('utf8');
-  var jsonBuffer = '';
-  var pendingHandle = null;
-  channel.buffering = false;
+  if (serialization === undefined)
+    serialization = require('internal/child_process/serialization');
+  const {
+    initMessageChannel,
+    parseChannelMessages,
+    writeChannelMessage
+  } = serialization[serializationMode];
+
+  let pendingHandle = null;
+  initMessageChannel(channel);
   channel.pendingHandle = null;
   channel.onread = function(arrayBuffer) {
     const recvHandle = channel.pendingHandle;
@@ -552,21 +565,7 @@ function setupChannel(target, channel) {
       if (recvHandle)
         pendingHandle = recvHandle;
 
-      // 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];
-
-      for (var i = 0; i < numCompleteChunks; i++) {
-        var message = JSON.parse(chunks[i]);
-
+      for (const message of parseChannelMessages(channel, pool)) {
         // There will be at most one NODE_HANDLE message in every chunk we
         // read because SCM_RIGHTS messages don't get coalesced. Make sure
         // that we deliver the handle with the right message however.
@@ -581,9 +580,6 @@ function setupChannel(target, channel) {
           handleMessage(message, undefined, false);
         }
       }
-      jsonBuffer = incompleteChunk;
-      this.buffering = jsonBuffer.length !== 0;
-
     } else {
       this.buffering = false;
       target.disconnect();
@@ -782,8 +778,7 @@ function setupChannel(target, channel) {
 
     const req = new WriteWrap();
 
-    const string = JSON.stringify(message) + '\n';
-    const err = channel.writeUtf8String(req, string, handle);
+    const err = writeChannelMessage(channel, req, message, handle);
     const wasAsyncWrite = streamBaseState[kLastWriteWasAsync];
 
     if (err === 0) {