src,lib: the handle keeps loop alive in cluster rr mode

Author: theanarkh

lib/internal/cluster/child.js

@@ -16,6 +16,8 @@ const { owner_symbol } = require('internal/async_hooks').symbols;
 const Worker = require('internal/cluster/worker');
 const { internal, sendHelper } = require('internal/cluster/utils');
 const { exitCodes: { kNoFailure } } = internalBinding('errors');
+const { TIMEOUT_MAX } = require('internal/timers');
+const { setInterval, clearInterval } = require('timers');
 
 const cluster = new EventEmitter();
 const handles = new SafeMap();
@@ -162,6 +164,21 @@ function rr(message, { indexesKey, index }, cb) {
 
   let key = message.key;
 
+  let fakeHandle = null;
+
+  function ref() {
+    if (!fakeHandle) {
+      fakeHandle = setInterval(noop, TIMEOUT_MAX);
+    }
+  }
+
+  function unref() {
+    if (fakeHandle) {
+      clearInterval(fakeHandle);
+      fakeHandle = null;
+    }
+  }
+
   function listen(backlog) {
     // TODO(bnoordhuis) Send a message to the primary that tells it to
     // update the backlog size. The actual backlog should probably be
@@ -177,8 +194,7 @@ function rr(message, { indexesKey, index }, cb) {
     // the primary.
     if (key === undefined)
       return;
-
-    send({ act: 'close', key });
+    send({ act: 'close', key, needAck: true }, () => unref());
     handles.delete(key);
     removeIndexesKey(indexesKey, index);
     key = undefined;
@@ -191,12 +207,10 @@ function rr(message, { indexesKey, index }, cb) {
     return 0;
   }
 
-  // Faux handle. Mimics a TCPWrap with just enough fidelity to get away
-  // with it. Fools net.Server into thinking that it's backed by a real
-  // handle. Use a noop function for ref() and unref() because the control
-  // channel is going to keep the worker alive anyway.
-  const handle = { close, listen, ref: noop, unref: noop };
-
+  // Faux handle. net.Server is not associated with handle,
+  // so we control its state(ref or unref) by setInterval.
+  const handle = { close, listen, ref, unref };
+  handle.ref();
   if (message.sockname) {
     handle.getsockname = getsockname;  // TCP handles only.
   }

lib/internal/cluster/primary.js

@@ -342,6 +342,9 @@ function close(worker, message) {
 
   if (handle && handle.remove(worker))
     handles.delete(key);
+  if (message.needAck) {
+    send(worker, { ack: message.seq });
+  }
 }
 
 function send(worker, message, handle, cb) {

test/parallel/test-cluster-rr-handle-close.js

@@ -0,0 +1,18 @@
+'use strict';
+
+const common = require('../common');
+const cluster = require('cluster');
+const net = require('net');
+
+cluster.schedulingPolicy = cluster.SCHED_RR;
+
+if (cluster.isPrimary) {
+  const worker = cluster.fork();
+  worker.on('exit', common.mustCall());
+} else {
+  const server = net.createServer(common.mustNotCall());
+  server.listen(0, common.mustCall(() => {
+    process.channel.unref();
+    server.close();
+  }));
+}

test/parallel/test-cluster-rr-handle-keep-loop-alive.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const common = require('../common');
+const cluster = require('cluster');
+const net = require('net');
+const assert = require('assert');
+
+cluster.schedulingPolicy = cluster.SCHED_RR;
+
+if (cluster.isPrimary) {
+  let exited = false;
+  const worker = cluster.fork();
+  worker.on('exit', () => {
+    exited = true;
+  });
+  setTimeout(() => {
+    assert.ok(!exited);
+    worker.kill();
+  }, 3000);
+} else {
+  const server = net.createServer(common.mustNotCall());
+  server.listen(0, common.mustCall(() => process.channel.unref()));
+}

test/parallel/test-cluster-rr-handle-ref-unref.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const common = require('../common');
+const cluster = require('cluster');
+const net = require('net');
+
+cluster.schedulingPolicy = cluster.SCHED_RR;
+
+if (cluster.isPrimary) {
+  const worker = cluster.fork();
+  worker.on('exit', common.mustCall());
+} else {
+  const server = net.createServer(common.mustNotCall());
+  server.listen(0, common.mustCall(() => {
+    server.ref();
+    server.unref();
+    process.channel.unref();
+  }));
+  server.unref();
+}