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