Browse Source

cluster: do not use internal server API

v0.9.1-release
Andreas Madsen 13 years ago
committed by isaacs
parent
commit
085a09874b
  1. 2
      lib/child_process.js
  2. 110
      lib/cluster.js
  3. 20
      lib/net.js
  4. 3
      test/simple/test-cluster-basic.js

2
lib/child_process.js

@ -101,7 +101,7 @@ var handleConversion = {
var self = this; var self = this;
var server = new net.Server(); var server = new net.Server();
server.listen(handle, function() { server._listen(handle, function() {
emit(server); emit(server);
}); });
} }

110
lib/cluster.js

@ -159,24 +159,20 @@ function handleResponse(outMessage, outHandle, inMessage, inHandle, worker) {
// Handle messages from both master and workers // Handle messages from both master and workers
var messageHandler = {}; var messageHandler = {};
function handleMessage(worker, inMessage, inHandle) { function handleMessage(worker, inMessage, inHandle) {
if (!isInternalMessage(inMessage)) return;
// Remove internal prefix // Remove internal prefix
var message = util._extend({}, inMessage); var message = util._extend({}, inMessage);
message.cmd = inMessage.cmd.substr(INTERNAL_PREFIX.length); message.cmd = inMessage.cmd.substr(INTERNAL_PREFIX.length);
var respondUsed = false;
function respond(outMessage, outHandler) { function respond(outMessage, outHandler) {
respondUsed = true;
handleResponse(outMessage, outHandler, inMessage, inHandle, worker); handleResponse(outMessage, outHandler, inMessage, inHandle, worker);
} }
// Run handler if it exists // Run handler if it exists
if (messageHandler[message.cmd]) { if (messageHandler[message.cmd]) {
messageHandler[message.cmd](message, worker, respond); messageHandler[message.cmd](message, worker, respond);
} } else {
// Send respond if it hasn't been called yet
if (respondUsed === false) {
respond(); respond();
} }
} }
@ -185,11 +181,13 @@ function handleMessage(worker, inMessage, inHandle) {
if (cluster.isMaster) { if (cluster.isMaster) {
// Handle online messages from workers // Handle online messages from workers
messageHandler.online = function(message, worker) { messageHandler.online = function(message, worker, send) {
worker.state = 'online'; worker.state = 'online';
debug('Worker ' + worker.process.pid + ' online'); debug('Worker ' + worker.process.pid + ' online');
worker.emit('online'); worker.emit('online');
cluster.emit('online', worker); cluster.emit('online', worker);
send();
}; };
// Handle queryServer messages from workers // Handle queryServer messages from workers
@ -197,46 +195,38 @@ if (cluster.isMaster) {
// This sequence of information is unique to the connection // This sequence of information is unique to the connection
// but not to the worker // but not to the worker
var args = [message.address, var args = message.args;
message.port, var key = JSON.stringify(args);
message.addressType,
message.fd];
var key = args.join(':');
var handler;
if (serverHandlers.hasOwnProperty(key)) { if (serverHandlers.hasOwnProperty(key)) {
handler = serverHandlers[key]; send({}, serverHandlers[key]);
} else { return;
handler = serverHandlers[key] = net._createServerHandle.apply(net, args);
} }
// echo callback with the fd handler associated with it var server = serverHandlers[key] = net.Server();
send({}, handler); server.once('listening', function() {
send({}, server);
});
server.listen.apply(server, args);
}; };
// Handle listening messages from workers // Handle listening messages from workers
messageHandler.listening = function(message, worker) { messageHandler.listening = function(message, worker, send) {
worker.state = 'listening'; worker.state = 'listening';
// Emit listening, now that we know the worker is listening // Emit listening, now that we know the worker is listening
worker.emit('listening', { worker.emit('listening', message.address);
address: message.address, cluster.emit('listening', worker, message.address);
port: message.port,
addressType: message.addressType, send();
fd: message.fd
});
cluster.emit('listening', worker, {
address: message.address,
port: message.port,
addressType: message.addressType,
fd: message.fd
});
}; };
// Handle suicide messages from workers // Handle suicide messages from workers
messageHandler.suicide = function(message, worker) { messageHandler.suicide = function(message, worker, send) {
worker.suicide = true; worker.suicide = true;
send();
}; };
} }
@ -245,8 +235,9 @@ if (cluster.isMaster) {
else if (cluster.isWorker) { else if (cluster.isWorker) {
// Handle worker.disconnect from master // Handle worker.disconnect from master
messageHandler.disconnect = function(message, worker) { messageHandler.disconnect = function(message, worker, send) {
worker.disconnect(); worker.disconnect();
send();
}; };
} }
@ -521,38 +512,37 @@ cluster._setupWorker = function() {
}; };
// Internal function. Called by lib/net.js when attempting to bind a server. // Internal function. Called by lib/net.js when attempting to bind a server.
cluster._getServer = function(tcpSelf, address, port, addressType, fd, cb) { if (cluster.isWorker) {
// This can only be called from a worker. var localListen = net.Server.prototype.listen;
assert(cluster.isWorker); net.Server.prototype.listen = function() {
var self = this;
var args = new Array(arguments.length);
for (var i = 0; i < arguments.length; i++) {
args[i] = arguments[i];
}
// filter out callback
if (typeof args[args.length - 1] === 'function') {
this.once('listening', args.pop());
}
// Store tcp instance for later use // add server (used by. dissconnect)
var key = [address, port, addressType, fd].join(':'); serverListeners[JSON.stringify(args)] = this;
serverListeners[key] = tcpSelf;
// Send a listening message to the master // send callback to master, telling that worker is listening
tcpSelf.once('listening', function() { this.once('listening', function() {
cluster.worker.state = 'listening'; cluster.worker.state = 'listening';
sendInternalMessage(cluster.worker, {
cmd: 'listening', var message = { cmd: 'listening', address: this.address() };
address: address, sendInternalMessage(cluster.worker, message);
port: port,
addressType: addressType,
fd: fd
});
}); });
// Request the fd handler from the master process // request server
var message = { var message = { cmd: 'queryServer', args: args };
cmd: 'queryServer',
address: address,
port: port,
addressType: addressType,
fd: fd
};
// The callback will be stored until the master has responded sendInternalMessage(cluster.worker, message, function(msg, server) {
sendInternalMessage(cluster.worker, message, function(msg, handle) { localListen.call(self, server);
cb(handle);
}); });
}; };
}

20
lib/net.js

@ -834,8 +834,7 @@ exports.Server = Server;
function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; } function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
var createServerHandle = exports._createServerHandle = var createServerHandle = function(address, port, addressType, fd) {
function(address, port, addressType, fd) {
var r = 0; var r = 0;
// assign handle in listen, and clean up if bind or listen fails // assign handle in listen, and clean up if bind or listen fails
var handle; var handle;
@ -922,17 +921,6 @@ Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
function listen(self, address, port, addressType, backlog, fd) { function listen(self, address, port, addressType, backlog, fd) {
if (!cluster) cluster = require('cluster');
if (cluster.isWorker) {
cluster._getServer(self, address, port, addressType, fd, function(handle) {
self._handle = handle;
var r = self._listen2(address, port, addressType, backlog, fd);
if (r === 0) {
self.emit('listening');
}
});
} else {
// If there is not yet a handle, we need to create one and bind. // If there is not yet a handle, we need to create one and bind.
// In the case of a server sent via IPC, we don't need to do this. // In the case of a server sent via IPC, we don't need to do this.
if (!self._handle) { if (!self._handle) {
@ -968,10 +956,8 @@ function listen(self, address, port, addressType, backlog, fd) {
}); });
} }
} }
}
Server.prototype._listen = function() {
Server.prototype.listen = function() {
var self = this; var self = this;
var lastArg = arguments[arguments.length - 1]; var lastArg = arguments[arguments.length - 1];
@ -1030,6 +1016,8 @@ Server.prototype.listen = function() {
return self; return self;
}; };
Server.prototype.listen = Server.prototype._listen;
Server.prototype.address = function() { Server.prototype.address = function() {
if (this._handle && this._handle.getsockname) { if (this._handle && this._handle.getsockname) {
return this._handle.getsockname(); return this._handle.getsockname();

3
test/simple/test-cluster-basic.js

@ -136,8 +136,7 @@ else if (cluster.isMaster) {
assert.equal(arguments.length, 1); assert.equal(arguments.length, 1);
var expect = { address: '127.0.0.1', var expect = { address: '127.0.0.1',
port: common.PORT, port: common.PORT,
addressType: 4, family: 'IPv4'};
fd: undefined };
assert.deepEqual(arguments[0], expect); assert.deepEqual(arguments[0], expect);
break; break;

Loading…
Cancel
Save