ConnectionManager allowed a server to listen on multiple tcp and datagram ports, but it only closed

the last opened.  Close all opened listening sockets.

Also, the last opened datagram socket would be used to send all datagram messages instead of the one
datagrams were received on.  It now goes through the same channel where datagrams were last
received.



git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6156 542714f4-19e9-0310-aa3c-eee0fc999fb1

src/java/com/threerings/presents/server/net/Connection.java

@@ -21,17 +21,16 @@
 
 package com.threerings.presents.server.net;
 
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-
 import java.io.EOFException;
 import java.io.IOException;
-
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 
 import com.threerings.io.FramedInputStream;
 import com.threerings.io.FramingOutputStream;
@@ -146,6 +145,15 @@ public class Connection implements NetEventHandler
         return _datagramAddress;
     }
 
+    /**
+     * Returns the channel through which datagrams should be sent or null if no datagram channel
+     * has been established.
+     */
+    public DatagramChannel getDatagramChannel ()
+    {
+        return _datagramChannel;
+    }
+
     /**
      * Sets the secret string used to authenticate datagrams from the client.
      */
@@ -236,7 +244,8 @@ public class Connection implements NetEventHandler
     /**
      * Processes a datagram sent to this connection.
      */
-    public void handleDatagram (InetSocketAddress source, ByteBuffer buf, long when)
+    public void handleDatagram (InetSocketAddress source, DatagramChannel channel,
+        ByteBuffer buf, long when)
     {
         // lazily create our various bits and bobs
         if (_digest == null) {
@@ -263,6 +272,7 @@ public class Connection implements NetEventHandler
 
         // update our target address
         _datagramAddress = source;
+        _datagramChannel = channel;
 
         // read the contents through the sequencer
         try {
@@ -450,6 +460,7 @@ public class Connection implements NetEventHandler
     protected ObjectOutputStream _oout;
 
     protected InetSocketAddress _datagramAddress;
+    protected DatagramChannel _datagramChannel;
     protected byte[] _datagramSecret;
     protected boolean _transmitDatagrams;
 

src/java/com/threerings/presents/server/net/ConnectionManager.java

@@ -34,7 +34,6 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.DatagramChannel;
 import java.nio.channels.NotYetConnectedException;
-import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;
@@ -116,20 +115,15 @@ public class ConnectionManager extends LoopingThread
         String bindHostname, String datagramHostname, int[] ports, int[] datagramPorts)
             throws IOException
     {
-        Preconditions.checkArgument(ports != null, "Ports must be non-null.");
+        Preconditions.checkNotNull(ports, "Ports must be non-null.");
-        Preconditions.checkArgument(datagramPorts != null, "Datagram ports must be non-null. " +
+        Preconditions.checkNotNull(datagramPorts, "Datagram ports must be non-null. " +
                                     "Pass a zero-length array to bind no datagram ports.");
-        Preconditions.checkState(!super.isRunning(), "Must initialize before starting.");
 
         _bindHostname = bindHostname;
         _datagramHostname = datagramHostname;
         _ports = ports;
         _datagramPorts = datagramPorts;
         _selector = SelectorProvider.provider().openSelector();
-
-        // create our stats record
-        _stats = new ConMgrStats();
-        _lastStats = new ConMgrStats();
     }
 
     /**
@@ -360,18 +354,20 @@ public class ConnectionManager extends LoopingThread
     @Override
     protected void willStart ()
     {
+        Preconditions.checkNotNull(_ports, "Must call init before starting.");
         int successes = 0;
         for (int port : _ports) {
             try {
                 // create a listening socket and add it to the select set
-                _ssocket = ServerSocketChannel.open();
+                ServerSocketChannel ssocket = ServerSocketChannel.open();
-                _ssocket.configureBlocking(false);
+                ssocket.configureBlocking(false);
 
                 InetSocketAddress isa = getAddress(_bindHostname, port);
-                _ssocket.socket().bind(isa);
+                ssocket.socket().bind(isa);
-                registerChannel(_ssocket);
+                registerChannel(ssocket);
                 successes++;
                 log.info("Server listening on " + isa + ".");
+                _ssockets.add(ssocket);
 
             } catch (IOException ioe) {
                 log.warning("Failure listening to socket", "hostname", _bindHostname,
@@ -416,12 +412,13 @@ public class ConnectionManager extends LoopingThread
         for (int port : _datagramPorts) {
             try {
                 // create a channel and add it to the select set
-                _datagramChannel = DatagramChannel.open();
+                DatagramChannel channel = DatagramChannel.open();
-                _datagramChannel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY
+                channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY
-                _datagramChannel.configureBlocking(false);
+                channel.configureBlocking(false);
                 InetSocketAddress isa = getAddress(_datagramHostname, port);
-                _datagramChannel.socket().bind(isa);
+                channel.socket().bind(isa);
-                registerChannel(_datagramChannel);
+                registerChannel(channel);
+                _datagramChannels.add(channel);
                 log.info("Server accepting datagrams on " + isa + ".");
 
             } catch (IOException ioe) {
@@ -815,7 +812,7 @@ public class ConnectionManager extends LoopingThread
         _databuf.clear();
         _databuf.put(data).flip();
         try {
-            return _datagramChannel.send(_databuf, target) > 0;
+            return conn.getDatagramChannel().send(_databuf, target) > 0;
         } catch (IOException ioe) {
             log.warning("Failed to send datagram.", ioe);
             return false;
@@ -849,10 +846,9 @@ public class ConnectionManager extends LoopingThread
 
             // create a new authing connection object to manage the authentication of this client
             // connection and register it with our selection set
-            SelectableChannel selchan = channel;
+            channel.configureBlocking(false);
-            selchan.configureBlocking(false);
             AuthingConnection aconn = new AuthingConnection();
-            aconn.selkey = selchan.register(_selector, SelectionKey.OP_READ);
+            aconn.selkey = channel.register(_selector, SelectionKey.OP_READ);
             aconn.init(this, channel, System.currentTimeMillis());
             _handlers.put(aconn.selkey, aconn);
             synchronized (this) {
@@ -909,7 +905,7 @@ public class ConnectionManager extends LoopingThread
         int connectionId = _databuf.getInt();
         Connection conn = _connections.get(connectionId);
         if (conn != null) {
-            conn.handleDatagram(source, _databuf, when);
+            conn.handleDatagram(source, listener, _databuf, when);
         } else {
             log.debug("Received datagram for unknown connection", "id", connectionId,
                       "source", source);
@@ -1074,16 +1070,18 @@ public class ConnectionManager extends LoopingThread
         // 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.
-        // TODO: consider shutting down the listen socker earlier, like in the shutdown method
+        // TODO: consider closing the listen sockets earlier, like in the shutdown method
-        try {
+        for (ServerSocketChannel ssocket : _ssockets) {
-            _ssocket.socket().close();
+            try {
-        } catch (IOException ioe) {
+                ssocket.socket().close();
-            log.warning("Failed to close listening socket.", ioe);
+            } catch (IOException ioe) {
+                log.warning("Failed to close listening socket: " + ssocket, ioe);
+            }
         }
 
-        // and the datagram socket, if any
+        // and the datagram sockets, if any
-        if (_datagramChannel != null) {
+        for (DatagramChannel datagramChannel : _datagramChannels) {
-            _datagramChannel.socket().close();
+            datagramChannel.socket().close();
         }
 
         // report if there's anything left on the outgoing message queue
@@ -1224,8 +1222,8 @@ public class ConnectionManager extends LoopingThread
     protected int[] _ports, _datagramPorts;
     protected String _bindHostname, _datagramHostname;
     protected Selector _selector;
-    protected ServerSocketChannel _ssocket;
+    protected List<ServerSocketChannel> _ssockets = Lists.newArrayList();
-    protected DatagramChannel _datagramChannel;
+    protected List<DatagramChannel> _datagramChannels = Lists.newArrayList();
 
     /** Counts consecutive runtime errors in select(). */
     protected int _runtimeExceptionCount;
@@ -1251,10 +1249,10 @@ public class ConnectionManager extends LoopingThread
     protected Map<Connection, OverflowQueue> _oflowqs = Maps.newHashMap();
 
     /** Our current runtime stats. */
-    protected ConMgrStats _stats;
+    protected ConMgrStats _stats = new ConMgrStats();
 
     /** A snapshot of our runtime stats as of our last report. */
-    protected ConMgrStats _lastStats;
+    protected ConMgrStats _lastStats = new ConMgrStats();
 
     /** Used to periodically report connection manager activity when in debug mode. */
     protected long _lastDebugStamp;