From 308a6f9d522d345c311d11c0100ad27a1a9023eb Mon Sep 17 00:00:00 2001 From: Charlie Groves Date: Thu, 16 Sep 2010 22:47:58 +0000 Subject: [PATCH] Turns out presents' datagram support doesn't make sense at an individual datagram level, and requires a supporting Connection provided by the ConnectionManager. As such, there's no point in adding an additional selector for datagrams outside the ConnectionManager, and this moves it back into BindingConectionManager. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6166 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../com/threerings/nio/DatagramAcceptor.java | 92 ----------- .../com/threerings/nio/SelectAcceptor.java | 66 -------- .../threerings/nio/SocketChannelAcceptor.java | 85 +++++++---- .../server/net/BindingConnectionManager.java | 143 ++++++++++++++---- .../server/net/ConnectionManager.java | 74 ++------- 5 files changed, 185 insertions(+), 275 deletions(-) delete mode 100644 src/java/com/threerings/nio/DatagramAcceptor.java delete mode 100644 src/java/com/threerings/nio/SelectAcceptor.java diff --git a/src/java/com/threerings/nio/DatagramAcceptor.java b/src/java/com/threerings/nio/DatagramAcceptor.java deleted file mode 100644 index 1f5ac0494..000000000 --- a/src/java/com/threerings/nio/DatagramAcceptor.java +++ /dev/null @@ -1,92 +0,0 @@ -package com.threerings.nio; - -import java.util.List; -import java.io.IOException; -import java.net.InetSocketAddress; -import java.nio.channels.DatagramChannel; -import java.nio.channels.SelectionKey; -import com.google.common.collect.Lists; -import com.threerings.nio.SelectorIterable.SelectFailureHandler; - -import static com.threerings.presents.Log.log; - -/** - * Listens for datagrams arriving on a set of ports for a hostname and passes those datagrams on - * to a listener when they're ready to read. {@link #listen()} must be called after creating the - * acceptor to open the channels, and then tick must be called periodically to process new - * datagrams. - */ -public class DatagramAcceptor extends SelectAcceptor -{ - /** - * Callback when datagrams are ready to be read. - */ - public interface DatagramHandler - { - /** - * Reads the message currentnly in the channel that was selected at the given time. - */ - void handleDatagram (DatagramChannel channel, long when); - } - - /** - * Creates an acceptor that passes datagrams on to the given handler. - * @param failureHandler - called when the selector is irredemably broken. - * @param hostname - the hostname to bind to or null for all interfaces - * @param ports - the ports to bind to, or an empty array to skip binding - * @param selectLoopTime - the amount of time to wait in select, or 0 to skip the wait at all. - */ - public DatagramAcceptor (DatagramHandler dgramHandler, SelectFailureHandler failureHandler, - String hostname, int[] ports, int selectLoopTime) - throws IOException - { - super(failureHandler, hostname, ports, selectLoopTime); - _dgramHandler = dgramHandler; - } - - /** - * Opens datagram channels on the configured addresses. - */ - public boolean listen () - { - // open up the datagram ports as well - for (int port : _ports) { - try { - // create a channel and add it to the select set - final DatagramChannel channel = DatagramChannel.open(); - channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY - channel.configureBlocking(false); - InetSocketAddress isa = getAddress(port); - channel.socket().bind(isa); - SelectionKey sk = channel.register(_selector, SelectionKey.OP_READ); - _handlers.put(sk, new SelectionHandler() { - public void handle (long when) { - _dgramHandler.handleDatagram(channel, when); - } - }); - _datagramChannels.add(channel); - log.info("Server accepting datagrams on " + isa + "."); - - } catch (IOException ioe) { - log.warning("Failure opening datagram channel", "hostname", _bindHostname, - "port", port, ioe); - } - } - return true; - } - - /** - * Closes any open channels. - */ - public void shutdown() - { - // unbind sockets, if any - for (DatagramChannel datagramChannel : _datagramChannels) { - datagramChannel.socket().close(); - } - - } - - protected final List _datagramChannels = Lists.newArrayList(); - protected final DatagramHandler _dgramHandler; -} diff --git a/src/java/com/threerings/nio/SelectAcceptor.java b/src/java/com/threerings/nio/SelectAcceptor.java deleted file mode 100644 index 1f381d8ae..000000000 --- a/src/java/com/threerings/nio/SelectAcceptor.java +++ /dev/null @@ -1,66 +0,0 @@ -package com.threerings.nio; - -import java.util.Map; - -import java.io.IOException; -import java.net.InetSocketAddress; -import java.nio.channels.SelectionKey; -import java.nio.channels.Selector; - -import com.google.common.base.Preconditions; -import com.google.common.collect.Maps; -import com.google.common.primitives.Ints; - -import com.samskivert.util.StringUtil; - -import com.threerings.nio.SelectorIterable.SelectFailureHandler; - -/** - * Base class for binding to ports with a selector and processing events on that selector. - */ -public abstract class SelectAcceptor -{ - public SelectAcceptor (SelectFailureHandler failureHandler, String bindHostname, int[] ports, - int selectLoopTime) - throws IOException - { - Preconditions.checkNotNull(ports, "Ports must be non-null."); - - _bindHostname = bindHostname; - _ports = ports; - - _selectorSelector = new SelectorIterable(_selector, selectLoopTime, failureHandler); - } - - /** - * Checks the selector for ready keys and passes any through to the handlers. - */ - public void tick (long when) - { - for (SelectionKey key : _selectorSelector) { - _handlers.get(key).handle(when); - } - } - - public Iterable getPorts() - { - return Ints.asList(_ports); - } - - /** Helper function for creating proper bindable socket addresses. */ - protected InetSocketAddress getAddress (int port) - { - return StringUtil.isBlank(_bindHostname) ? - new InetSocketAddress(port) : new InetSocketAddress(_bindHostname, port); - } - - protected interface SelectionHandler { - void handle(long when); - } - - protected final int[] _ports; - protected final String _bindHostname; - protected final Map _handlers = Maps.newHashMap(); - protected final Selector _selector = Selector.open(); - protected final SelectorIterable _selectorSelector; -} diff --git a/src/java/com/threerings/nio/SocketChannelAcceptor.java b/src/java/com/threerings/nio/SocketChannelAcceptor.java index 10fe62acf..96825c167 100644 --- a/src/java/com/threerings/nio/SocketChannelAcceptor.java +++ b/src/java/com/threerings/nio/SocketChannelAcceptor.java @@ -1,14 +1,22 @@ package com.threerings.nio; import java.util.List; +import java.util.Map; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; +import com.google.common.base.Preconditions; import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.primitives.Ints; + +import com.samskivert.util.StringUtil; + import com.threerings.nio.SelectorIterable.SelectFailureHandler; import static com.threerings.presents.Log.log; @@ -20,8 +28,18 @@ import static com.threerings.presents.Log.log; * creating the acceptor to open the channels, and then tick must be called periodically to process * new connections. */ -public class SocketChannelAcceptor extends SelectAcceptor +public class SocketChannelAcceptor { + /** + * Creates a address to the given host, or the wildcard host if the hostname is + * {@link StringUtil#blank}. + */ + public static InetSocketAddress getAddress (String hostname, int port) + { + return StringUtil.isBlank(hostname) ? + new InetSocketAddress(port) : new InetSocketAddress(hostname, port); + } + public interface SocketChannelHandler { void handleSocketChannel (SocketChannel channel, long when); @@ -39,53 +57,62 @@ public class SocketChannelAcceptor extends SelectAcceptor SelectFailureHandler failureHandler, String bindHostname, int[] ports, int selectLoopTime) throws IOException { - super(failureHandler, bindHostname, ports, selectLoopTime); + Preconditions.checkNotNull(ports, "Ports must be non-null."); + + _bindHostname = bindHostname; + _ports = ports; + + _selectorSelector = new SelectorIterable(_selector, selectLoopTime, failureHandler); _connHandler = connectionHandler; } - protected void acceptConnection (ServerSocketChannel listener, long when) + /** + * Checks the selector for ready keys and passes any through to the handlers. + */ + public void tick (long when) { - SocketChannel channel = null; - try { - channel = listener.accept(); - } catch (IOException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } - if (channel == null) { - // in theory this shouldn't happen because we got an ACCEPT_READY event... - log.info("Psych! Got ACCEPT_READY, but no connection."); - return; - } + for (SelectionKey key : _selectorSelector) { + ServerSocketChannel ssocket = _channels.get(key); + SocketChannel channel = null; + try { + channel = ssocket.accept(); + } catch (IOException e) { + log.warning("Got exception on accept", e); + continue; + } + if (channel == null) { + // in theory this shouldn't happen because we got an ACCEPT_READY event... + log.info("Psych! Got ACCEPT_READY, but no connection."); + continue; + } -// log.debug("Accepted connection " + channel + "."); +// log.debug("Accepted connection " + channel + "."); - _connHandler.handleSocketChannel(channel, when); + _connHandler.handleSocketChannel(channel, when); + } } + public Iterable getPorts () + { + return Ints.asList(_ports); + } /** * Listens for socket connections on the configured addresses. */ public boolean listen () { - int successes = 0; for (int port : _ports) { try { // create a listening socket and add it to the select set final ServerSocketChannel ssocket = ServerSocketChannel.open(); ssocket.configureBlocking(false); - InetSocketAddress isa = getAddress(port); + InetSocketAddress isa = getAddress(_bindHostname, port); ssocket.socket().bind(isa); SelectionKey sk = ssocket.register(_selector, SelectionKey.OP_ACCEPT); - _handlers.put(sk, new SelectionHandler() { - public void handle (long when) { - acceptConnection(ssocket, when); - } - }); - successes++; + _channels.put(sk, ssocket); log.info("Server listening on " + isa + "."); _ssockets.add(ssocket); @@ -122,7 +149,7 @@ public class SocketChannelAcceptor extends SelectAcceptor // } // if we failed to listen on at least one port, give up the ghost - return successes > 0; + return !_channels.isEmpty(); } public void shutdown() @@ -139,6 +166,12 @@ public class SocketChannelAcceptor extends SelectAcceptor } } + + protected final int[] _ports; + protected final String _bindHostname; + protected final Map _channels = Maps.newHashMap(); + protected final Selector _selector = Selector.open(); + protected final SelectorIterable _selectorSelector; protected final List _ssockets = Lists.newArrayList(); protected final SocketChannelHandler _connHandler; } diff --git a/src/java/com/threerings/presents/server/net/BindingConnectionManager.java b/src/java/com/threerings/presents/server/net/BindingConnectionManager.java index 8575800eb..86ca25099 100644 --- a/src/java/com/threerings/presents/server/net/BindingConnectionManager.java +++ b/src/java/com/threerings/presents/server/net/BindingConnectionManager.java @@ -1,14 +1,21 @@ package com.threerings.presents.server.net; +import java.util.List; + import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.channels.DatagramChannel; +import java.nio.channels.SelectionKey; import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; import com.google.inject.Inject; import com.google.inject.Singleton; + import com.samskivert.util.Lifecycle; import com.threerings.presents.server.ReportManager; -import com.threerings.nio.DatagramAcceptor; + import com.threerings.nio.SocketChannelAcceptor; import static com.threerings.presents.Log.log; @@ -26,25 +33,6 @@ public class BindingConnectionManager extends ConnectionManager super(cycle, repmgr, incomingEventWait); } - @Override - protected void willStart () - { - if (!_socketAcceptor.listen()) { - log.warning("ConnectionManager failed to bind to any ports. Shutting down."); - _server.queueShutdown(); - } else { - _dgramAcceptor.listen(); - } - } - - @Override - protected void processIncomingEvents (long iterStamp) - { - _socketAcceptor.tick(iterStamp); - _dgramAcceptor.tick(iterStamp); - super.processIncomingEvents(iterStamp); - } - /** * Configures the connection manager with the hostname and ports on which it will listen for * socket connections and datagram packets. This must be called before the connection manager @@ -61,17 +49,43 @@ public class BindingConnectionManager extends ConnectionManager int[] datagramPorts) throws IOException { - Preconditions.checkNotNull(socketPorts, "Ports must be non-null."); + Preconditions.checkNotNull(socketPorts, "Socket ports must be non-null."); Preconditions.checkNotNull(datagramPorts, "Datagram ports must be non-null. " + "Pass a zero-length array to bind no datagram ports."); - // Listen for socket connections and datagram connections, but don't wait for anything to - // show up since that check is occurring as part of ConnectionManager's incoming event loop - // that already has a wait. + // Listen for socket connections, but don't wait to select on them. The connection check + // occurs as part of ConnectionManager's incoming event loop, which already has a wait. _socketAcceptor = new SocketChannelAcceptor(this, _failureHandler, socketHostname, socketPorts, 0); - _dgramAcceptor = new DatagramAcceptor(this, _failureHandler, datagramHostname, - datagramPorts, 0); + + _datagramPorts = datagramPorts; + _datagramHostname = datagramHostname; + } + + @Override + protected void willStart () + { + 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) { + try { + acceptDatagrams(port); + } catch (IOException ioe) { + log.warning("Failure opening datagram channel", "hostname", _datagramHostname, + "port", port, ioe); + } + } + } + } + + @Override + protected void processIncomingEvents (long iterStamp) + { + _socketAcceptor.tick(iterStamp); + super.processIncomingEvents(iterStamp); } @Override @@ -81,10 +95,81 @@ public class BindingConnectionManager extends ConnectionManager // TODO: consider closing the listen sockets earlier, like in the shutdown method _socketAcceptor.shutdown(); - _dgramAcceptor.shutdown(); + // unbind datagram channels, if any + for (DatagramChannel datagramChannel : _datagramChannels) { + datagramChannel.socket().close(); + } } + protected void acceptDatagrams (int port) + throws IOException + { + // create a channel and add it to the select set + final DatagramChannel channel = DatagramChannel.open(); + channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY + channel.configureBlocking(false); + InetSocketAddress isa = SocketChannelAcceptor.getAddress(_datagramHostname, port); + channel.socket().bind(isa); + SelectionKey sk = channel.register(_selector, SelectionKey.OP_READ); + _handlers.put(sk, new NetEventHandler() { + public int handleEvent (long when) { + return handleDatagram(channel, when); + } + + public boolean checkIdle (long now) { + return false;// Can't be idle + } + + public void becameIdle () {} + }); + _datagramChannels.add(channel); + log.info("Server accepting datagrams on " + isa + "."); + } + + /** + * Called when a datagram message is ready to be read off its channel. + */ + protected int handleDatagram (DatagramChannel listener, long when) + { + InetSocketAddress source; + _databuf.clear(); + try { + source = (InetSocketAddress)listener.receive(_databuf); + } catch (IOException ioe) { + log.warning("Failure receiving datagram.", ioe); + return 0; + } + + // make sure we actually read a packet + if (source == null) { + log.info("Psych! Got READ_READY, but no datagram."); + return 0; + } + + // flip the buffer and record the size (which must be at least 14 to contain the connection + // id, authentication hash, and a class reference) + int size = _databuf.flip().remaining(); + if (size < 14) { + log.warning("Received undersized datagram", "source", source, "size", size); + return 0; + } + + // the first four bytes are the connection id + int connectionId = _databuf.getInt(); + Connection conn = _connections.get(connectionId); + if (conn != null) { + conn.handleDatagram(source, listener, _databuf, when); + } else { + log.debug("Received datagram for unknown connection", "id", connectionId, + "source", source); + } + + return size; + } protected SocketChannelAcceptor _socketAcceptor; - protected DatagramAcceptor _dgramAcceptor; -} + protected int[] _datagramPorts; + protected String _datagramHostname; + + protected final List _datagramChannels = Lists.newArrayList(); +} \ No newline at end of file diff --git a/src/java/com/threerings/presents/server/net/ConnectionManager.java b/src/java/com/threerings/presents/server/net/ConnectionManager.java index 9b0317a4b..f0e3b2f4e 100644 --- a/src/java/com/threerings/presents/server/net/ConnectionManager.java +++ b/src/java/com/threerings/presents/server/net/ConnectionManager.java @@ -29,7 +29,6 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; -import java.nio.channels.DatagramChannel; import java.nio.channels.NotYetConnectedException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; @@ -69,32 +68,27 @@ import com.threerings.presents.server.PresentsServer; import com.threerings.presents.server.ReportManager; import com.threerings.presents.util.DatagramSequencer; -import com.threerings.nio.DatagramAcceptor; import com.threerings.nio.SocketChannelAcceptor; import com.threerings.nio.SocketChannelAcceptor.SocketChannelHandler; -import com.threerings.nio.DatagramAcceptor.DatagramHandler; import com.threerings.nio.SelectorIterable; import com.threerings.nio.SelectorIterable.SelectFailureHandler; import static com.threerings.presents.Log.log; /** - * Manages socket connections and datagram messages. It creates connection objects for each socket - * connection, but those connection objects interact closely with the connection manager because - * network I/O is done via a poll()-like mechanism rather than via threads.

+ * Manages socket connections. It creates connection objects for each socket connection, but those + * connection objects interact closely with the connection manager because network I/O is done via + * a poll()-like mechanism rather than via threads.

* - * ConnectionManager doesn't handle accepting tcp connections or listening for datagrams; it - * expects an external entity to do so and call its handleSocketChannel and - * handleDatagram methods. + * ConnectionManager doesn't handle accepting tcp connections; it expects an external entity to do + * so and call its handleSocketChannel method. * * @see BindingConnectionManager * @see SocketChannelAcceptor - * @see DatagramAcceptor */ @Singleton public class ConnectionManager extends LoopingThread - implements Lifecycle.ShutdownComponent, ReportManager.Reporter, DatagramHandler, - SocketChannelHandler + implements Lifecycle.ShutdownComponent, ReportManager.Reporter, SocketChannelHandler { /** * Creates a connection manager instance. @@ -266,47 +260,6 @@ public class ConnectionManager extends LoopingThread } } - /** - * Called by our DatagramAcceptor when a datagram message is ready to be read off its channel. - */ - public void handleDatagram (DatagramChannel listener, long when) - { - InetSocketAddress source; - _databuf.clear(); - try { - source = (InetSocketAddress)listener.receive(_databuf); - } catch (IOException ioe) { - log.warning("Failure receiving datagram.", ioe); - return; - } - - // make sure we actually read a packet - if (source == null) { - log.info("Psych! Got READ_READY, but no datagram."); - return; - } - - // flip the buffer and record the size (which must be at least 14 to contain the connection - // id, authentication hash, and a class reference) - int size = _databuf.flip().remaining(); - if (size < 14) { - log.warning("Received undersized datagram", "source", source, "size", size); - return; - } - - // the first four bytes are the connection id - int connectionId = _databuf.getInt(); - Connection conn = _connections.get(connectionId); - if (conn != null) { - conn.handleDatagram(source, listener, _databuf, when); - } else { - log.debug("Received datagram for unknown connection", "id", connectionId, - "source", source); - } - - noteRead(size, 1, 1); - } - /** * Performs the authentication process on the specified connection. This is called by {@link * AuthingConnection} itself once it receives its auth request. @@ -538,7 +491,12 @@ public class ConnectionManager extends LoopingThread } } - noteRead(bytesIn, msgsIn, eventCount); + synchronized (this) { + // update our stats + _stats.eventCount += eventCount; + _stats.bytesIn += bytesIn; + _stats.msgsIn += msgsIn; + } } /** @@ -701,14 +659,6 @@ public class ConnectionManager extends LoopingThread _stats.bytesOut += bytes; } - protected synchronized void noteRead (long bytesIn, long msgsIn, long eventCount) - { - // update our stats - _stats.eventCount += eventCount; - _stats.bytesIn += bytesIn; - _stats.msgsIn += msgsIn; - } - /** * Called by a connection when it has a downstream message that needs to be delivered. * Note: this method is called as a result of a call to {@link Connection#postMessage}