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));
}
};