From a285695ca8a280a66a8bd05e054caf2b123bc5c0 Mon Sep 17 00:00:00 2001 From: Michael Bayne Date: Sat, 19 Mar 2005 08:39:19 +0000 Subject: [PATCH] I finally broke down and did the rewrite to fix the potential race condition between the omgr thread and the conmgr thread. Now when the omgr thread processes an event that is going out to the clients, it flattents the message itself for each client that is to receive the message and the flattened data is posted to the conmgr outgoing queue. This means that once an event is finished processing, no further modifications to any of the data associated with the event can effect the data queued up to be sent to the client. This is a good thing, it will eliminate or illuminate a very baffling class of bugs that we've sort of been ignoring because we knew this could be the cause. We used to take an event and flatten it directly into the direct buffer from which we would do our socket write. Now we flatten it into a temporary byte array. This means a metric shitload more garbage generation and collection. We used to do the flattening on the conmgr thread, now we do it on the omgr thread. This means a big redistribution of CPU demand. Either of those things could result in a significant negative impact on our performance, but we'll just have to deploy this stuff and find out. Whee! If it turns out to be a serious problem, there are potential optimizations that could be done by keeping a pool of direct buffers around and flattening messages into them, relying on the fact that the outgoing conmgr queue generally doesn't grow too large and we could allocate tens to a hundred megabytes of memory for the outgoing queue if we really needed to. I'd also like to test the overflow handling stuff more. It didn't really change in that everything just deals with arrays of bytes now instead of unflattened messages, but I'll be more comfortable once I've seen all this in action on ice where there may be few users, but they are just as likely to experience lag and receive an overflow queue as users on the higher traffic servers. There is code to log when overflow queues are created and finally flushed and how much use they got while they were around, so that should give us an indication of whether things are operating properly. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@3419 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../com/threerings/io/FramedInputStream.java | 2 +- .../threerings/io/FramingOutputStream.java | 4 +- .../presents/client/Communicator.java | 3 +- .../presents/server/Authenticator.java | 9 +- .../presents/server/DummyAuthenticator.java | 14 ++- .../presents/server/PresentsClient.java | 15 ++- .../threerings/presents/server/Rejector.java | 14 ++- .../server/net/ConnectionManager.java | 109 +++++++++++++----- 8 files changed, 122 insertions(+), 48 deletions(-) diff --git a/src/java/com/threerings/io/FramedInputStream.java b/src/java/com/threerings/io/FramedInputStream.java index 571e8a720..966944b1b 100644 --- a/src/java/com/threerings/io/FramedInputStream.java +++ b/src/java/com/threerings/io/FramedInputStream.java @@ -1,5 +1,5 @@ // -// $Id: FramedInputStream.java,v 1.8 2004/08/27 02:12:36 mdb Exp $ +// $Id$ // // Narya library - tools for developing networked games // Copyright (C) 2002-2004 Three Rings Design, Inc., All Rights Reserved 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 101958c75..f9d57a75c 100644 --- a/src/java/com/threerings/presents/server/Authenticator.java +++ b/src/java/com/threerings/presents/server/Authenticator.java @@ -1,5 +1,5 @@ // -// $Id: Authenticator.java,v 1.6 2004/08/27 02:20:23 mdb Exp $ +// $Id$ // // Narya library - tools for developing networked games // Copyright (C) 2002-2004 Three Rings Design, Inc., All Rights Reserved @@ -53,7 +53,8 @@ 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}. + * #connectionWasAuthenticated} from the distributed object + * thread. */ public abstract void authenticateConnection (AuthingConnection conn); @@ -61,7 +62,9 @@ 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. + * 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. */ 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 eacd5c7ca..db16643e3 100644 --- a/src/java/com/threerings/presents/server/DummyAuthenticator.java +++ b/src/java/com/threerings/presents/server/DummyAuthenticator.java @@ -1,5 +1,5 @@ // -// $Id: DummyAuthenticator.java,v 1.7 2004/08/27 02:20:23 mdb Exp $ +// $Id$ // // Narya library - tools for developing networked games // Copyright (C) 2002-2004 Three Rings Design, Inc., All Rights Reserved @@ -35,11 +35,15 @@ public class DummyAuthenticator extends Authenticator /** * Accept all authentication requests. */ - public void authenticateConnection (AuthingConnection conn) + public void authenticateConnection (final AuthingConnection conn) { Log.info("Accepting request: " + conn.getAuthRequest()); - AuthResponseData rdata = new AuthResponseData(); - rdata.code = AuthResponseData.SUCCESS; - connectionWasAuthenticated(conn, new AuthResponse(rdata)); + PresentsServer.omgr.postRunnable(new Runnable() { + public void run () { + 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 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/Rejector.java b/src/java/com/threerings/presents/server/Rejector.java index d5e72d520..4681c339f 100644 --- a/src/java/com/threerings/presents/server/Rejector.java +++ b/src/java/com/threerings/presents/server/Rejector.java @@ -1,5 +1,5 @@ // -// $Id: Rejector.java,v 1.2 2004/10/13 01:09:14 mdb Exp $ +// $Id$ package com.threerings.presents.server; @@ -55,12 +55,16 @@ public class Rejector extends PresentsServer protected class RejectingAuthenticator extends Authenticator { /** Reject all authentication requests. */ - public void authenticateConnection (AuthingConnection conn) + public void authenticateConnection (final AuthingConnection conn) { Log.info("Rejecting request: " + conn.getAuthRequest()); - AuthResponseData rdata = new AuthResponseData(); - rdata.code = _errmsg; - connectionWasAuthenticated(conn, new AuthResponse(rdata)); + omgr.postRunnable(new Runnable() { + public void run () { + 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 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)); } };