AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Close our bound socket when the connection manager is shutdown. New-styled

Browse Source 
logging.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4727 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2007-06-13 17:48:42 +00:00
parent e7379cd41a
commit 6db6eef7e1
1 changed files with 61 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/server/net/ConnectionManager.java
+61 -55
View File
@@ -37,14 +37,13 @@ import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.Set; import java.util.Set;
import java.util.logging.Level;
import com.samskivert.util.*; import com.samskivert.util.*;
import com.threerings.io.FramingOutputStream; import com.threerings.io.FramingOutputStream;
import com.threerings.io.ObjectOutputStream; import com.threerings.io.ObjectOutputStream;
import com.threerings.presents.Log;
import com.threerings.presents.data.ConMgrStats; import com.threerings.presents.data.ConMgrStats;
import com.threerings.presents.net.AuthRequest; import com.threerings.presents.net.AuthRequest;
import com.threerings.presents.net.AuthResponse; import com.threerings.presents.net.AuthResponse;
@@ -53,6 +52,8 @@ import com.threerings.presents.net.DownstreamMessage;
import com.threerings.presents.server.Authenticator; import com.threerings.presents.server.Authenticator;
import com.threerings.presents.server.PresentsServer; import com.threerings.presents.server.PresentsServer;
import static com.threerings.presents.Log.log;
/** /**
* The connection manager manages the socket on which connections are received. It creates * The connection manager manages the socket on which connections are received. It creates
* connection objects to manage each individual connection, but those connection objects interact * connection objects to manage each individual connection, but those connection objects interact
@@ -260,18 +261,18 @@ public class ConnectionManager extends LoopingThread
for (int ii = 0; ii < _ports.length; ii++) { for (int ii = 0; ii < _ports.length; ii++) {
try { try {
// create a listening socket and add it to the select set // create a listening socket and add it to the select set
ServerSocketChannel listener = ServerSocketChannel.open(); _ssocket = ServerSocketChannel.open();
listener.configureBlocking(false); _ssocket.configureBlocking(false);
InetSocketAddress isa = new InetSocketAddress(_ports[ii]); InetSocketAddress isa = new InetSocketAddress(_ports[ii]);
listener.socket().bind(isa); _ssocket.socket().bind(isa);
registerChannel(listener); registerChannel(_ssocket);
successes++; successes++;
Log.info("Server listening on " + isa + "."); log.info("Server listening on " + isa + ".");
} catch (IOException ioe) { } catch (IOException ioe) {
Log.warning("Failure listening to socket on port '" +_ports[ii] + "'."); log.log(Level.WARNING, "Failure listening to socket on port '" +
Log.logStackTrace(ioe); _ports[ii] + "'.", ioe);
_failure = ioe; _failure = ioe;
} }
} }
@@ -287,20 +288,19 @@ public class ConnectionManager extends LoopingThread
// try { // try {
// Channel inherited = System.inheritedChannel(); // Channel inherited = System.inheritedChannel();
// if (inherited instanceof ServerSocketChannel) { // if (inherited instanceof ServerSocketChannel) {
// ServerSocketChannel listener = (ServerSocketChannel)inherited; // _ssocket = (ServerSocketChannel)inherited;
// listener.configureBlocking(false); // _ssocket.configureBlocking(false);
// registerChannel(listener); // registerChannel(_ssocket);
// successes++; // successes++;
// Log.info("Server listening on " + // log.info("Server listening on " +
// listener.socket().getInetAddress() + ":" + // _ssocket.socket().getInetAddress() + ":" +
// listener.socket().getLocalPort() + "."); // _ssocket.socket().getLocalPort() + ".");
// } else if (inherited != null) { // } else if (inherited != null) {
// Log.warning("Inherited non-server-socket channel " + // log.warning("Inherited non-server-socket channel " + inherited + ".");
// inherited + ".");
// } // }
// } catch (IOException ioe) { // } catch (IOException ioe) {
// Log.warning("Failed to check for inherited channel."); // log.warning("Failed to check for inherited channel.");
// } // }
// if we failed to listen on at least one port, give up the ghost // if we failed to listen on at least one port, give up the ghost
@@ -394,7 +394,7 @@ public class ConnectionManager extends LoopingThread
if (oqueue != null) { if (oqueue != null) {
int size = oqueue.size(); int size = oqueue.size();
if ((size > 500) && (size % 50 == 0)) { if ((size > 500) && (size % 50 == 0)) {
Log.warning("Aiya, big overflow queue for " + conn + " [size=" + size + log.warning("Aiya, big overflow queue for " + conn + " [size=" + size +
", adding=" + tup.right + "]."); ", adding=" + tup.right + "].");
} }
oqueue.add(tup.right); oqueue.add(tup.right);
@@ -433,15 +433,14 @@ public class ConnectionManager extends LoopingThread
conn.getAuthRequest(), conn.getAuthResponse()); conn.getAuthRequest(), conn.getAuthResponse());
} catch (IOException ioe) { } catch (IOException ioe) {
Log.warning("Failure upgrading authing connection to running connection."); log.log(Level.WARNING, "Failure upgrading authing connection to running.", ioe);
Log.logStackTrace(ioe);
} }
} }
Set<SelectionKey> ready = null; Set<SelectionKey> ready = null;
try { try {
// check for incoming network events // check for incoming network events
// Log.debug("Selecting from " + StringUtil.toString(_selector.keys()) + " (" + // log.fine("Selecting from " + StringUtil.toString(_selector.keys()) + " (" +
// SELECT_LOOP_TIME + ")."); // SELECT_LOOP_TIME + ").");
int ecount = _selector.select(SELECT_LOOP_TIME); int ecount = _selector.select(SELECT_LOOP_TIME);
ready = _selector.selectedKeys(); ready = _selector.selectedKeys();
@@ -449,16 +448,17 @@ public class ConnectionManager extends LoopingThread
if (ready.size() == 0) { if (ready.size() == 0) {
return; return;
} else { } else {
Log.warning("select() returned no selected sockets, but there are " + log.warning("select() returned no selected sockets, but there are " +
ready.size() + " in the ready set."); ready.size() + " in the ready set.");
} }
} }
} catch (IOException ioe) { } catch (IOException ioe) {
Log.warning("Failure select()ing [ioe=" + ioe + "].");
if ("Invalid argument".equals(ioe.getMessage())) { if ("Invalid argument".equals(ioe.getMessage())) {
// what is this, anyway? // what is this, anyway?
Log.logStackTrace(ioe); log.log(Level.WARNING, "Failure select()ing.", ioe);
} else {
log.warning("Failure select()ing [ioe=" + ioe + "].");
} }
return; return;
@@ -466,10 +466,9 @@ public class ConnectionManager extends LoopingThread
// this block of code deals with a bug in the _selector that we observed on 2005-05-02, // this block of code deals with a bug in the _selector that we observed on 2005-05-02,
// instead of looping indefinitely after things go pear-shaped, shut us down in an // instead of looping indefinitely after things go pear-shaped, shut us down in an
// orderly fashion // orderly fashion
Log.warning("Failure select()ing [re=" + re + "]."); log.log(Level.WARNING, "Failure select()ing.", re);
Log.logStackTrace(re);
if (_runtimeExceptionCount++ >= 20) { if (_runtimeExceptionCount++ >= 20) {
Log.warning("Too many errors, bailing."); log.warning("Too many errors, bailing.");
shutdown(); shutdown();
} }
return; return;
@@ -478,13 +477,13 @@ public class ConnectionManager extends LoopingThread
_runtimeExceptionCount = 0; _runtimeExceptionCount = 0;
// process those events // process those events
// Log.info("Ready set " + StringUtil.toString(ready) + "."); // log.info("Ready set " + StringUtil.toString(ready) + ".");
for (SelectionKey selkey : ready) { for (SelectionKey selkey : ready) {
NetEventHandler handler = null; NetEventHandler handler = null;
try { try {
handler = _handlers.get(selkey); handler = _handlers.get(selkey);
if (handler == null) { if (handler == null) {
Log.warning("Received network event but have no registered handler " + log.warning("Received network event but have no registered handler " +
"[selkey=" + selkey + "]."); "[selkey=" + selkey + "].");
// request that this key be removed from our selection set, which normally // request that this key be removed from our selection set, which normally
// happens automatically but for some reason didn't // happens automatically but for some reason didn't
@@ -492,7 +491,7 @@ public class ConnectionManager extends LoopingThread
continue; continue;
} }
// Log.info("Got event [selkey=" + selkey + ", handler=" + handler + "]."); // log.info("Got event [selkey=" + selkey + ", handler=" + handler + "].");
int got = handler.handleEvent(iterStamp); int got = handler.handleEvent(iterStamp);
if (got != 0) { if (got != 0) {
@@ -507,8 +506,7 @@ public class ConnectionManager extends LoopingThread
} }
} catch (Exception e) { } catch (Exception e) {
Log.warning("Error processing network data: " + handler + "."); log.log(Level.WARNING, "Error processing network data: " + handler + ".", e);
Log.logStackTrace(e);
// if you freak out here, you go straight in the can // if you freak out here, you go straight in the can
if (handler != null && handler instanceof Connection) { if (handler != null && handler instanceof Connection) {
@@ -537,7 +535,7 @@ public class ConnectionManager extends LoopingThread
// sanity check the message size // sanity check the message size
if (data.length > 1024 * 1024) { if (data.length > 1024 * 1024) {
Log.warning("Refusing to write absurdly large message [conn=" + conn + log.warning("Refusing to write absurdly large message [conn=" + conn +
", size=" + data.length + "]."); ", size=" + data.length + "].");
return true; return true;
} }
@@ -546,13 +544,13 @@ public class ConnectionManager extends LoopingThread
if (data.length > _outbuf.capacity()) { if (data.length > _outbuf.capacity()) {
// increase the buffer size in large increments // increase the buffer size in large increments
int ncapacity = Math.max(_outbuf.capacity() << 1, data.length); int ncapacity = Math.max(_outbuf.capacity() << 1, data.length);
Log.info("Expanding output buffer size [nsize=" + ncapacity + "]."); log.info("Expanding output buffer size [nsize=" + ncapacity + "].");
_outbuf = ByteBuffer.allocateDirect(ncapacity); _outbuf = ByteBuffer.allocateDirect(ncapacity);
} }
boolean fully = true; boolean fully = true;
try { try {
// Log.info("Writing " + data.length + " byte message to " + conn + "."); // log.info("Writing " + data.length + " byte message to " + conn + ".");
// first copy the data into our "direct" output buffer // first copy the data into our "direct" output buffer
_outbuf.put(data); _outbuf.put(data);
@@ -564,13 +562,13 @@ public class ConnectionManager extends LoopingThread
if (_outbuf.remaining() > 0) { if (_outbuf.remaining() > 0) {
fully = false; fully = false;
// Log.info("Partial write [conn=" + conn + // log.info("Partial write [conn=" + conn +
// ", msg=" + StringUtil.shortClassName(outmsg) + ", wrote=" + wrote + // ", msg=" + StringUtil.shortClassName(outmsg) + ", wrote=" + wrote +
// ", size=" + buffer.limit() + "]."); // ", size=" + buffer.limit() + "].");
pwh.handlePartialWrite(conn, _outbuf); pwh.handlePartialWrite(conn, _outbuf);
// } else if (wrote > 10000) { // } else if (wrote > 10000) {
// Log.info("Big write [conn=" + conn + // log.info("Big write [conn=" + conn +
// ", msg=" + StringUtil.shortClassName(outmsg) + // ", msg=" + StringUtil.shortClassName(outmsg) +
// ", wrote=" + wrote + "]."); // ", wrote=" + wrote + "].");
} }
@@ -599,19 +597,26 @@ public class ConnectionManager extends LoopingThread
protected void handleIterateFailure (Exception e) protected void handleIterateFailure (Exception e)
{ {
// log the exception // log the exception
Log.warning("ConnectionManager.iterate() uncaught exception."); log.log(Level.WARNING, "ConnectionManager.iterate() uncaught exception.", e);
Log.logStackTrace(e);
} }
// documentation inherited // documentation inherited
protected void didShutdown () protected void didShutdown ()
{ {
// unbind our listening socket
try {
_ssocket.socket().close();
} catch (IOException ioe) {
log.log(Level.WARNING, "Failed to close listening socket.", ioe);
}
// run our on-exit handler if we have one
Runnable onExit = _onExit; Runnable onExit = _onExit;
if (onExit != null) { if (onExit != null) {
Log.info("Connection Manager thread exited (running onExit)."); log.info("Connection Manager thread exited (running onExit).");
onExit.run(); onExit.run();
} else { } else {
Log.info("Connection Manager thread exited."); log.info("Connection Manager thread exited.");
} }
} }
@@ -628,23 +633,23 @@ public class ConnectionManager extends LoopingThread
if (channel == null) { if (channel == null) {
// in theory this shouldn't happen because we got an ACCEPT_READY event, but better // in theory this shouldn't happen because we got an ACCEPT_READY event, but better
// safe than sorry // safe than sorry
Log.info("Psych! Got ACCEPT_READY, but no connection."); log.info("Psych! Got ACCEPT_READY, but no connection.");
return; return;
} }
if (!(channel instanceof SelectableChannel)) { if (!(channel instanceof SelectableChannel)) {
try { try {
Log.warning("Provided with un-selectable socket as result of accept(), can't " + log.warning("Provided with un-selectable socket as result of accept(), can't " +
"cope [channel=" + channel + "]."); "cope [channel=" + channel + "].");
} catch (Error err) { } catch (Error err) {
Log.warning("Un-selectable channel also couldn't be printed."); log.warning("Un-selectable channel also couldn't be printed.");
} }
// stick a fork in the socket // stick a fork in the socket
channel.socket().close(); channel.socket().close();
return; return;
} }
// Log.debug("Accepted connection " + channel + "."); // log.fine("Accepted connection " + channel + ".");
// create a new authing connection object to manage the authentication of this client // create a new authing connection object to manage the authentication of this client
// connection and register it with our selection set // connection and register it with our selection set
@@ -658,8 +663,8 @@ public class ConnectionManager extends LoopingThread
return; return;
} catch (IOException ioe) { } catch (IOException ioe) {
// Log.warning("Failure accepting new connection: " + ioe); // no need to complain this happens in the normal course of events
// ioe.printStackTrace(System.err); // log.log(Level.WARNING, "Failure accepting new connection.", ioe);
} }
// make sure we don't leak a socket if something went awry // make sure we don't leak a socket if something went awry
@@ -667,7 +672,7 @@ public class ConnectionManager extends LoopingThread
try { try {
channel.socket().close(); channel.socket().close();
} catch (IOException ioe) { } catch (IOException ioe) {
Log.warning("Failed closing aborted connection: " + ioe); log.warning("Failed closing aborted connection: " + ioe);
} }
} }
} }
@@ -682,14 +687,14 @@ public class ConnectionManager extends LoopingThread
{ {
// sanity check // sanity check
if (conn == null || msg == null) { if (conn == null || msg == null) {
Log.warning("postMessage() bogosity [conn=" + conn + ", msg=" + msg + "]."); log.warning("postMessage() bogosity [conn=" + conn + ", msg=" + msg + "].");
Thread.dumpStack(); Thread.dumpStack();
return; return;
} }
// more sanity check; messages must only be posted from the dobjmgr thread // more sanity check; messages must only be posted from the dobjmgr thread
if (!PresentsServer.omgr.isDispatchThread()) { if (!PresentsServer.omgr.isDispatchThread()) {
Log.warning("Message posted on non-distributed object thread [conn=" + conn + log.warning("Message posted on non-distributed object thread [conn=" + conn +
", msg=" + msg + ", thread=" + Thread.currentThread() + "]."); ", msg=" + msg + ", thread=" + Thread.currentThread() + "].");
Thread.dumpStack(); Thread.dumpStack();
// let it through though as we don't want to break things unnecessarily // let it through though as we don't want to break things unnecessarily
@@ -708,14 +713,14 @@ public class ConnectionManager extends LoopingThread
byte[] data = new byte[buffer.limit()]; byte[] data = new byte[buffer.limit()];
buffer.get(data); buffer.get(data);
// Log.info("Flattened " + msg + " into " + data.length + " bytes."); // log.info("Flattened " + msg + " into " + data.length + " bytes.");
// and slap both on the queue // and slap both on the queue
_outq.append(new Tuple<Connection,byte[]>(conn, data)); _outq.append(new Tuple<Connection,byte[]>(conn, data));
} catch (Exception e) { } catch (Exception e) {
Log.warning("Failure flattening message [conn=" + conn + ", msg=" + msg + "]."); log.log(Level.WARNING, "Failure flattening message [conn=" + conn +
Log.logStackTrace(e); ", msg=" + msg + "].", e);
} }
} }
@@ -805,7 +810,7 @@ public class ConnectionManager extends LoopingThread
_partial = null; _partial = null;
_partials++; _partials++;
} else { } else {
// Log.info("Still going [conn=" + conn + ", wrote=" + wrote + // log.info("Still going [conn=" + conn + ", wrote=" + wrote +
// ", remain=" + _partial.remaining() + "]."); // ", remain=" + _partial.remaining() + "].");
return false; return false;
} }
@@ -851,6 +856,7 @@ public class ConnectionManager extends LoopingThread
protected int[] _ports; protected int[] _ports;
protected Authenticator _author; protected Authenticator _author;
protected Selector _selector; protected Selector _selector;
protected ServerSocketChannel _ssocket;
protected ResultListener<Object> _startlist; protected ResultListener<Object> _startlist;
/** Counts consecutive runtime errors in select(). */ /** Counts consecutive runtime errors in select(). */