Browse Source

net: fix listen() regression, revert patches

This commit reverts the following commits (in reverse chronological order):

  74d076c errnoException must be done immediately
  ddb02b9 net: support Server.listen(Pipe)
  085a098 cluster: do not use internal server API
  d138875 net: lazy listen on handler

Commit d138875 introduced a backwards incompatible change that broke the
simple/test-net-socket-timeout and simple/test-net-lazy-listen tests - it
defers listening on the target port until the `net.Server` instance has at
least one 'connection' event listener.

The other patches had to be reverted in order to revert d138875.

Fixes #3832.
v0.9.1-release
Ben Noordhuis 13 years ago
parent
commit
4c150ca0d0
  1. 5
      doc/api/child_process.markdown
  2. 2
      doc/api/net.markdown
  3. 2
      lib/child_process.js
  4. 120
      lib/cluster.js
  5. 66
      lib/net.js
  6. 3
      test/simple/test-cluster-basic.js
  7. 182
      test/simple/test-net-lazy-listen.js

5
doc/api/child_process.markdown

@ -191,9 +191,8 @@ And the child would the recive the server object as:
}
});
Note that the server is now shared between the parent and child. This means
that any connections will be load balanced between the servers provided there
is a `connection` listener.
Note that the server is now shared between the parent and child, this means
that some connections will be handled by the parent and some by the child.
**send socket object**

2
doc/api/net.markdown

@ -10,8 +10,6 @@ this module with `require('net');`
Creates a new TCP server. The `connectionListener` argument is
automatically set as a listener for the ['connection'][] event.
If the server has no `connection` listener it will not accept
connections until it gets one.
`options` is an object with the following defaults:

2
lib/child_process.js

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

120
lib/cluster.js

@ -159,20 +159,24 @@ 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);
} else {
}
// Send respond if it hasn't been called yet
if (respondUsed === false) {
respond();
}
}
@ -181,13 +185,11 @@ function handleMessage(worker, inMessage, inHandle) {
if (cluster.isMaster) {
// Handle online messages from workers
messageHandler.online = function(message, worker, send) {
messageHandler.online = function(message, worker) {
worker.state = 'online';
debug('Worker ' + worker.process.pid + ' online');
worker.emit('online');
cluster.emit('online', worker);
send();
};
// Handle queryServer messages from workers
@ -195,38 +197,46 @@ if (cluster.isMaster) {
// This sequence of information is unique to the connection
// but not to the worker
var args = message.args;
var key = JSON.stringify(args);
var args = [message.address,
message.port,
message.addressType,
message.fd];
var key = args.join(':');
var handler;
if (serverHandlers.hasOwnProperty(key)) {
send({}, serverHandlers[key]);
return;
handler = serverHandlers[key];
} else {
handler = serverHandlers[key] = net._createServerHandle.apply(net, args);
}
var server = serverHandlers[key] = net.Server();
server.once('listening', function() {
send({}, server);
});
server.listen.apply(server, args);
// echo callback with the fd handler associated with it
send({}, handler);
};
// Handle listening messages from workers
messageHandler.listening = function(message, worker, send) {
messageHandler.listening = function(message, worker) {
worker.state = 'listening';
// Emit listening, now that we know the worker is listening
worker.emit('listening', message.address);
cluster.emit('listening', worker, message.address);
send();
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
});
};
// Handle suicide messages from workers
messageHandler.suicide = function(message, worker, send) {
messageHandler.suicide = function(message, worker) {
worker.suicide = true;
send();
};
}
@ -235,9 +245,8 @@ if (cluster.isMaster) {
else if (cluster.isWorker) {
// Handle worker.disconnect from master
messageHandler.disconnect = function(message, worker, send) {
messageHandler.disconnect = function(message, worker) {
worker.disconnect();
send();
};
}
@ -512,37 +521,38 @@ cluster._setupWorker = function() {
};
// Internal function. Called by lib/net.js when attempting to bind a server.
if (cluster.isWorker) {
var localListen = net.Server.prototype.listen;
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());
}
// 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);
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
});
});
// request server
var message = { cmd: 'queryServer', args: args };
sendInternalMessage(cluster.worker, message, function(msg, server) {
localListen.call(self, server);
});
// Request the fd handler from the master process
var message = {
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, handle) {
cb(handle);
});
};

66
lib/net.js

@ -834,7 +834,8 @@ exports.Server = Server;
function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
var createServerHandle = function(address, port, addressType, fd) {
var createServerHandle = exports._createServerHandle =
function(address, port, addressType, fd) {
var r = 0;
// assign handle in listen, and clean up if bind or listen fails
var handle;
@ -895,6 +896,18 @@ Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
var self = this;
var r = 0;
// 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) {
process.nextTick(function() {
self.emit('error', errnoException(errno, 'listen'));
});
return;
}
}
self._handle.onconnection = onconnection;
self._handle.owner = self;
@ -910,55 +923,33 @@ Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
process.nextTick(function() {
self.emit('error', ex);
});
return r;
return;
}
// generate connection key, this should be unique to the connection
this._connectionKey = addressType + ':' + address + ':' + port;
return r;
process.nextTick(function() {
self.emit('listening');
});
};
function listen(self, address, port, addressType, backlog, fd) {
// 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) {
var error = errnoException(errno, 'listen');
process.nextTick(function() {
self.emit('error', error);
});
return;
}
}
if (!cluster) cluster = require('cluster');
// 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);
if (cluster.isWorker) {
cluster._getServer(self, address, port, addressType, fd, function(handle) {
self._handle = handle;
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');
});
} else {
self._listen2(address, port, addressType, backlog, fd);
}
}
Server.prototype._listen = function() {
Server.prototype.listen = function() {
var self = this;
var lastArg = arguments[arguments.length - 1];
@ -973,7 +964,6 @@ Server.prototype._listen = function() {
var backlog = toNumber(arguments[1]) || toNumber(arguments[2]);
var TCP = process.binding('tcp_wrap').TCP;
var Pipe = process.binding('pipe_wrap').Pipe;
if (arguments.length == 0 || typeof arguments[0] == 'function') {
// Bind to a random port.
@ -986,7 +976,7 @@ Server.prototype._listen = function() {
} else if (h.handle) {
h = h.handle;
}
if (h instanceof TCP || h instanceof Pipe) {
if (h instanceof TCP) {
self._handle = h;
listen(self, null, -1, -1, backlog);
} else if (typeof h.fd === 'number' && h.fd >= 0) {
@ -1018,8 +1008,6 @@ 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();

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

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

182
test/simple/test-net-lazy-listen.js

@ -1,182 +0,0 @@
// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.
var common = require('../common');
var net = require('net');
var fork = require('child_process').fork;
if (process.argv[2] === 'worker') {
process.once('disconnect', function () {
process.exit(0);
});
process.on('message', function (msg, server) {
if (msg !== 'server') return;
server.on('connection', function (socket) {
socket.end(process.argv[3]);
});
process.send('listen');
});
process.send('online');
return;
}
var workers = [];
function spawn(server, id, cb) {
// create a worker
var worker = fork(__filename, ['worker', id]);
workers[id] = worker;
// wait for ready
worker.on('message', function (msg) {
switch (msg) {
case 'online':
worker.send('server', server);
break;
case 'listen':
cb(worker);
break;
default:
throw new Error('got wrong message' + msg);
}
});
return worker;
}
// create a server instance there don't take connections
var server = net.createServer().listen(common.PORT, function () {
setTimeout(function() {
console.log('spawn worker#0');
spawn(server, 0, function () {
console.log('worker#0 spawned');
});
}, 250);
console.log('testing for standby, expect id 0');
testResponse([0], function () {
// test with two worker, expect id response to be 0 or 1
testNewWorker(1, [0, 1], function () {
// kill worker#0, expect id response to be 1
testWorkerDeath(0, [1], function () {
// close server, expect all connections to continue
testServerClose(server, [1], function () {
// killing worker#1, expect all connections to fail
testWorkerDeath(1, [], function () {
console.log('test complete');
});
});
});
});
});
});
function testServerClose(server, expect, cb) {
console.log('closeing server');
server.close(function () {
testResponse(expect, cb);
});
}
function testNewWorker(workerId, exptect, cb) {
console.log('spawning worker#' + workerId);
spawn(server, 1, function () {
testResponse(exptect, cb);
});
}
function testWorkerDeath(workerID, exptect, cb) {
// killing worker#1, expect all connections to fail
console.log('killing worker#' + workerID);
workers[workerID].kill();
workers[workerID].once('exit', function () {
testResponse(exptect, cb);
});
}
function testResponse(expect, cb) {
// make 25 connections
var count = 25;
var missing = expect.slice(0);
var i = count;
while (i--) {
makeConnection(function (error, id) {
if (expect.length === 0) {
if (error === null) {
throw new Error('connect should not be possible');
}
} else {
if (expect.indexOf(id) === -1) {
throw new Error('got unexpected response: ' + id +
', expected: ' + expect.join(', '));
}
var index = missing.indexOf(id);
if (index !== -1) missing.splice(index, 1);
}
count -= 1;
if (count === 0) {
if (missing.length !== 0) {
throw new Error('no connection responsed with the following id: ' +
missing.join(', '));
}
cb();
}
});
}
}
function makeConnection(cb) {
var client = net.connect({ port: common.PORT });
var id = null;
client.once('data', function(data) {
id = parseInt(data, 10);
});
var error = null;
client.once('error', function (e) {
error = e;
});
client.setTimeout(500);
client.on('close', function () {
cb(error, id);
});
}
process.on('exit', function () {
workers.forEach(function (worker) {
try { worker.kill(); } catch(e) {}
});
});
Loading…
Cancel
Save