Some tidying up and comment improvements.
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6169 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
@@ -40,11 +40,19 @@ import static com.threerings.presents.Log.log;
|
||||
public class SelectorIterable
|
||||
implements Iterable<SelectionKey>
|
||||
{
|
||||
/** The maximum allowed consecutive select() failures, after which we will declare ourselves
|
||||
* irreparably hosed. */
|
||||
public static final int MAX_SELECT_FAILURES = 20;
|
||||
|
||||
/**
|
||||
* Callback for the Selector failing.
|
||||
* An interface for hearing about catastrophic selection failure. If select fails (in a
|
||||
* non-expected way) more than {@link #MAX_SELECT_FAILURES} times in a row, this interface will
|
||||
* be notified. This allows a server to reboot itself in an orderly manner, rather than
|
||||
* continue in an inoperable state.
|
||||
*/
|
||||
public interface SelectFailureHandler {
|
||||
void handleSelectFailure(Exception e);
|
||||
/** Called to report a select() failure. */
|
||||
void handleSelectFailure (Exception e);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -59,6 +67,7 @@ public class SelectorIterable
|
||||
_failureHandler = handler;
|
||||
}
|
||||
|
||||
// from interface Iterable<SelectionKey>
|
||||
public Iterator<SelectionKey> iterator ()
|
||||
{
|
||||
return Iterators.consumingIterator(select().iterator());
|
||||
@@ -89,7 +98,7 @@ public class SelectorIterable
|
||||
// instead of looping indefinitely after things go pear-shaped, shut us down in an
|
||||
// orderly fashion
|
||||
log.warning("Failure select()ing.", re);
|
||||
if (_runtimeExceptionCount++ >= 20) {
|
||||
if (_runtimeExceptionCount++ >= MAX_SELECT_FAILURES) {
|
||||
_failureHandler.handleSelectFailure(re);
|
||||
}
|
||||
}
|
||||
@@ -97,10 +106,7 @@ public class SelectorIterable
|
||||
}
|
||||
|
||||
protected int _runtimeExceptionCount;
|
||||
|
||||
protected final int _selectLoopTime;
|
||||
|
||||
protected final Selector _selector;
|
||||
|
||||
protected final SelectFailureHandler _failureHandler;
|
||||
}
|
||||
|
||||
@@ -42,7 +42,6 @@ import com.threerings.nio.SelectorIterable.SelectFailureHandler;
|
||||
|
||||
import static com.threerings.presents.Log.log;
|
||||
|
||||
|
||||
/**
|
||||
* Listens for socket connections on a set of ports for a hostname and passes those connected
|
||||
* sockets on to a listener when they're ready. {@link #listen()} must be called after
|
||||
@@ -51,6 +50,13 @@ import static com.threerings.presents.Log.log;
|
||||
*/
|
||||
public class SocketChannelAcceptor
|
||||
{
|
||||
/** Used to pass freshly accepted sockets to entities that want them. */
|
||||
public interface SocketChannelHandler
|
||||
{
|
||||
/** Called when a new connection has been accepted and is ready for use. */
|
||||
void handleSocketChannel (SocketChannel channel, long when);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a address to the given host, or the wildcard host if the hostname is
|
||||
* {@link StringUtil#blank}.
|
||||
@@ -61,11 +67,6 @@ public class SocketChannelAcceptor
|
||||
new InetSocketAddress(port) : new InetSocketAddress(hostname, port);
|
||||
}
|
||||
|
||||
public interface SocketChannelHandler
|
||||
{
|
||||
void handleSocketChannel (SocketChannel channel, long when);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an acceptor that passes socket connections on to the given handler.
|
||||
*
|
||||
@@ -79,12 +80,9 @@ public class SocketChannelAcceptor
|
||||
throws IOException
|
||||
{
|
||||
Preconditions.checkNotNull(ports, "Ports must be non-null.");
|
||||
|
||||
_bindHostname = bindHostname;
|
||||
_ports = ports;
|
||||
|
||||
_selectorSelector = new SelectorIterable(_selector, selectLoopTime, failureHandler);
|
||||
|
||||
_connHandler = connectionHandler;
|
||||
}
|
||||
|
||||
@@ -107,13 +105,14 @@ public class SocketChannelAcceptor
|
||||
log.info("Psych! Got ACCEPT_READY, but no connection.");
|
||||
continue;
|
||||
}
|
||||
|
||||
// log.debug("Accepted connection " + channel + ".");
|
||||
|
||||
_connHandler.handleSocketChannel(channel, when);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the ports on which we are listening for connections.
|
||||
*/
|
||||
public Iterable<Integer> getPorts ()
|
||||
{
|
||||
return Ints.asList(_ports);
|
||||
@@ -173,11 +172,13 @@ public class SocketChannelAcceptor
|
||||
return !_channels.isEmpty();
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes all listening sockets and shuts down.
|
||||
*/
|
||||
public void shutdown()
|
||||
{
|
||||
// unbind our listening socket
|
||||
// Note: because we wait for the object manager to exit before we do, we will still be
|
||||
// accepting connections as long as there are events pending.
|
||||
// unbind our listening sockets; note: because we wait for the objmgr to exit before we do,
|
||||
// we will still be accepting connections as long as there are events pending
|
||||
for (ServerSocketChannel ssocket : _ssockets) {
|
||||
try {
|
||||
ssocket.socket().close();
|
||||
@@ -187,7 +188,6 @@ public class SocketChannelAcceptor
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
protected final int[] _ports;
|
||||
protected final String _bindHostname;
|
||||
protected final Map<SelectionKey, ServerSocketChannel> _channels = Maps.newHashMap();
|
||||
|
||||
@@ -157,10 +157,11 @@ public class PresentsServer
|
||||
// provide our client manager with the injector it needs
|
||||
_clmgr.setInjector(injector);
|
||||
|
||||
// configure our connection manager
|
||||
// if we have a connection manager that handles its own socket accepting, initialize it
|
||||
// with the hostname and ports on which to listen
|
||||
if (_conmgr instanceof BindingConnectionManager) {
|
||||
((BindingConnectionManager)_conmgr).init(getBindHostname(), getDatagramHostname(),
|
||||
getListenPorts(), getDatagramPorts());
|
||||
((BindingConnectionManager)_conmgr).init(
|
||||
getBindHostname(), getDatagramHostname(), getListenPorts(), getDatagramPorts());
|
||||
}
|
||||
|
||||
// initialize the time base services
|
||||
|
||||
@@ -48,7 +48,7 @@ import static com.threerings.presents.Log.log;
|
||||
public class BindingConnectionManager extends ConnectionManager
|
||||
{
|
||||
@Inject public BindingConnectionManager (Lifecycle cycle, ReportManager repmgr,
|
||||
IncomingEventWaitHolder incomingEventWait)
|
||||
IncomingEventWaitHolder incomingEventWait)
|
||||
throws IOException
|
||||
{
|
||||
super(cycle, repmgr, incomingEventWait);
|
||||
@@ -89,6 +89,7 @@ public class BindingConnectionManager extends ConnectionManager
|
||||
if (!_socketAcceptor.listen()) {
|
||||
log.warning("ConnectionManager failed to bind to any ports. Shutting down.");
|
||||
_server.queueShutdown();
|
||||
|
||||
} else {
|
||||
// open up the datagram ports as well
|
||||
for (int port : _datagramPorts) {
|
||||
@@ -105,7 +106,10 @@ public class BindingConnectionManager extends ConnectionManager
|
||||
@Override
|
||||
protected void processIncomingEvents (long iterStamp)
|
||||
{
|
||||
// first check for any new sockets, ready to be accepted
|
||||
_socketAcceptor.tick(iterStamp);
|
||||
|
||||
// then do our standard processing for existing connected sockets
|
||||
super.processIncomingEvents(iterStamp);
|
||||
}
|
||||
|
||||
@@ -193,4 +197,4 @@ public class BindingConnectionManager extends ConnectionManager
|
||||
protected String _datagramHostname;
|
||||
|
||||
protected final List<DatagramChannel> _datagramChannels = Lists.newArrayList();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -69,7 +69,6 @@ import com.threerings.presents.server.ReportManager;
|
||||
import com.threerings.presents.util.DatagramSequencer;
|
||||
|
||||
import com.threerings.nio.SocketChannelAcceptor;
|
||||
import com.threerings.nio.SocketChannelAcceptor.SocketChannelHandler;
|
||||
import com.threerings.nio.SelectorIterable;
|
||||
import com.threerings.nio.SelectorIterable.SelectFailureHandler;
|
||||
|
||||
@@ -80,15 +79,16 @@ import static com.threerings.presents.Log.log;
|
||||
* connection objects interact closely with the connection manager because network I/O is done via
|
||||
* a poll()-like mechanism rather than via threads.<p>
|
||||
*
|
||||
* ConnectionManager doesn't handle accepting tcp connections; it expects an external entity to do
|
||||
* so and call its <code>handleSocketChannel</code> method.
|
||||
* ConnectionManager doesn't directly accept TCP connections; it expects an external entity to do
|
||||
* so and call its {@link #handleSocketChannel} method.
|
||||
*
|
||||
* @see BindingConnectionManager
|
||||
* @see SocketChannelAcceptor
|
||||
*/
|
||||
@Singleton
|
||||
public class ConnectionManager extends LoopingThread
|
||||
implements Lifecycle.ShutdownComponent, ReportManager.Reporter, SocketChannelHandler
|
||||
implements Lifecycle.ShutdownComponent, ReportManager.Reporter,
|
||||
SocketChannelAcceptor.SocketChannelHandler
|
||||
{
|
||||
/**
|
||||
* Creates a connection manager instance.
|
||||
@@ -100,8 +100,8 @@ public class ConnectionManager extends LoopingThread
|
||||
super("ConnectionManager");
|
||||
cycle.addComponent(this);
|
||||
repmgr.registerReporter(this);
|
||||
_selectorSelector = new SelectorIterable(_selector, incomingEventWait.value,
|
||||
_failureHandler);
|
||||
_selectorSelector = new SelectorIterable(
|
||||
_selector, incomingEventWait.value, _failureHandler);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -224,16 +224,7 @@ public class ConnectionManager extends LoopingThread
|
||||
report.append(bytesOut*1000/sinceLast).append(" bps\n");
|
||||
}
|
||||
|
||||
@Override // from LoopingThread
|
||||
public boolean isRunning ()
|
||||
{
|
||||
// Prevent exiting our thread until the object manager is done.
|
||||
return super.isRunning() || _omgr.isRunning();
|
||||
}
|
||||
|
||||
/**
|
||||
* Called by our SocketChannelAcceptor when a new connection has been accepted on its socket.
|
||||
*/
|
||||
// from interface SocketChannelAcceptor.SocketChannelHandler
|
||||
public void handleSocketChannel (SocketChannel channel, long when)
|
||||
{
|
||||
try {
|
||||
@@ -260,6 +251,13 @@ public class ConnectionManager extends LoopingThread
|
||||
}
|
||||
}
|
||||
|
||||
@Override // from LoopingThread
|
||||
public boolean isRunning ()
|
||||
{
|
||||
// Prevent exiting our thread until the object manager is done.
|
||||
return super.isRunning() || _omgr.isRunning();
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs the authentication process on the specified connection. This is called by {@link
|
||||
* AuthingConnection} itself once it receives its auth request.
|
||||
|
||||
Reference in New Issue
Block a user