AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

I finally broke down and did the rewrite to fix the potential race

Browse Source 
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
This commit is contained in:
Michael Bayne
2005-03-19 08:39:19 +00:00
parent 0e2f51b91f
commit a285695ca8
8 changed files with 122 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/server/net/ConnectionManager.java
+80 -29
View File
@@ -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. <em>Note:</em> 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 <em>should</em> 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));
}
};