From d58d1001acb82148377cdba278be1970a9787a88 Mon Sep 17 00:00:00 2001 From: Ray Greenwell Date: Mon, 28 Mar 2005 19:33:28 +0000 Subject: [PATCH] Roll it all back. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@3432 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../threerings/io/FramingOutputStream.java | 4 +- .../presents/client/Communicator.java | 3 +- .../presents/server/Authenticator.java | 7 +- .../presents/server/DummyAuthenticator.java | 12 +- .../presents/server/PresentsClient.java | 15 +-- .../threerings/presents/server/Rejector.java | 12 +- .../server/net/ConnectionManager.java | 109 +++++------------- 7 files changed, 44 insertions(+), 118 deletions(-) diff --git a/src/java/com/threerings/io/FramingOutputStream.java b/src/java/com/threerings/io/FramingOutputStream.java index b1d9c0ecd..a7e4c8119 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.allocate(INITIAL_BUFFER_SIZE); + _buffer = ByteBuffer.allocateDirect(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.allocate(ncapacity); + ByteBuffer newbuf = ByteBuffer.allocateDirect(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 fbcaec9f9..396965ca3 100644 --- a/src/java/com/threerings/presents/client/Communicator.java +++ b/src/java/com/threerings/presents/client/Communicator.java @@ -316,8 +316,7 @@ 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 f9d57a75c..eda0a38a0 100644 --- a/src/java/com/threerings/presents/server/Authenticator.java +++ b/src/java/com/threerings/presents/server/Authenticator.java @@ -53,8 +53,7 @@ public abstract class Authenticator * off a task on the {@link PresentsServer#invoker} to perform the * authentication. When the authentication is complete, the * authenticator implementation should call {@link - * #connectionWasAuthenticated} from the distributed object - * thread. + * #connectionWasAuthenticated}. */ public abstract void authenticateConnection (AuthingConnection conn); @@ -62,9 +61,7 @@ public abstract class Authenticator * This is called by authenticator implementations when they have * completed the authentication process. It will deliver the response * to the user and let the connection manager know to report the - * authentication if it succeeded. Note: this method should - * only be called on the distributed object thread as all messages - * being sent to the client should originate therefrom. + * authentication if it succeeded. */ protected void connectionWasAuthenticated ( AuthingConnection conn, AuthResponse rsp) diff --git a/src/java/com/threerings/presents/server/DummyAuthenticator.java b/src/java/com/threerings/presents/server/DummyAuthenticator.java index db16643e3..823d10089 100644 --- a/src/java/com/threerings/presents/server/DummyAuthenticator.java +++ b/src/java/com/threerings/presents/server/DummyAuthenticator.java @@ -35,15 +35,11 @@ public class DummyAuthenticator extends Authenticator /** * Accept all authentication requests. */ - public void authenticateConnection (final AuthingConnection conn) + public void authenticateConnection (AuthingConnection conn) { Log.info("Accepting request: " + conn.getAuthRequest()); - PresentsServer.omgr.postRunnable(new Runnable() { - public void run () { - AuthResponseData rdata = new AuthResponseData(); - rdata.code = AuthResponseData.SUCCESS; - connectionWasAuthenticated(conn, new AuthResponse(rdata)); - } - }); + AuthResponseData rdata = new AuthResponseData(); + rdata.code = AuthResponseData.SUCCESS; + connectionWasAuthenticated(conn, new AuthResponse(rdata)); } } diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java index 2814803c0..30e4066d2 100644 --- a/src/java/com/threerings/presents/server/PresentsClient.java +++ b/src/java/com/threerings/presents/server/PresentsClient.java @@ -758,17 +758,6 @@ 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) { @@ -867,7 +856,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.safePostMessage(new UnsubscribeResponse(oid)); + client.postMessage(new UnsubscribeResponse(oid)); } } @@ -901,7 +890,7 @@ public class PresentsClient { // send a pong response PingRequest req = (PingRequest)msg; - client.safePostMessage(new PongResponse(req.getUnpackStamp())); + client.postMessage(new PongResponse(req.getUnpackStamp())); } } diff --git a/src/java/com/threerings/presents/server/Rejector.java b/src/java/com/threerings/presents/server/Rejector.java index 4681c339f..8a14ed0fa 100644 --- a/src/java/com/threerings/presents/server/Rejector.java +++ b/src/java/com/threerings/presents/server/Rejector.java @@ -55,16 +55,12 @@ public class Rejector extends PresentsServer protected class RejectingAuthenticator extends Authenticator { /** Reject all authentication requests. */ - public void authenticateConnection (final AuthingConnection conn) + public void authenticateConnection (AuthingConnection conn) { Log.info("Rejecting request: " + conn.getAuthRequest()); - omgr.postRunnable(new Runnable() { - public void run () { - AuthResponseData rdata = new AuthResponseData(); - rdata.code = _errmsg; - connectionWasAuthenticated(conn, new AuthResponse(rdata)); - } - }); + AuthResponseData rdata = new AuthResponseData(); + rdata.code = _errmsg; + connectionWasAuthenticated(conn, new AuthResponse(rdata)); } } diff --git a/src/java/com/threerings/presents/server/net/ConnectionManager.java b/src/java/com/threerings/presents/server/net/ConnectionManager.java index 5e0e52820..0e3abf9c0 100644 --- a/src/java/com/threerings/presents/server/net/ConnectionManager.java +++ b/src/java/com/threerings/presents/server/net/ConnectionManager.java @@ -333,7 +333,8 @@ 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 " + oq + "."); +// Log.info("Flushed overflow queue for " + +// oq.conn + "."); } } catch (IOException ioe) { @@ -363,7 +364,7 @@ public class ConnectionManager extends LoopingThread } // otherwise write the message out to the client directly - writeMessage(conn, (byte[])tup.right, _oflowHandler); + writeMessage(conn, (DownstreamMessage)tup.right, _oflowHandler); } // check for connections that have completed authentication @@ -471,7 +472,7 @@ public class ConnectionManager extends LoopingThread * have been invoked). */ protected boolean writeMessage ( - Connection conn, byte[] data, PartialWriteHandler pwh) + Connection conn, DownstreamMessage outmsg, PartialWriteHandler pwh) { // if the connection to which this message is destined is closed, // drop the message and move along quietly; this is perfectly @@ -481,55 +482,44 @@ 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 { -// Log.info("Writing " + data.length + " byte message to " + -// conn + "."); + // 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(); - // first copy the data into our "direct" output buffer - _outbuf.put(data); - _outbuf.flip(); + try { + // then write the framed message to the socket + ByteBuffer buffer = _framer.frameAndReturnBuffer(); + int wrote = conn.getChannel().write(buffer); + noteWrite(1, wrote); - // then write the data to the socket - int wrote = conn.getChannel().write(_outbuf); - noteWrite(1, wrote); - - if (_outbuf.remaining() > 0) { - fully = false; + if (buffer.remaining() > 0) { + fully = false; // Log.info("Partial write [conn=" + conn + // ", msg=" + StringUtil.shortClassName(outmsg) + // ", wrote=" + wrote + // ", size=" + buffer.limit() + "]."); - pwh.handlePartialWrite(conn, _outbuf); + pwh.handlePartialWrite(conn, buffer); // } 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; @@ -619,11 +609,7 @@ public class ConnectionManager extends LoopingThread /** * 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} 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. + * to be delivered. */ void postMessage (Connection conn, DownstreamMessage msg) { @@ -633,29 +619,8 @@ public class ConnectionManager extends LoopingThread Thread.dumpStack(); } else { - // 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); - } + // slap both these suckers onto the outgoing message queue + _outq.append(new Tuple(conn, msg)); } } @@ -745,7 +710,6 @@ public class ConnectionManager extends LoopingThread if (_partial.remaining() == 0) { _partial = null; - _partials++; } else { // Log.info("Still going [conn=" + conn + // ", wrote=" + wrote + @@ -755,11 +719,10 @@ public class ConnectionManager extends LoopingThread } while (size() > 0) { - byte[] data = (byte[])remove(0); + DownstreamMessage outmsg = (DownstreamMessage)remove(0); // if any of these messages are partially written, we have // to stop and wait for the next tick - _msgs++; - if (!writeMessage(conn, data, this)) { + if (!writeMessage(conn, outmsg, this)) { return false; } } @@ -776,21 +739,9 @@ 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; @@ -807,7 +758,6 @@ 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(); @@ -830,7 +780,6 @@ 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)); } };