Skip to content
This repository has been archived by the owner on Apr 22, 2023. It is now read-only.

Commit

Permalink
cluster: do not use internal server API
Browse files Browse the repository at this point in the history
  • Loading branch information
@AndreasMadsen @isaacs
AndreasMadsen authored and isaacs committed Aug 5, 2012
1 parent d138875 commit 085a098
Show file tree
Hide file tree
Showing 4 changed files with 88 additions and 111 deletions.
2 changes: 1 addition & 1 deletion lib/child_process.js
View file
Expand Up @@ -101,7 +101,7 @@ var handleConversion = {
var self = this;

var server = new net.Server();
server.listen(handle, function() {
server._listen(handle, function() {
emit(server);
});
}
Expand Down
120 changes: 55 additions & 65 deletions lib/cluster.js
View file
Expand Up @@ -159,24 +159,20 @@ function handleResponse(outMessage, outHandle, inMessage, inHandle, worker) {
// Handle messages from both master and workers
var messageHandler = {};
function handleMessage(worker, inMessage, inHandle) {
if (!isInternalMessage(inMessage)) return;

// Remove internal prefix
var message = util._extend({}, inMessage);
message.cmd = inMessage.cmd.substr(INTERNAL_PREFIX.length);

var respondUsed = false;
function respond(outMessage, outHandler) {
respondUsed = true;
handleResponse(outMessage, outHandler, inMessage, inHandle, worker);
}

// Run handler if it exists
if (messageHandler[message.cmd]) {
messageHandler[message.cmd](message, worker, respond);
}

// Send respond if it hasn't been called yet
if (respondUsed === false) {
} else {
respond();
}
}
Expand All @@ -185,58 +181,52 @@ function handleMessage(worker, inMessage, inHandle) {
if (cluster.isMaster) {

// Handle online messages from workers
messageHandler.online = function(message, worker) {
messageHandler.online = function(message, worker, send) {
worker.state = 'online';
debug('Worker ' + worker.process.pid + ' online');
worker.emit('online');
cluster.emit('online', worker);

send();
};

// Handle queryServer messages from workers
messageHandler.queryServer = function(message, worker, send) {

// This sequence of information is unique to the connection
// but not to the worker
var args = [message.address,
message.port,
message.addressType,
message.fd];
var key = args.join(':');
var handler;
var args = message.args;
var key = JSON.stringify(args);

if (serverHandlers.hasOwnProperty(key)) {
handler = serverHandlers[key];
} else {
handler = serverHandlers[key] = net._createServerHandle.apply(net, args);
send({}, serverHandlers[key]);
return;
}

// echo callback with the fd handler associated with it
send({}, handler);
var server = serverHandlers[key] = net.Server();
server.once('listening', function() {
send({}, server);
});
server.listen.apply(server, args);
};

// Handle listening messages from workers
messageHandler.listening = function(message, worker) {
messageHandler.listening = function(message, worker, send) {

worker.state = 'listening';

// Emit listening, now that we know the worker is listening
worker.emit('listening', {
address: message.address,
port: message.port,
addressType: message.addressType,
fd: message.fd
});
cluster.emit('listening', worker, {
address: message.address,
port: message.port,
addressType: message.addressType,
fd: message.fd
});
worker.emit('listening', message.address);
cluster.emit('listening', worker, message.address);

send();
};

// Handle suicide messages from workers
messageHandler.suicide = function(message, worker) {
messageHandler.suicide = function(message, worker, send) {
worker.suicide = true;

send();
};

}
Expand All @@ -245,8 +235,9 @@ if (cluster.isMaster) {
else if (cluster.isWorker) {

// Handle worker.disconnect from master
messageHandler.disconnect = function(message, worker) {
messageHandler.disconnect = function(message, worker, send) {
worker.disconnect();
send();
};
}

Expand Down Expand Up @@ -521,38 +512,37 @@ cluster._setupWorker = function() {
};

// Internal function. Called by lib/net.js when attempting to bind a server.
cluster._getServer = function(tcpSelf, address, port, addressType, fd, cb) {
// This can only be called from a worker.
assert(cluster.isWorker);

// Store tcp instance for later use
var key = [address, port, addressType, fd].join(':');
serverListeners[key] = tcpSelf;

// Send a listening message to the master
tcpSelf.once('listening', function() {
cluster.worker.state = 'listening';
sendInternalMessage(cluster.worker, {
cmd: 'listening',
address: address,
port: port,
addressType: addressType,
fd: fd
});
});
if (cluster.isWorker) {
var localListen = net.Server.prototype.listen;
net.Server.prototype.listen = function() {
var self = this;

// Request the fd handler from the master process
var message = {
cmd: 'queryServer',
address: address,
port: port,
addressType: addressType,
fd: fd
};
var args = new Array(arguments.length);
for (var i = 0; i < arguments.length; i++) {
args[i] = arguments[i];
}

// The callback will be stored until the master has responded
sendInternalMessage(cluster.worker, message, function(msg, handle) {
cb(handle);
});
// filter out callback
if (typeof args[args.length - 1] === 'function') {
this.once('listening', args.pop());
}

};
// add server (used by. dissconnect)
serverListeners[JSON.stringify(args)] = this;

// send callback to master, telling that worker is listening
this.once('listening', function() {
cluster.worker.state = 'listening';

var message = { cmd: 'listening', address: this.address() };
sendInternalMessage(cluster.worker, message);
});

// request server
var message = { cmd: 'queryServer', args: args };

sendInternalMessage(cluster.worker, message, function(msg, server) {
localListen.call(self, server);
});
};
}
74 changes: 31 additions & 43 deletions lib/net.js
View file
Expand Up @@ -834,8 +834,7 @@ exports.Server = Server;
function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }


var createServerHandle = exports._createServerHandle =
function(address, port, addressType, fd) {
var createServerHandle = function(address, port, addressType, fd) {
var r = 0;
// assign handle in listen, and clean up if bind or listen fails
var handle;
Expand Down Expand Up @@ -922,56 +921,43 @@ Server.prototype._listen2 = function(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.
// In the case of a server sent via IPC, we don't need to do this.
// 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.
if (!self._handle) {
self._handle = createServerHandle(address, port, addressType, fd);
if (!self._handle) {
self._handle = createServerHandle(address, port, addressType, fd);
if (!self._handle) {
process.nextTick(function() {
self.emit('error', errnoException(errno, 'listen'));
});
return;
}
}

// self._handle.listen will be called lazy
// if there are no connection listeners
if (self.listeners('connection').length === 0) {
self.on('newListener', function removeme(name) {
if (name !== 'connection') return;

self.removeListener('newListener', removeme);
self._listen2(address, port, addressType, backlog, fd);
});

process.nextTick(function() {
self.emit('listening');
self.emit('error', errnoException(errno, 'listen'));
});
return;
}
}

var r = self._listen2(address, port, addressType, backlog, fd);
if (r === 0) {
process.nextTick(function() {
self.emit('listening');
});
}
// self._handle.listen will be called lazy
// if there are no connection listeners
if (self.listeners('connection').length === 0) {
self.on('newListener', function removeme(name) {
if (name !== 'connection') return;

self.removeListener('newListener', removeme);
self._listen2(address, port, addressType, backlog, fd);
});

process.nextTick(function() {
self.emit('listening');
});
return;
}
}

var r = self._listen2(address, port, addressType, backlog, fd);
if (r === 0) {
process.nextTick(function() {
self.emit('listening');
});
}
}

Server.prototype.listen = function() {
Server.prototype._listen = function() {
var self = this;

var lastArg = arguments[arguments.length - 1];
Expand Down Expand Up @@ -1030,6 +1016,8 @@ Server.prototype.listen = function() {
return self;
};

Server.prototype.listen = Server.prototype._listen;

Server.prototype.address = function() {
if (this._handle && this._handle.getsockname) {
return this._handle.getsockname();
Expand Down
3 changes: 1 addition & 2 deletions test/simple/test-cluster-basic.js
View file
Expand Up @@ -136,8 +136,7 @@ else if (cluster.isMaster) {
assert.equal(arguments.length, 1);
var expect = { address: '127.0.0.1',
port: common.PORT,
addressType: 4,
fd: undefined };
family: 'IPv4'};
assert.deepEqual(arguments[0], expect);
break;

Expand Down

0 comments on commit 085a098

Please sign in to comment.