From a4ac47fb04e8dd9c84c12345925144ee2e9ca30c Mon Sep 17 00:00:00 2001 From: Michael Bayne Date: Mon, 28 Mar 2005 21:49:19 +0000 Subject: [PATCH] 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 --- .../threerings/io/FramingOutputStream.java | 4 +- .../presents/client/Communicator.java | 3 +- .../presents/server/Authenticator.java | 28 +++-- .../presents/server/PresentsClient.java | 15 ++- .../server/net/ConnectionManager.java | 109 +++++++++++++----- 5 files changed, 114 insertions(+), 45 deletions(-) diff --git a/src/java/com/threerings/io/FramingOutputStream.java b/src/java/com/threerings/io/FramingOutputStream.java index a7e4c8119..b1d9c0ecd 100644 --- a/src/java/com/threerings/io/FramingOutputStream.java +++ b/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; } diff --git a/src/java/com/threerings/presents/client/Communicator.java b/src/java/com/threerings/presents/client/Communicator.java index 396965ca3..fbcaec9f9 100644 --- a/src/java/com/threerings/presents/client/Communicator.java +++ b/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() + "]."); } diff --git a/src/java/com/threerings/presents/server/Authenticator.java b/src/java/com/threerings/presents/server/Authenticator.java index eda0a38a0..a955a7d9f 100644 --- a/src/java/com/threerings/presents/server/Authenticator.java +++ b/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 - conn.postMessage(rsp); + // make double plus extra sure we're on the omgr thread + 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 - // that we have access to it later in the authentication process - conn.setAuthResponse(rsp); + // send the response back to the client + conn.postMessage(rsp); - // if the authentication request was granted, let the connection - // manager know that we just authed - if (AuthResponseData.SUCCESS.equals(rsp.getData().code)) { - _conmgr.connectionDidAuthenticate(conn); - } + // if the authentication request was granted, let the + // connection manager know that we just authed + if (AuthResponseData.SUCCESS.equals(rsp.getData().code)) { + _conmgr.connectionDidAuthenticate(conn); + } + } + }); } /** The connection manager with which we're working. */ diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java index 30e4066d2..2814803c0 100644 --- a/src/java/com/threerings/presents/server/PresentsClient.java +++ b/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())); } } diff --git a/src/java/com/threerings/presents/server/net/ConnectionManager.java b/src/java/com/threerings/presents/server/net/ConnectionManager.java index 0e3abf9c0..5e0e52820 100644 --- a/src/java/com/threerings/presents/server/net/ConnectionManager.java +++ b/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 " + -// oq.conn + "."); + Log.info("Flushed overflow queue " + oq + "."); } } 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 - // (which we configure to write data to our framing output - // stream) - ObjectOutputStream oout = conn.getObjectOutputStream(_framer); -// Log.info("Sending " + outmsg + "."); - oout.writeObject(outmsg); - oout.flush(); +// Log.info("Writing " + data.length + " byte message to " + +// conn + "."); - try { - // then write the framed message to the socket - ByteBuffer buffer = _framer.frameAndReturnBuffer(); - int wrote = conn.getChannel().write(buffer); - noteWrite(1, wrote); + // first copy the data into our "direct" output buffer + _outbuf.put(data); + _outbuf.flip(); - if (buffer.remaining() > 0) { - fully = false; + // then write the data to the socket + 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. Note: 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 should 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 - _outq.append(new Tuple(conn, msg)); + // flatten this message using the connection's output stream + 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)); } };