3 const EventEmitter = require('events');
4 const assert = require('assert');
5 const dgram = require('dgram');
6 const fork = require('child_process').fork;
7 const net = require('net');
8 const util = require('util');
12 const uv = process.binding('uv');
14 const cluster = new EventEmitter();
15 module.exports = cluster;
16 cluster.Worker = Worker;
17 cluster.isWorker = ('NODE_UNIQUE_ID' in process.env);
18 cluster.isMaster = (cluster.isWorker === false);
21 function Worker(options) {
22 if (!(this instanceof Worker))
23 return new Worker(options);
25 EventEmitter.call(this);
27 if (options === null || typeof options !== 'object')
30 this.suicide = undefined;
31 this.state = options.state || 'none';
32 this.id = options.id | 0;
34 if (options.process) {
35 this.process = options.process;
36 this.process.on('error', this.emit.bind(this, 'error'));
37 this.process.on('message', this.emit.bind(this, 'message'));
40 util.inherits(Worker, EventEmitter);
42 Worker.prototype.kill = function() {
43 this.destroy.apply(this, arguments);
46 Worker.prototype.send = function() {
47 this.process.send.apply(this.process, arguments);
50 Worker.prototype.isDead = function isDead() {
51 return this.process.exitCode != null || this.process.signalCode != null;
54 Worker.prototype.isConnected = function isConnected() {
55 return this.process.connected;
58 // Master/worker specific methods are defined in the *Init() functions.
60 function SharedHandle(key, address, port, addressType, backlog, fd, flags) {
66 // FIXME(bnoordhuis) Polymorphic return type for lack of a better solution.
68 if (addressType === 'udp4' || addressType === 'udp6')
69 rval = dgram._createSocketHandle(address, port, addressType, fd, flags);
71 rval = net._createServerHandle(address, port, addressType, fd);
73 if (typeof rval === 'number')
79 SharedHandle.prototype.add = function(worker, send) {
80 assert(this.workers.indexOf(worker) === -1);
81 this.workers.push(worker);
82 send(this.errno, null, this.handle);
85 SharedHandle.prototype.remove = function(worker) {
86 var index = this.workers.indexOf(worker);
87 if (index === -1) return false; // The worker wasn't sharing this handle.
88 this.workers.splice(index, 1);
89 if (this.workers.length !== 0) return false;
96 // Start a round-robin server. Master accepts connections and distributes
97 // them over the workers.
98 function RoundRobinHandle(key, address, port, addressType, backlog, fd) {
104 this.server = net.createServer(assert.fail);
107 this.server.listen({ fd: fd });
109 this.server.listen(port, address);
111 this.server.listen(address); // UNIX socket path.
114 this.server.once('listening', function() {
115 self.handle = self.server._handle;
116 self.handle.onconnection = self.distribute.bind(self);
117 self.server._handle = null;
122 RoundRobinHandle.prototype.add = function(worker, send) {
123 assert(worker.id in this.all === false);
124 this.all[worker.id] = worker;
128 if (self.handle.getsockname) {
130 self.handle.getsockname(out);
131 // TODO(bnoordhuis) Check err.
132 send(null, { sockname: out }, null);
134 send(null, null, null); // UNIX socket.
136 self.handoff(worker); // In case there are connections pending.
139 if (this.server === null) return done();
140 // Still busy binding.
141 this.server.once('listening', done);
142 this.server.once('error', function(err) {
143 // Hack: translate 'EADDRINUSE' error string back to numeric error code.
144 // It works but ideally we'd have some backchannel between the net and
145 // cluster modules for stuff like this.
146 var errno = uv['UV_' + err.errno];
151 RoundRobinHandle.prototype.remove = function(worker) {
152 if (worker.id in this.all === false) return false;
153 delete this.all[worker.id];
154 var index = this.free.indexOf(worker);
155 if (index !== -1) this.free.splice(index, 1);
156 if (Object.getOwnPropertyNames(this.all).length !== 0) return false;
157 for (var handle; handle = this.handles.shift(); handle.close());
163 RoundRobinHandle.prototype.distribute = function(err, handle) {
164 this.handles.push(handle);
165 var worker = this.free.shift();
166 if (worker) this.handoff(worker);
169 RoundRobinHandle.prototype.handoff = function(worker) {
170 if (worker.id in this.all === false) {
171 return; // Worker is closing (or has closed) the server.
173 var handle = this.handles.shift();
174 if (handle === undefined) {
175 this.free.push(worker); // Add to ready queue again.
178 var message = { act: 'newconn', key: this.key };
180 sendHelper(worker.process, message, handle, function(reply) {
184 self.distribute(0, handle); // Worker is shutting down. Send to another.
185 self.handoff(worker);
190 if (cluster.isMaster)
195 function masterInit() {
196 cluster.workers = {};
198 var intercom = new EventEmitter();
199 cluster.settings = {};
201 // XXX(bnoordhuis) Fold cluster.schedulingPolicy into cluster.settings?
202 var schedulingPolicy = {
205 }[process.env.NODE_CLUSTER_SCHED_POLICY];
207 if (schedulingPolicy === undefined) {
208 // FIXME Round-robin doesn't perform well on Windows right now due to the
209 // way IOCP is wired up. Bert is going to fix that, eventually.
210 schedulingPolicy = (process.platform === 'win32') ? SCHED_NONE : SCHED_RR;
213 cluster.schedulingPolicy = schedulingPolicy;
214 cluster.SCHED_NONE = SCHED_NONE; // Leave it to the operating system.
215 cluster.SCHED_RR = SCHED_RR; // Master distributes connections.
217 // Keyed on address:port:etc. When a worker dies, we walk over the handles
218 // and remove() the worker from each one. remove() may do a linear scan
219 // itself so we might end up with an O(n*m) operation. Ergo, FIXME.
220 const handles = require('internal/cluster').handles;
222 var initialized = false;
223 cluster.setupMaster = function(options) {
225 args: process.argv.slice(2),
226 exec: process.argv[1],
227 execArgv: process.execArgv,
230 settings = util._extend(settings, cluster.settings);
231 settings = util._extend(settings, options || {});
232 // Tell V8 to write profile data for each process to a separate file.
233 // Without --logfile=v8-%p.log, everything ends up in a single, unusable
234 // file. (Unusable because what V8 logs are memory addresses and each
235 // process has its own memory mappings.)
236 if (settings.execArgv.some(function(s) { return /^--prof/.test(s); }) &&
237 !settings.execArgv.some(function(s) { return /^--logfile=/.test(s); }))
239 settings.execArgv = settings.execArgv.concat(['--logfile=v8-%p.log']);
241 cluster.settings = settings;
242 if (initialized === true)
243 return process.nextTick(setupSettingsNT, settings);
245 schedulingPolicy = cluster.schedulingPolicy; // Freeze policy.
246 assert(schedulingPolicy === SCHED_NONE || schedulingPolicy === SCHED_RR,
247 'Bad cluster.schedulingPolicy: ' + schedulingPolicy);
249 var hasDebugArg = process.execArgv.some(function(argv) {
250 return /^(--debug|--debug-brk)(=\d+)?$/.test(argv);
253 process.nextTick(setupSettingsNT, settings);
255 // Send debug signal only if not started in debug mode, this helps a lot
256 // on windows, because RegisterDebugHandler is not called when node starts
257 // with --debug.* arg.
261 process.on('internalMessage', function(message) {
262 if (message.cmd !== 'NODE_DEBUG_ENABLED') return;
264 for (key in cluster.workers) {
265 var worker = cluster.workers[key];
266 if (worker.state === 'online' || worker.state === 'listening') {
267 process._debugProcess(worker.process.pid);
269 worker.once('online', function() {
270 process._debugProcess(this.process.pid);
277 function setupSettingsNT(settings) {
278 cluster.emit('setup', settings);
281 var debugPortOffset = 1;
283 function createWorkerProcess(id, env) {
284 var workerEnv = util._extend({}, process.env);
285 var execArgv = cluster.settings.execArgv.slice();
287 workerEnv = util._extend(workerEnv, env);
288 workerEnv.NODE_UNIQUE_ID = '' + id;
290 for (var i = 0; i < execArgv.length; i++) {
291 var match = execArgv[i].match(/^(--debug|--debug-(brk|port))(=\d+)?$/);
294 const debugPort = process.debugPort + debugPortOffset;
296 execArgv[i] = match[1] + '=' + debugPort;
300 return fork(cluster.settings.exec, cluster.settings.args, {
302 silent: cluster.settings.silent,
304 gid: cluster.settings.gid,
305 uid: cluster.settings.uid
311 function removeWorker(worker) {
314 delete cluster.workers[worker.id];
316 if (Object.keys(cluster.workers).length === 0) {
317 assert(Object.keys(handles).length === 0, 'Resource leak detected.');
318 intercom.emit('disconnect');
322 function removeHandlesForWorker(worker) {
325 for (var key in handles) {
326 var handle = handles[key];
327 if (handle.remove(worker)) delete handles[key];
331 cluster.fork = function(env) {
332 cluster.setupMaster();
334 const workerProcess = createWorkerProcess(id, env);
335 const worker = new Worker({
337 process: workerProcess
340 worker.on('message', this.emit.bind(this, 'message'));
342 worker.process.once('exit', function(exitCode, signalCode) {
344 * Remove the worker from the workers list only
345 * if it has disconnected, otherwise we might
346 * still want to access it.
348 if (!worker.isConnected()) {
349 removeHandlesForWorker(worker);
350 removeWorker(worker);
353 worker.suicide = !!worker.suicide;
354 worker.state = 'dead';
355 worker.emit('exit', exitCode, signalCode);
356 cluster.emit('exit', worker, exitCode, signalCode);
359 worker.process.once('disconnect', function() {
361 * Now is a good time to remove the handles
362 * associated with this worker because it is
363 * not connected to the master anymore.
365 removeHandlesForWorker(worker);
368 * Remove the worker from the workers list only
369 * if its process has exited. Otherwise, we might
370 * still want to access it.
372 if (worker.isDead()) removeWorker(worker);
374 worker.suicide = !!worker.suicide;
375 worker.state = 'disconnected';
376 worker.emit('disconnect');
377 cluster.emit('disconnect', worker);
380 worker.process.on('internalMessage', internal(worker, onmessage));
381 process.nextTick(emitForkNT, worker);
382 cluster.workers[worker.id] = worker;
386 function emitForkNT(worker) {
387 cluster.emit('fork', worker);
390 cluster.disconnect = function(cb) {
391 var workers = Object.keys(cluster.workers);
392 if (workers.length === 0) {
393 process.nextTick(intercom.emit.bind(intercom, 'disconnect'));
395 for (var key in workers) {
397 if (cluster.workers[key].isConnected())
398 cluster.workers[key].disconnect();
401 if (cb) intercom.once('disconnect', cb);
404 Worker.prototype.disconnect = function() {
406 send(this, { act: 'disconnect' });
407 removeHandlesForWorker(this);
411 Worker.prototype.destroy = function(signo) {
412 signo = signo || 'SIGTERM';
413 var proc = this.process;
414 if (this.isConnected()) {
415 this.once('disconnect', proc.kill.bind(proc, signo));
422 function onmessage(message, handle) {
424 if (message.act === 'online')
426 else if (message.act === 'queryServer')
427 queryServer(worker, message);
428 else if (message.act === 'listening')
429 listening(worker, message);
430 else if (message.act === 'suicide')
431 worker.suicide = true;
432 else if (message.act === 'close')
433 close(worker, message);
436 function online(worker) {
437 worker.state = 'online';
438 worker.emit('online');
439 cluster.emit('online', worker);
442 function queryServer(worker, message) {
443 var args = [message.address,
448 var key = args.join(':');
449 var handle = handles[key];
450 if (handle === undefined) {
451 var constructor = RoundRobinHandle;
452 // UDP is exempt from round-robin connection balancing for what should
453 // be obvious reasons: it's connectionless. There is nothing to send to
454 // the workers except raw datagrams and that's pointless.
455 if (schedulingPolicy !== SCHED_RR ||
456 message.addressType === 'udp4' ||
457 message.addressType === 'udp6') {
458 constructor = SharedHandle;
460 handles[key] = handle = new constructor(key,
468 if (!handle.data) handle.data = message.data;
470 // Set custom server data
471 handle.add(worker, function(errno, reply, handle) {
472 reply = util._extend({
476 data: handles[key].data
478 if (errno) delete handles[key]; // Gives other workers a chance to retry.
479 send(worker, reply, handle);
483 function listening(worker, message) {
485 addressType: message.addressType,
486 address: message.address,
490 worker.state = 'listening';
491 worker.emit('listening', info);
492 cluster.emit('listening', worker, info);
495 // Server in worker is closing, remove from list. The handle may have been
496 // removed by a prior call to removeHandlesForWorker() so guard against that.
497 function close(worker, message) {
498 var key = message.key;
499 var handle = handles[key];
500 if (handle && handle.remove(worker)) delete handles[key];
503 function send(worker, message, handle, cb) {
504 sendHelper(worker.process, message, handle, cb);
509 function workerInit() {
513 // Called from src/node.js
514 cluster._setupWorker = function() {
515 var worker = new Worker({
516 id: +process.env.NODE_UNIQUE_ID | 0,
520 cluster.worker = worker;
521 process.once('disconnect', function() {
522 worker.emit('disconnect');
523 if (!worker.suicide) {
524 // Unexpected disconnect, master exited, or some such nastiness, so
525 // worker exits immediately.
529 process.on('internalMessage', internal(worker, onmessage));
530 send({ act: 'online' });
531 function onmessage(message, handle) {
532 if (message.act === 'newconn')
533 onconnection(message, handle);
534 else if (message.act === 'disconnect')
539 // obj is a net#Server or a dgram#Socket object.
540 cluster._getServer = function(obj, options, cb) {
541 const key = [ options.address,
544 options.fd ].join(':');
545 if (indexes[key] === undefined)
550 const message = util._extend({
556 // Set custom data on handle (i.e. tls tickets key)
557 if (obj._getServerData) message.data = obj._getServerData();
558 send(message, function(reply, handle) {
559 if (obj._setServerData) obj._setServerData(reply.data);
562 shared(reply, handle, cb); // Shared listen socket.
564 rr(reply, cb); // Round-robin.
566 obj.once('listening', function() {
567 cluster.worker.state = 'listening';
568 const address = obj.address();
569 message.act = 'listening';
570 message.port = address && address.port || options.port;
575 // Shared listen socket.
576 function shared(message, handle, cb) {
577 var key = message.key;
578 // Monkey-patch the close() method so we can keep track of when it's
579 // closed. Avoids resource leaks when the handle is short-lived.
580 var close = handle.close;
581 handle.close = function() {
582 send({ act: 'close', key: key });
584 return close.apply(this, arguments);
586 assert(handles[key] === undefined);
587 handles[key] = handle;
588 cb(message.errno, handle);
591 // Round-robin. Master distributes handles across workers.
592 function rr(message, cb) {
594 return cb(message.errno, null);
596 var key = message.key;
597 function listen(backlog) {
598 // TODO(bnoordhuis) Send a message to the master that tells it to
599 // update the backlog size. The actual backlog should probably be
600 // the largest requested size by any worker.
605 // lib/net.js treats server._handle.close() as effectively synchronous.
606 // That means there is a time window between the call to close() and
607 // the ack by the master process in which we can still receive handles.
608 // onconnection() below handles that by sending those handles back to
610 if (key === undefined) return;
611 send({ act: 'close', key: key });
616 function getsockname(out) {
617 if (key) util._extend(out, message.sockname);
621 // XXX(bnoordhuis) Probably no point in implementing ref() and unref()
622 // because the control channel is going to keep the worker alive anyway.
629 // Faux handle. Mimics a TCPWrap with just enough fidelity to get away
630 // with it. Fools net.Server into thinking that it's backed by a real
638 if (message.sockname) {
639 handle.getsockname = getsockname; // TCP handles only.
641 assert(handles[key] === undefined);
642 handles[key] = handle;
646 // Round-robin connection.
647 function onconnection(message, handle) {
648 var key = message.key;
649 var server = handles[key];
650 var accepted = server !== undefined;
651 send({ ack: message.seq, accepted: accepted });
652 if (accepted) server.onconnection(0, handle);
655 Worker.prototype.disconnect = function() {
657 let waitingCount = 1;
659 function checkWaitingCount() {
661 if (waitingCount === 0) {
662 send({ act: 'suicide' });
663 process.disconnect();
667 for (const key in handles) {
668 const handle = handles[key];
671 handle.owner.close(checkWaitingCount);
677 Worker.prototype.destroy = function() {
679 if (!this.isConnected()) process.exit(0);
680 var exit = process.exit.bind(null, 0);
681 send({ act: 'suicide' }, exit);
682 process.once('disconnect', exit);
683 process.disconnect();
686 function send(message, cb) {
687 sendHelper(process, message, null, cb);
694 function sendHelper(proc, message, handle, cb) {
695 // Mark message as internal. See INTERNAL_PREFIX in lib/child_process.js
696 message = util._extend({ cmd: 'NODE_CLUSTER' }, message);
697 if (cb) callbacks[seq] = cb;
700 proc.send(message, handle);
704 // Returns an internalMessage listener that hands off normal messages
705 // to the callback but intercepts and redirects ACK messages.
706 function internal(worker, cb) {
707 return function(message, handle) {
708 if (message.cmd !== 'NODE_CLUSTER') return;
710 if (message.ack !== undefined) {
711 fn = callbacks[message.ack];
712 delete callbacks[message.ack];
714 fn.apply(worker, arguments);