Skip to content

Commit d751afa

Browse files
cjihrigitaloacasas
cjihrig
authored and
italoacasas
committedJan 30, 2017
cluster: refactor module into multiple files
This commit splits the existing cluster module into several internal modules. More specifically, the cluster master and worker implementations are separated, and the various data structures are separated. PR-URL: #10746 Reviewed-By: James M Snell <[email protected]> Reviewed-By: Sam Roberts <[email protected]>
·
v7.10.1v7.5.0
1 parent 4a7bb5b commit d751afa

File tree

10 files changed

+871
-772
lines changed

10 files changed

+871
-772
lines changed
 

‎lib/cluster.js

+3-766
Large diffs are not rendered by default.

‎lib/internal/cluster.js

-4
This file was deleted.

‎lib/internal/cluster/child.js

+224
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,224 @@
1+
'use strict';
2+
const assert = require('assert');
3+
const util = require('util');
4+
const EventEmitter = require('events');
5+
const Worker = require('internal/cluster/worker');
6+
const { internal, sendHelper } = require('internal/cluster/utils');
7+
const cluster = new EventEmitter();
8+
const handles = {};
9+
const indexes = {};
10+
const noop = () => {};
11+
12+
module.exports = cluster;
13+
14+
cluster.isWorker = true;
15+
cluster.isMaster = false;
16+
cluster.worker = null;
17+
cluster.Worker = Worker;
18+
19+
cluster._setupWorker = function() {
20+
const worker = new Worker({
21+
id: +process.env.NODE_UNIQUE_ID | 0,
22+
process: process,
23+
state: 'online'
24+
});
25+
26+
cluster.worker = worker;
27+
28+
process.once('disconnect', () => {
29+
worker.emit('disconnect');
30+
31+
if (!worker.exitedAfterDisconnect) {
32+
// Unexpected disconnect, master exited, or some such nastiness, so
33+
// worker exits immediately.
34+
process.exit(0);
35+
}
36+
});
37+
38+
process.on('internalMessage', internal(worker, onmessage));
39+
send({ act: 'online' });
40+
41+
function onmessage(message, handle) {
42+
if (message.act === 'newconn')
43+
onconnection(message, handle);
44+
else if (message.act === 'disconnect')
45+
_disconnect.call(worker, true);
46+
}
47+
};
48+
49+
// obj is a net#Server or a dgram#Socket object.
50+
cluster._getServer = function(obj, options, cb) {
51+
const indexesKey = [options.address,
52+
options.port,
53+
options.addressType,
54+
options.fd ].join(':');
55+
56+
if (indexes[indexesKey] === undefined)
57+
indexes[indexesKey] = 0;
58+
else
59+
indexes[indexesKey]++;
60+
61+
const message = util._extend({
62+
act: 'queryServer',
63+
index: indexes[indexesKey],
64+
data: null
65+
}, options);
66+
67+
// Set custom data on handle (i.e. tls tickets key)
68+
if (obj._getServerData)
69+
message.data = obj._getServerData();
70+
71+
send(message, (reply, handle) => {
72+
if (typeof obj._setServerData === 'function')
73+
obj._setServerData(reply.data);
74+
75+
if (handle)
76+
shared(reply, handle, indexesKey, cb); // Shared listen socket.
77+
else
78+
rr(reply, indexesKey, cb); // Round-robin.
79+
});
80+
81+
obj.once('listening', () => {
82+
cluster.worker.state = 'listening';
83+
const address = obj.address();
84+
message.act = 'listening';
85+
message.port = address && address.port || options.port;
86+
send(message);
87+
});
88+
};
89+
90+
// Shared listen socket.
91+
function shared(message, handle, indexesKey, cb) {
92+
const key = message.key;
93+
// Monkey-patch the close() method so we can keep track of when it's
94+
// closed. Avoids resource leaks when the handle is short-lived.
95+
const close = handle.close;
96+
97+
handle.close = function() {
98+
send({ act: 'close', key });
99+
delete handles[key];
100+
delete indexes[indexesKey];
101+
return close.apply(this, arguments);
102+
};
103+
assert(handles[key] === undefined);
104+
handles[key] = handle;
105+
cb(message.errno, handle);
106+
}
107+
108+
// Round-robin. Master distributes handles across workers.
109+
function rr(message, indexesKey, cb) {
110+
if (message.errno)
111+
return cb(message.errno, null);
112+
113+
var key = message.key;
114+
115+
function listen(backlog) {
116+
// TODO(bnoordhuis) Send a message to the master that tells it to
117+
// update the backlog size. The actual backlog should probably be
118+
// the largest requested size by any worker.
119+
return 0;
120+
}
121+
122+
function close() {
123+
// lib/net.js treats server._handle.close() as effectively synchronous.
124+
// That means there is a time window between the call to close() and
125+
// the ack by the master process in which we can still receive handles.
126+
// onconnection() below handles that by sending those handles back to
127+
// the master.
128+
if (key === undefined)
129+
return;
130+
131+
send({ act: 'close', key });
132+
delete handles[key];
133+
delete indexes[indexesKey];
134+
key = undefined;
135+
}
136+
137+
function getsockname(out) {
138+
if (key)
139+
util._extend(out, message.sockname);
140+
141+
return 0;
142+
}
143+
144+
// Faux handle. Mimics a TCPWrap with just enough fidelity to get away
145+
// with it. Fools net.Server into thinking that it's backed by a real
146+
// handle. Use a noop function for ref() and unref() because the control
147+
// channel is going to keep the worker alive anyway.
148+
const handle = { close, listen, ref: noop, unref: noop };
149+
150+
if (message.sockname) {
151+
handle.getsockname = getsockname; // TCP handles only.
152+
}
153+
154+
assert(handles[key] === undefined);
155+
handles[key] = handle;
156+
cb(0, handle);
157+
}
158+
159+
// Round-robin connection.
160+
function onconnection(message, handle) {
161+
const key = message.key;
162+
const server = handles[key];
163+
const accepted = server !== undefined;
164+
165+
send({ ack: message.seq, accepted });
166+
167+
if (accepted)
168+
server.onconnection(0, handle);
169+
}
170+
171+
function send(message, cb) {
172+
return sendHelper(process, message, null, cb);
173+
}
174+
175+
function _disconnect(masterInitiated) {
176+
this.exitedAfterDisconnect = true;
177+
let waitingCount = 1;
178+
179+
function checkWaitingCount() {
180+
waitingCount--;
181+
182+
if (waitingCount === 0) {
183+
// If disconnect is worker initiated, wait for ack to be sure
184+
// exitedAfterDisconnect is properly set in the master, otherwise, if
185+
// it's master initiated there's no need to send the
186+
// exitedAfterDisconnect message
187+
if (masterInitiated) {
188+
process.disconnect();
189+
} else {
190+
send({ act: 'exitedAfterDisconnect' }, () => process.disconnect());
191+
}
192+
}
193+
}
194+
195+
for (const key in handles) {
196+
const handle = handles[key];
197+
delete handles[key];
198+
waitingCount++;
199+
200+
if (handle.owner)
201+
handle.owner.close(checkWaitingCount);
202+
else
203+
handle.close(checkWaitingCount);
204+
}
205+
206+
checkWaitingCount();
207+
}
208+
209+
// Extend generic Worker with methods specific to worker processes.
210+
Worker.prototype.disconnect = function() {
211+
_disconnect.call(this);
212+
return this;
213+
};
214+
215+
Worker.prototype.destroy = function() {
216+
this.exitedAfterDisconnect = true;
217+
218+
if (!this.isConnected()) {
219+
process.exit(0);
220+
} else {
221+
send({ act: 'exitedAfterDisconnect' }, () => process.disconnect());
222+
process.once('disconnect', () => process.exit(0));
223+
}
224+
};

‎lib/internal/cluster/master.js

+367
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,367 @@
1+
'use strict';
2+
const assert = require('assert');
3+
const fork = require('child_process').fork;
4+
const util = require('util');
5+
const EventEmitter = require('events');
6+
const RoundRobinHandle = require('internal/cluster/round_robin_handle');
7+
const SharedHandle = require('internal/cluster/shared_handle');
8+
const Worker = require('internal/cluster/worker');
9+
const { internal, sendHelper, handles } = require('internal/cluster/utils');
10+
const keys = Object.keys;
11+
const cluster = new EventEmitter();
12+
const intercom = new EventEmitter();
13+
const SCHED_NONE = 1;
14+
const SCHED_RR = 2;
15+
16+
module.exports = cluster;
17+
18+
cluster.isWorker = false;
19+
cluster.isMaster = true;
20+
cluster.Worker = Worker;
21+
cluster.workers = {};
22+
cluster.settings = {};
23+
cluster.SCHED_NONE = SCHED_NONE; // Leave it to the operating system.
24+
cluster.SCHED_RR = SCHED_RR; // Master distributes connections.
25+
26+
var ids = 0;
27+
var debugPortOffset = 1;
28+
var initialized = false;
29+
30+
// XXX(bnoordhuis) Fold cluster.schedulingPolicy into cluster.settings?
31+
var schedulingPolicy = {
32+
'none': SCHED_NONE,
33+
'rr': SCHED_RR
34+
}[process.env.NODE_CLUSTER_SCHED_POLICY];
35+
36+
if (schedulingPolicy === undefined) {
37+
// FIXME Round-robin doesn't perform well on Windows right now due to the
38+
// way IOCP is wired up.
39+
schedulingPolicy = (process.platform === 'win32') ? SCHED_NONE : SCHED_RR;
40+
}
41+
42+
cluster.schedulingPolicy = schedulingPolicy;
43+
44+
cluster.setupMaster = function(options) {
45+
var settings = {
46+
args: process.argv.slice(2),
47+
exec: process.argv[1],
48+
execArgv: process.execArgv,
49+
silent: false
50+
};
51+
settings = util._extend(settings, cluster.settings);
52+
settings = util._extend(settings, options || {});
53+
54+
// Tell V8 to write profile data for each process to a separate file.
55+
// Without --logfile=v8-%p.log, everything ends up in a single, unusable
56+
// file. (Unusable because what V8 logs are memory addresses and each
57+
// process has its own memory mappings.)
58+
if (settings.execArgv.some((s) => s.startsWith('--prof')) &&
59+
!settings.execArgv.some((s) => s.startsWith('--logfile='))) {
60+
settings.execArgv = settings.execArgv.concat(['--logfile=v8-%p.log']);
61+
}
62+
63+
cluster.settings = settings;
64+
65+
if (initialized === true)
66+
return process.nextTick(setupSettingsNT, settings);
67+
68+
initialized = true;
69+
schedulingPolicy = cluster.schedulingPolicy; // Freeze policy.
70+
assert(schedulingPolicy === SCHED_NONE || schedulingPolicy === SCHED_RR,
71+
`Bad cluster.schedulingPolicy: ${schedulingPolicy}`);
72+
73+
const hasDebugArg = process.execArgv.some((argv) => {
74+
return /^(--debug|--debug-brk)(=\d+)?$/.test(argv);
75+
});
76+
77+
process.nextTick(setupSettingsNT, settings);
78+
79+
// Send debug signal only if not started in debug mode, this helps a lot
80+
// on windows, because RegisterDebugHandler is not called when node starts
81+
// with --debug.* arg.
82+
if (hasDebugArg)
83+
return;
84+
85+
process.on('internalMessage', (message) => {
86+
if (message.cmd !== 'NODE_DEBUG_ENABLED')
87+
return;
88+
89+
var key;
90+
for (key in cluster.workers) {
91+
const worker = cluster.workers[key];
92+
93+
if (worker.state === 'online' || worker.state === 'listening') {
94+
process._debugProcess(worker.process.pid);
95+
} else {
96+
worker.once('online', function() {
97+
process._debugProcess(this.process.pid);
98+
});
99+
}
100+
}
101+
});
102+
};
103+
104+
function setupSettingsNT(settings) {
105+
cluster.emit('setup', settings);
106+
}
107+
108+
function createWorkerProcess(id, env) {
109+
var workerEnv = util._extend({}, process.env);
110+
var execArgv = cluster.settings.execArgv.slice();
111+
var debugPort = 0;
112+
113+
workerEnv = util._extend(workerEnv, env);
114+
workerEnv.NODE_UNIQUE_ID = '' + id;
115+
116+
for (var i = 0; i < execArgv.length; i++) {
117+
const match = execArgv[i].match(
118+
/^(--inspect|--debug|--debug-(brk|port))(=\d+)?$/
119+
);
120+
121+
if (match) {
122+
if (debugPort === 0) {
123+
debugPort = process.debugPort + debugPortOffset;
124+
++debugPortOffset;
125+
}
126+
127+
execArgv[i] = match[1] + '=' + debugPort;
128+
}
129+
}
130+
131+
return fork(cluster.settings.exec, cluster.settings.args, {
132+
env: workerEnv,
133+
silent: cluster.settings.silent,
134+
execArgv: execArgv,
135+
stdio: cluster.settings.stdio,
136+
gid: cluster.settings.gid,
137+
uid: cluster.settings.uid
138+
});
139+
}
140+
141+
function removeWorker(worker) {
142+
assert(worker);
143+
delete cluster.workers[worker.id];
144+
145+
if (keys(cluster.workers).length === 0) {
146+
assert(keys(handles).length === 0, 'Resource leak detected.');
147+
intercom.emit('disconnect');
148+
}
149+
}
150+
151+
function removeHandlesForWorker(worker) {
152+
assert(worker);
153+
154+
for (var key in handles) {
155+
const handle = handles[key];
156+
157+
if (handle.remove(worker))
158+
delete handles[key];
159+
}
160+
}
161+
162+
cluster.fork = function(env) {
163+
cluster.setupMaster();
164+
const id = ++ids;
165+
const workerProcess = createWorkerProcess(id, env);
166+
const worker = new Worker({
167+
id: id,
168+
process: workerProcess
169+
});
170+
171+
worker.on('message', function(message, handle) {
172+
cluster.emit('message', this, message, handle);
173+
});
174+
175+
worker.process.once('exit', (exitCode, signalCode) => {
176+
/*
177+
* Remove the worker from the workers list only
178+
* if it has disconnected, otherwise we might
179+
* still want to access it.
180+
*/
181+
if (!worker.isConnected()) {
182+
removeHandlesForWorker(worker);
183+
removeWorker(worker);
184+
}
185+
186+
worker.exitedAfterDisconnect = !!worker.exitedAfterDisconnect;
187+
worker.state = 'dead';
188+
worker.emit('exit', exitCode, signalCode);
189+
cluster.emit('exit', worker, exitCode, signalCode);
190+
});
191+
192+
worker.process.once('disconnect', () => {
193+
/*
194+
* Now is a good time to remove the handles
195+
* associated with this worker because it is
196+
* not connected to the master anymore.
197+
*/
198+
removeHandlesForWorker(worker);
199+
200+
/*
201+
* Remove the worker from the workers list only
202+
* if its process has exited. Otherwise, we might
203+
* still want to access it.
204+
*/
205+
if (worker.isDead())
206+
removeWorker(worker);
207+
208+
worker.exitedAfterDisconnect = !!worker.exitedAfterDisconnect;
209+
worker.state = 'disconnected';
210+
worker.emit('disconnect');
211+
cluster.emit('disconnect', worker);
212+
});
213+
214+
worker.process.on('internalMessage', internal(worker, onmessage));
215+
process.nextTick(emitForkNT, worker);
216+
cluster.workers[worker.id] = worker;
217+
return worker;
218+
};
219+
220+
function emitForkNT(worker) {
221+
cluster.emit('fork', worker);
222+
}
223+
224+
cluster.disconnect = function(cb) {
225+
const workers = keys(cluster.workers);
226+
227+
if (workers.length === 0) {
228+
process.nextTick(() => intercom.emit('disconnect'));
229+
} else {
230+
for (var key in workers) {
231+
key = workers[key];
232+
233+
if (cluster.workers[key].isConnected())
234+
cluster.workers[key].disconnect();
235+
}
236+
}
237+
238+
if (typeof cb === 'function')
239+
intercom.once('disconnect', cb);
240+
};
241+
242+
function onmessage(message, handle) {
243+
const worker = this;
244+
245+
if (message.act === 'online')
246+
online(worker);
247+
else if (message.act === 'queryServer')
248+
queryServer(worker, message);
249+
else if (message.act === 'listening')
250+
listening(worker, message);
251+
else if (message.act === 'exitedAfterDisconnect')
252+
exitedAfterDisconnect(worker, message);
253+
else if (message.act === 'close')
254+
close(worker, message);
255+
}
256+
257+
function online(worker) {
258+
worker.state = 'online';
259+
worker.emit('online');
260+
cluster.emit('online', worker);
261+
}
262+
263+
function exitedAfterDisconnect(worker, message) {
264+
worker.exitedAfterDisconnect = true;
265+
send(worker, { ack: message.seq });
266+
}
267+
268+
function queryServer(worker, message) {
269+
// Stop processing if worker already disconnecting
270+
if (worker.exitedAfterDisconnect)
271+
return;
272+
273+
const args = [message.address,
274+
message.port,
275+
message.addressType,
276+
message.fd,
277+
message.index];
278+
const key = args.join(':');
279+
var handle = handles[key];
280+
281+
if (handle === undefined) {
282+
var constructor = RoundRobinHandle;
283+
// UDP is exempt from round-robin connection balancing for what should
284+
// be obvious reasons: it's connectionless. There is nothing to send to
285+
// the workers except raw datagrams and that's pointless.
286+
if (schedulingPolicy !== SCHED_RR ||
287+
message.addressType === 'udp4' ||
288+
message.addressType === 'udp6') {
289+
constructor = SharedHandle;
290+
}
291+
292+
handles[key] = handle = new constructor(key,
293+
message.address,
294+
message.port,
295+
message.addressType,
296+
message.fd,
297+
message.flags);
298+
}
299+
300+
if (!handle.data)
301+
handle.data = message.data;
302+
303+
// Set custom server data
304+
handle.add(worker, (errno, reply, handle) => {
305+
reply = util._extend({
306+
errno: errno,
307+
key: key,
308+
ack: message.seq,
309+
data: handles[key].data
310+
}, reply);
311+
312+
if (errno)
313+
delete handles[key]; // Gives other workers a chance to retry.
314+
315+
send(worker, reply, handle);
316+
});
317+
}
318+
319+
function listening(worker, message) {
320+
const info = {
321+
addressType: message.addressType,
322+
address: message.address,
323+
port: message.port,
324+
fd: message.fd
325+
};
326+
327+
worker.state = 'listening';
328+
worker.emit('listening', info);
329+
cluster.emit('listening', worker, info);
330+
}
331+
332+
// Server in worker is closing, remove from list. The handle may have been
333+
// removed by a prior call to removeHandlesForWorker() so guard against that.
334+
function close(worker, message) {
335+
const key = message.key;
336+
const handle = handles[key];
337+
338+
if (handle && handle.remove(worker))
339+
delete handles[key];
340+
}
341+
342+
function send(worker, message, handle, cb) {
343+
return sendHelper(worker.process, message, handle, cb);
344+
}
345+
346+
// Extend generic Worker with methods specific to the master process.
347+
Worker.prototype.disconnect = function() {
348+
this.exitedAfterDisconnect = true;
349+
send(this, { act: 'disconnect' });
350+
removeHandlesForWorker(this);
351+
removeWorker(this);
352+
return this;
353+
};
354+
355+
Worker.prototype.destroy = function(signo) {
356+
const proc = this.process;
357+
358+
signo = signo || 'SIGTERM';
359+
360+
if (this.isConnected()) {
361+
this.once('disconnect', () => proc.kill(signo));
362+
this.disconnect();
363+
return;
364+
}
365+
366+
proc.kill(signo);
367+
};

‎lib/internal/cluster/round_robin_handle.js

+115
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
'use strict';
2+
const assert = require('assert');
3+
const net = require('net');
4+
const { sendHelper } = require('internal/cluster/utils');
5+
const getOwnPropertyNames = Object.getOwnPropertyNames;
6+
const uv = process.binding('uv');
7+
8+
module.exports = RoundRobinHandle;
9+
10+
function RoundRobinHandle(key, address, port, addressType, fd) {
11+
this.key = key;
12+
this.all = {};
13+
this.free = [];
14+
this.handles = [];
15+
this.handle = null;
16+
this.server = net.createServer(assert.fail);
17+
18+
if (fd >= 0)
19+
this.server.listen({ fd });
20+
else if (port >= 0)
21+
this.server.listen(port, address);
22+
else
23+
this.server.listen(address); // UNIX socket path.
24+
25+
this.server.once('listening', () => {
26+
this.handle = this.server._handle;
27+
this.handle.onconnection = (err, handle) => this.distribute(err, handle);
28+
this.server._handle = null;
29+
this.server = null;
30+
});
31+
}
32+
33+
RoundRobinHandle.prototype.add = function(worker, send) {
34+
assert(worker.id in this.all === false);
35+
this.all[worker.id] = worker;
36+
37+
const done = () => {
38+
if (this.handle.getsockname) {
39+
const out = {};
40+
this.handle.getsockname(out);
41+
// TODO(bnoordhuis) Check err.
42+
send(null, { sockname: out }, null);
43+
} else {
44+
send(null, null, null); // UNIX socket.
45+
}
46+
47+
this.handoff(worker); // In case there are connections pending.
48+
};
49+
50+
if (this.server === null)
51+
return done();
52+
53+
// Still busy binding.
54+
this.server.once('listening', done);
55+
this.server.once('error', (err) => {
56+
// Hack: translate 'EADDRINUSE' error string back to numeric error code.
57+
// It works but ideally we'd have some backchannel between the net and
58+
// cluster modules for stuff like this.
59+
const errno = uv['UV_' + err.errno];
60+
send(errno, null);
61+
});
62+
};
63+
64+
RoundRobinHandle.prototype.remove = function(worker) {
65+
if (worker.id in this.all === false)
66+
return false;
67+
68+
delete this.all[worker.id];
69+
const index = this.free.indexOf(worker);
70+
71+
if (index !== -1)
72+
this.free.splice(index, 1);
73+
74+
if (getOwnPropertyNames(this.all).length !== 0)
75+
return false;
76+
77+
for (var handle; handle = this.handles.shift(); handle.close())
78+
;
79+
80+
this.handle.close();
81+
this.handle = null;
82+
return true;
83+
};
84+
85+
RoundRobinHandle.prototype.distribute = function(err, handle) {
86+
this.handles.push(handle);
87+
const worker = this.free.shift();
88+
89+
if (worker)
90+
this.handoff(worker);
91+
};
92+
93+
RoundRobinHandle.prototype.handoff = function(worker) {
94+
if (worker.id in this.all === false) {
95+
return; // Worker is closing (or has closed) the server.
96+
}
97+
98+
const handle = this.handles.shift();
99+
100+
if (handle === undefined) {
101+
this.free.push(worker); // Add to ready queue again.
102+
return;
103+
}
104+
105+
const message = { act: 'newconn', key: this.key };
106+
107+
sendHelper(worker.process, message, handle, (reply) => {
108+
if (reply.accepted)
109+
handle.close();
110+
else
111+
this.distribute(0, handle); // Worker is shutting down. Send to another.
112+
113+
this.handoff(worker);
114+
});
115+
};

‎lib/internal/cluster/shared_handle.js

+48
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
'use strict';
2+
const assert = require('assert');
3+
const dgram = require('dgram');
4+
const net = require('net');
5+
6+
module.exports = SharedHandle;
7+
8+
function SharedHandle(key, address, port, addressType, fd, flags) {
9+
this.key = key;
10+
this.workers = [];
11+
this.handle = null;
12+
this.errno = 0;
13+
14+
// FIXME(bnoordhuis) Polymorphic return type for lack of a better solution.
15+
var rval;
16+
17+
if (addressType === 'udp4' || addressType === 'udp6')
18+
rval = dgram._createSocketHandle(address, port, addressType, fd, flags);
19+
else
20+
rval = net._createServerHandle(address, port, addressType, fd);
21+
22+
if (typeof rval === 'number')
23+
this.errno = rval;
24+
else
25+
this.handle = rval;
26+
}
27+
28+
SharedHandle.prototype.add = function(worker, send) {
29+
assert(this.workers.indexOf(worker) === -1);
30+
this.workers.push(worker);
31+
send(this.errno, null, this.handle);
32+
};
33+
34+
SharedHandle.prototype.remove = function(worker) {
35+
const index = this.workers.indexOf(worker);
36+
37+
if (index === -1)
38+
return false; // The worker wasn't sharing this handle.
39+
40+
this.workers.splice(index, 1);
41+
42+
if (this.workers.length !== 0)
43+
return false;
44+
45+
this.handle.close();
46+
this.handle = null;
47+
return true;
48+
};

‎lib/internal/cluster/utils.js

+44
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
'use strict';
2+
const util = require('util');
3+
4+
module.exports = {
5+
sendHelper,
6+
internal,
7+
handles: {} // Used in tests.
8+
};
9+
10+
const callbacks = {};
11+
var seq = 0;
12+
13+
function sendHelper(proc, message, handle, cb) {
14+
if (!proc.connected)
15+
return false;
16+
17+
// Mark message as internal. See INTERNAL_PREFIX in lib/child_process.js
18+
message = util._extend({ cmd: 'NODE_CLUSTER' }, message);
19+
20+
if (typeof cb === 'function')
21+
callbacks[seq] = cb;
22+
23+
message.seq = seq;
24+
seq += 1;
25+
return proc.send(message, handle);
26+
}
27+
28+
// Returns an internalMessage listener that hands off normal messages
29+
// to the callback but intercepts and redirects ACK messages.
30+
function internal(worker, cb) {
31+
return function onInternalMessage(message, handle) {
32+
if (message.cmd !== 'NODE_CLUSTER')
33+
return;
34+
35+
var fn = cb;
36+
37+
if (message.ack !== undefined && callbacks[message.ack] !== undefined) {
38+
fn = callbacks[message.ack];
39+
delete callbacks[message.ack];
40+
}
41+
42+
fn.apply(worker, arguments);
43+
};
44+
}

‎lib/internal/cluster/worker.js

+63
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
'use strict';
2+
const EventEmitter = require('events');
3+
const internalUtil = require('internal/util');
4+
const util = require('util');
5+
const defineProperty = Object.defineProperty;
6+
const suicideDeprecationMessage =
7+
'worker.suicide is deprecated. Please use worker.exitedAfterDisconnect.';
8+
9+
module.exports = Worker;
10+
11+
// Common Worker implementation shared between the cluster master and workers.
12+
function Worker(options) {
13+
if (!(this instanceof Worker))
14+
return new Worker(options);
15+
16+
EventEmitter.call(this);
17+
18+
if (options === null || typeof options !== 'object')
19+
options = {};
20+
21+
this.exitedAfterDisconnect = undefined;
22+
23+
defineProperty(this, 'suicide', {
24+
get: internalUtil.deprecate(
25+
() => this.exitedAfterDisconnect,
26+
suicideDeprecationMessage),
27+
set: internalUtil.deprecate(
28+
(val) => { this.exitedAfterDisconnect = val; },
29+
suicideDeprecationMessage),
30+
enumerable: true
31+
});
32+
33+
this.state = options.state || 'none';
34+
this.id = options.id | 0;
35+
36+
if (options.process) {
37+
this.process = options.process;
38+
this.process.on('error', (code, signal) =>
39+
this.emit('error', code, signal)
40+
);
41+
this.process.on('message', (message, handle) =>
42+
this.emit('message', message, handle)
43+
);
44+
}
45+
}
46+
47+
util.inherits(Worker, EventEmitter);
48+
49+
Worker.prototype.kill = function() {
50+
this.destroy.apply(this, arguments);
51+
};
52+
53+
Worker.prototype.send = function() {
54+
return this.process.send.apply(this.process, arguments);
55+
};
56+
57+
Worker.prototype.isDead = function() {
58+
return this.process.exitCode != null || this.process.signalCode != null;
59+
};
60+
61+
Worker.prototype.isConnected = function() {
62+
return this.process.connected;
63+
};

‎node.gyp

+6-1
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,12 @@
7676
'lib/zlib.js',
7777
'lib/internal/buffer.js',
7878
'lib/internal/child_process.js',
79-
'lib/internal/cluster.js',
79+
'lib/internal/cluster/child.js',
80+
'lib/internal/cluster/master.js',
81+
'lib/internal/cluster/round_robin_handle.js',
82+
'lib/internal/cluster/shared_handle.js',
83+
'lib/internal/cluster/utils.js',
84+
'lib/internal/cluster/worker.js',
8085
'lib/internal/freelist.js',
8186
'lib/internal/fs.js',
8287
'lib/internal/linkedlist.js',

‎test/parallel/test-cluster-disconnect-handles.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ cluster.schedulingPolicy = cluster.SCHED_RR;
2424
// pending handle queue.
2525
if (cluster.isMaster) {
2626
let isKilling = false;
27-
const handles = require('internal/cluster').handles;
27+
const handles = require('internal/cluster/utils').handles;
2828
const address = common.hasIPv6 ? '[::1]' : common.localhostIPv4;
2929
cluster.setupMaster({ execArgv: [`--debug=${address}:${common.PORT}`] });
3030
const worker = cluster.fork();

0 commit comments

Comments
 (0)
Please sign in to comment.