s, }; message.address = address; // Set custom data on handle (i.e. tls tickets key) if (obj._getServerData) message.data = obj._getServerData(); send(message, (reply, handle) => { if (typeof obj._setServerData === 'function') obj._setServerData(reply.data); if (handle) { // Shared listen socket shared(reply, { handle, indexesKey, index }, cb); } else { // Round-robin. rr(reply, { indexesKey, index }, cb); } }); obj.once('listening', () => { // short-lived sockets might have been closed if (!indexes.has(indexesKey)) { return; } cluster.worker.state = 'listening'; const address = obj.address(); message.act = 'listening'; message.port = (address && address.port) || options.port; send(message); }); }; function removeIndexesKey(indexesKey, index) { const indexSet = indexes.get(indexesKey); if (!indexSet) { return; } indexSet.set.delete(index); if (indexSet.set.size === 0) { indexes.delete(indexesKey); } } // Shared listen socket. function shared(message, { handle, indexesKey, index }, cb) { const key = message.key; // Monkey-patch the close() method so we can keep track of when it's // closed. Avoids resource leaks when the handle is short-lived. const close = handle.close; handle.close = function() { send({ act: 'close', key }); handles.delete(key); removeIndexesKey(indexesKey, index); return ReflectApply(close, handle, arguments); }; assert(handles.has(key) === false); handles.set(key, handle); cb(message.errno, handle); } // Round-robin. Master distributes handles across workers. function rr(message, { indexesKey, index }, cb) { if (message.errno) return cb(message.errno, null); 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 // the largest requested size by any worker. return 0; } function close() { // lib/net.js treats server._handle.close() as effectively synchronous. // That means there is a time window between the call to close() and // the ack by the primary process in which we can still receive handles. // onconnection() below handles that by sending those handles back to // the primary. if (key === undefined) return; unref(); // If the handle is the last handle in process, // the parent process will delete the handle when worker process exits. // So it is ok if the close message get lost. // See the comments of https://github.com/nodejs/node/pull/46161 send({ act: 'close', key }); handles.delete(key); removeIndexesKey(indexesKey, index); key = undefined; } function getsockname(out) { if (key) ObjectAssign(out, message.sockname); return 0; } // 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. } assert(handles.has(key) === false); handles.set(key, handle); cb(0, handle); } // Round-robin connection. function onconnection(message, handle) { const key = message.key; const server = handles.get(key); let accepted = server !== undefined; if (accepted && server[owner_symbol]) { const self = server[owner_symbol]; if (self.maxConnections && self._connections >= self.maxConnections) { accepted = false; } } send({ ack: message.seq, accepted }); if (accepted) server.onconnection(0, handle); else handle.close(); } function send(message, cb) { return sendHelper(process, message, null, cb); } function _disconnect(primaryInitiated) { this.exitedAfterDisconnect = true; let waitingCount = 1; function checkWaitingCount() { waitingCount--; if (waitingCount === 0) { // If disconnect is worker initiated, wait for ack to be sure // exitedAfterDisconnect is properly set in the primary, otherwise, if // it's primary initiated there's no need to send the // exitedAfterDisconnect message if (primaryInitiated) { process.disconnect(); } else { send({ act: 'exitedAfterDisconnect' }, () => process.disconnect()); } } } handles.forEach((handle) => { waitingCount++; if (handle[owner_symbol]) handle[owner_symbol].close(checkWaitingCount); else handle.close(checkWaitingCount); }); handles.clear(); checkWaitingCount(); } // Extend generic Worker with methods specific to worker processes. Worker.prototype.disconnect = function() { if (this.state !== 'disconnecting' && this.state !== 'destroying') { this.state = 'disconnecting'; ReflectApply(_disconnect, this, []); } return this; }; Worker.prototype.destroy = function() { if (this.state === 'destroying') return; this.exitedAfterDisconnect = true; if (!this.isConnected()) { process.exit(0); } else { this.state = 'destroying'; send({ act: 'exitedAfterDisconnect' }, () => process.disconnect()); process.once('disconnect', () => process.exit(0)); } };