Reinstated the Presents I/O refactor with the modification of ensuring

that authentication is processed on the dobjmgr thread rather than
requiring the caller to do the right thing (or not as the case happened to
be).


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

src/java/com/threerings/io/FramingOutputStream.java

@@ -44,7 +44,7 @@ public class FramingOutputStream extends OutputStream
 {
     public FramingOutputStream ()
     {
-        _buffer = ByteBuffer.allocateDirect(INITIAL_BUFFER_SIZE);
+        _buffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
         _buffer.put(HEADER_PAD);
     }
 
@@ -99,7 +99,7 @@ public class FramingOutputStream extends OutputStream
 	if (ncapacity > ocapacity) {
             // increase the buffer size in large increments
             ncapacity = Math.max(ocapacity << 1, ncapacity);
-            ByteBuffer newbuf = ByteBuffer.allocateDirect(ncapacity);
+            ByteBuffer newbuf = ByteBuffer.allocate(ncapacity);
             newbuf.put((ByteBuffer)_buffer.flip());
 	    _buffer = newbuf;
 	}

src/java/com/threerings/presents/client/Communicator.java

@@ -316,7 +316,8 @@ public class Communicator
         try {
             ByteBuffer buffer = _fout.frameAndReturnBuffer();
             if (buffer.limit() > 4096) {
-                String txt = StringUtil.truncate(String.valueOf(msg), 80, "...");
+                String txt = StringUtil.truncate(
+                    String.valueOf(msg), 80, "...");
                 Log.info("Whoa, writin' a big one [msg=" + txt +
                          ", size=" + buffer.limit() + "].");
             }

src/java/com/threerings/presents/server/Authenticator.java

@@ -64,20 +64,26 @@ public abstract class Authenticator
      * authentication if it succeeded.
      */
     protected void connectionWasAuthenticated (
-        AuthingConnection conn, AuthResponse rsp)
+        final AuthingConnection conn, final AuthResponse rsp)
     {
-        // now ship the response back
+        // make double plus extra sure we're on the omgr thread
-        conn.postMessage(rsp);
+        PresentsServer.omgr.postRunnable(new Runnable() {
+            public void run () {
+                // stuff a reference to the auth response into the
+                // connection so that we have access to it later in the
+                // authentication process
+                conn.setAuthResponse(rsp);
 
-        // stuff a reference to the auth response into the connection so
+                // send the response back to the client
-        // that we have access to it later in the authentication process
+                conn.postMessage(rsp);
-        conn.setAuthResponse(rsp);
 
-        // if the authentication request was granted, let the connection
+                // if the authentication request was granted, let the
-        // manager know that we just authed
+                // connection manager know that we just authed
-        if (AuthResponseData.SUCCESS.equals(rsp.getData().code)) {
+                if (AuthResponseData.SUCCESS.equals(rsp.getData().code)) {
-            _conmgr.connectionDidAuthenticate(conn);
+                    _conmgr.connectionDidAuthenticate(conn);
-        }
+                }
+            }
+        });
     }
 
     /** The connection manager with which we're working. */

src/java/com/threerings/presents/server/PresentsClient.java

@@ -758,6 +758,17 @@ public class PresentsClient
         }
     }
 
+    /** Callable from non-dobjmgr thread, this queues up a runnable on the
+     * dobjmgr thread to post the supplied message to this client. */
+    protected final void safePostMessage (final DownstreamMessage msg)
+    {
+        PresentsServer.omgr.postRunnable(new Runnable() {
+            public void run () {
+                postMessage(msg);
+            }
+        });
+    }
+
     /** Queues a message for delivery to the client. */
     protected final boolean postMessage (DownstreamMessage msg)
     {
@@ -856,7 +867,7 @@ public class PresentsClient
 
             // post a response to the client letting them know that we
             // will no longer send them events regarding this object
-            client.postMessage(new UnsubscribeResponse(oid));
+            client.safePostMessage(new UnsubscribeResponse(oid));
         }
     }
 
@@ -890,7 +901,7 @@ public class PresentsClient
         {
             // send a pong response
             PingRequest req = (PingRequest)msg;
-            client.postMessage(new PongResponse(req.getUnpackStamp()));
+            client.safePostMessage(new PongResponse(req.getUnpackStamp()));
         }
     }
 

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

@@ -333,8 +333,7 @@ public class ConnectionManager extends LoopingThread
                     if (oq.writeOverflowMessages(iterStamp)) {
                         // if they were all written, we can remove it
                         oqiter.remove();
-//                         Log.info("Flushed overflow queue for " +
+                        Log.info("Flushed overflow queue " + oq + ".");
-//                                  oq.conn + ".");
                     }
 
                 } catch (IOException ioe) {
@@ -364,7 +363,7 @@ public class ConnectionManager extends LoopingThread
             }
 
             // otherwise write the message out to the client directly
-            writeMessage(conn, (DownstreamMessage)tup.right, _oflowHandler);
+            writeMessage(conn, (byte[])tup.right, _oflowHandler);
         }
 
         // check for connections that have completed authentication
@@ -472,7 +471,7 @@ public class ConnectionManager extends LoopingThread
      * have been invoked).
      */
     protected boolean writeMessage (
-        Connection conn, DownstreamMessage outmsg, PartialWriteHandler pwh)
+        Connection conn, byte[] data, PartialWriteHandler pwh)
     {
         // if the connection to which this message is destined is closed,
         // drop the message and move along quietly; this is perfectly
@@ -482,44 +481,55 @@ public class ConnectionManager extends LoopingThread
         if (conn.isClosed()) {
             return true;
         }
-            
+
+        // sanity check the message size
+        if (data.length > 1024 * 1024) {
+            Log.warning("Refusing to write absurdly large message " +
+                        "[conn=" + conn + ", size=" + data.length + "].");
+            return true;
+        }
+
+        // expand our output buffer if needed to accomodate this message
+        if (data.length > _outbuf.capacity()) {
+            // increase the buffer size in large increments
+            int ncapacity = Math.max(_outbuf.capacity() << 1, data.length);
+            Log.info("Expanding output buffer size [nsize=" + ncapacity + "].");
+            _outbuf = ByteBuffer.allocateDirect(ncapacity);
+	}
+
         boolean fully = true;
         try {
-            // write the message via the connection's object output stream
+//             Log.info("Writing " + data.length + " byte message to " +
-            // (which we configure to write data to our framing output
+//                      conn + ".");
-            // stream)
-            ObjectOutputStream oout = conn.getObjectOutputStream(_framer);
-//             Log.info("Sending " + outmsg + ".");
-            oout.writeObject(outmsg);
-            oout.flush();
 
-            try {
+            // first copy the data into our "direct" output buffer
-                // then write the framed message to the socket
+            _outbuf.put(data);
-                ByteBuffer buffer = _framer.frameAndReturnBuffer();
+            _outbuf.flip();
-                int wrote = conn.getChannel().write(buffer);
-                noteWrite(1, wrote);
 
-                if (buffer.remaining() > 0) {
+            // then write the data to the socket
-                    fully = false;
+            int wrote = conn.getChannel().write(_outbuf);
+            noteWrite(1, wrote);
+
+            if (_outbuf.remaining() > 0) {
+                fully = false;
 //                     Log.info("Partial write [conn=" + conn +
 //                              ", msg=" + StringUtil.shortClassName(outmsg) +
 //                              ", wrote=" + wrote +
 //                              ", size=" + buffer.limit() + "].");
-                    pwh.handlePartialWrite(conn, buffer);
+                pwh.handlePartialWrite(conn, _outbuf);
 
 //                 } else if (wrote > 10000) {
 //                     Log.info("Big write [conn=" + conn +
 //                              ", msg=" + StringUtil.shortClassName(outmsg) +
 //                              ", wrote=" + wrote + "].");
-                }
-
-            } finally {
-                _framer.resetFrame();
             }
 
         } catch (IOException ioe) {
             // instruct the connection to deal with its failure
             conn.handleFailure(ioe);
+
+        } finally {
+            _outbuf.clear();
         }
 
         return fully;
@@ -609,7 +619,11 @@ public class ConnectionManager extends LoopingThread
 
     /**
      * Called by a connection when it has a downstream message that needs
-     * to be delivered.
+     * to be delivered. <em>Note:</em> this method is called as a result
+     * of a call to {@link Connection#postMessage} which happens when
+     * forwarding an event to a client and at the completion of
+     * authentication, both of which <em>should</em> happen only on the
+     * distributed object thread.
      */
     void postMessage (Connection conn, DownstreamMessage msg)
     {
@@ -619,8 +633,29 @@ public class ConnectionManager extends LoopingThread
             Thread.dumpStack();
 
         } else {
-            // slap both these suckers onto the outgoing message queue
+            // flatten this message using the connection's output stream
-            _outq.append(new Tuple(conn, msg));
+            try {
+                ObjectOutputStream oout = conn.getObjectOutputStream(_framer);
+                oout.writeObject(msg);
+                oout.flush();
+
+                // now extract that data into a byte array
+                ByteBuffer buffer = _framer.frameAndReturnBuffer();
+                byte[] data = new byte[buffer.limit()];
+                buffer.get(data);
+                _framer.resetFrame();
+
+//                 Log.info("Flattened " + msg + " into " +
+//                          data.length + " bytes.");
+
+                // and slap both on the queue
+                _outq.append(new Tuple(conn, data));
+
+            } catch (Exception e) {
+                Log.warning("Failure flattening message [conn=" + conn +
+                            ", msg=" + msg + "]. Dropping.");
+                Log.logStackTrace(e);
+            }
         }
     }
 
@@ -710,6 +745,7 @@ public class ConnectionManager extends LoopingThread
 
                 if (_partial.remaining() == 0) {
                     _partial = null;
+                    _partials++;
                 } else {
 //                     Log.info("Still going [conn=" + conn +
 //                              ", wrote=" + wrote +
@@ -719,10 +755,11 @@ public class ConnectionManager extends LoopingThread
             }
 
             while (size() > 0) {
-                DownstreamMessage outmsg = (DownstreamMessage)remove(0);
+                byte[] data = (byte[])remove(0);
                 // if any of these messages are partially written, we have
                 // to stop and wait for the next tick
-                if (!writeMessage(conn, outmsg, this)) {
+                _msgs++;
+                if (!writeMessage(conn, data, this)) {
                     return false;
                 }
             }
@@ -739,9 +776,21 @@ public class ConnectionManager extends LoopingThread
             _partial.flip();
         }
 
+        /**
+         * Returns a string representation of this instance.
+         */
+        public String toString ()
+        {
+            return "[conn=" + conn + ", partials=" + _partials +
+                ", msgs=" + _msgs + "]";
+        }
+
         /** The remains of a message that was only partially written on
          * its first attempt. */
         protected ByteBuffer _partial;
+
+        /** A couple of counters. */
+        protected int _msgs, _partials;
     }
 
     protected int _port;
@@ -758,6 +807,7 @@ public class ConnectionManager extends LoopingThread
 
     protected Queue _outq = new Queue();
     protected FramingOutputStream _framer;
+    protected ByteBuffer _outbuf = ByteBuffer.allocateDirect(64 * 1024);
 
     protected HashMap _oflowqs = new HashMap();
 
@@ -780,6 +830,7 @@ public class ConnectionManager extends LoopingThread
         public void handlePartialWrite (Connection conn, ByteBuffer msgbuf) {
             // if we couldn't write all the data for this message, we'll
             // need to establish an overflow queue
+            Log.info("Starting overflow queue for " + conn + ".");
             _oflowqs.put(conn, new OverflowQueue(conn, msgbuf));
         }
     };