child_process: emit IPC messages on next tick

This commit fixes a regression related to IPC 'message' events. When messages are not emitted in the next tick, a 'message' handler that throws can break the IPC read loop. Refs: #6909 Refs: #13459 Refs: #13648 PR-URL: #13856 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Santiago Gimeno <[email protected]>

Author: cjihrig

lib/internal/child_process.js

@@ -456,7 +456,6 @@ function setupChannel(target, channel){
 }
 chunks[0]=jsonBuffer+chunks[0];
 
-varnextTick=false;
 for(vari=0;i<numCompleteChunks;i++){
 varmessage=JSON.parse(chunks[i]);
 
@@ -465,12 +464,11 @@ function setupChannel(target, channel){
 // that we deliver the handle with the right message however.
 if(isInternal(message)){
 if(message.cmd==='NODE_HANDLE')
-handleMessage(message,recvHandle,true,false);
+handleMessage(message,recvHandle,true);
 else
-handleMessage(message,undefined,true,false);
+handleMessage(message,undefined,true);
 }else{
-handleMessage(message,undefined,false,nextTick);
-nextTick=true;
+handleMessage(message,undefined,false);
 }
 }
 jsonBuffer=incompleteChunk;
@@ -532,7 +530,7 @@ function setupChannel(target, channel){
 
 // Convert handle object
 obj.got.call(this,message,handle,function(handle){
-handleMessage(message.msg,handle,isInternal(message.msg),false);
+handleMessage(message.msg,handle,isInternal(message.msg));
 });
 });
 
@@ -742,15 +740,13 @@ function setupChannel(target, channel){
 target.emit(event,message,handle);
 }
 
-functionhandleMessage(message,handle,internal,nextTick){
+functionhandleMessage(message,handle,internal){
 if(!target.channel)
 return;
 
 vareventName=(internal ? 'internalMessage' : 'message');
-if(nextTick)
-process.nextTick(emit,eventName,message,handle);
-else
-target.emit(eventName,message,handle);
+
+process.nextTick(emit,eventName,message,handle);
 }
 
 channel.readStart();

test/parallel/test-child-process-ipc-next-tick.js

@@ -0,0 +1,39 @@
+'use strict';
+constcommon=require('../common');
+constassert=require('assert');
+constcp=require('child_process');
+constNUM_MESSAGES=10;
+constvalues=[];
+
+for(leti=0;i<NUM_MESSAGES;++i){
+values[i]=i;
+}
+
+if(process.argv[2]==='child'){
+constreceived=values.map(()=>{returnfalse;});
+
+process.on('uncaughtException',common.mustCall((err)=>{
+received[err]=true;
+constdone=received.every((element)=>{returnelement===true;});
+
+if(done)
+process.disconnect();
+},NUM_MESSAGES));
+
+process.on('message',(msg)=>{
+// If messages are handled synchronously, throwing should break the IPC
+// message processing.
+throwmsg;
+});
+
+process.send('ready');
+}else{
+constchild=cp.fork(__filename,['child']);
+
+child.on('message',common.mustCall((msg)=>{
+assert.strictEqual(msg,'ready');
+values.forEach((value)=>{
+child.send(value);
+});
+}));
+}