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

Implemented code to handle the rather complicated circumstance of not

Browse Source 
being able to send a message immediately to the client due to it being too
large or the client's outgoing networking buffer being full for some other
reason (many messages sent very quickly or client that's reading messages
slowly).

I still need to add code to disconnect a client who fails to read messages
in a sufficiently timely manner.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@2038 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2002-12-07 22:33:04 +00:00
parent 53d4108602
commit d60e6a69d8
1 changed files with 200 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/server/net/ConnectionManager.java
+200 -45
View File
@@ -1,5 +1,5 @@
//
// $Id: ConnectionManager.java,v 1.26 2002/11/18 18:56:52 mdb Exp $
// $Id: ConnectionManager.java,v 1.27 2002/12/07 22:33:04 mdb Exp $
package com.threerings.presents.server.net;
@@ -247,58 +247,41 @@ public class ConnectionManager extends LoopingThread
}
}
// attempt to send any messages waiting on the overflow queues
if (_oflowqs.size() > 0) {
Iterator oqiter = _oflowqs.values().iterator();
while (oqiter.hasNext()) {
OverflowQueue oq = (OverflowQueue)oqiter.next();
try {
// try writing the messages in this overflow queue
if (oq.writeOverflowMessages(iterStamp)) {
// if they were all written, we can remove it
oqiter.remove();
Log.info("Flushed overflow queue for " + oq.conn + ".");
}
} catch (IOException ioe) {
oq.conn.handleFailure(ioe);
}
}
}
// send any messages that are waiting on the outgoing queue
Tuple tup;
while ((tup = (Tuple)_outq.getNonBlocking()) != null) {
Connection conn = (Connection)tup.left;
// if the connection to which this message is destined is
// closed, drop the message and move along quietly; this is
// perfectly legal, a user can logoff whenever they like, even
// if we still have things to tell them; such is life in a
// fully asynchronous distributed system
if (conn.isClosed()) {
// if an overflow queue exists for this client, go ahead and
// slap the message on there because we can't send it until
// all other messages in their queue have gone out
OverflowQueue oqueue = (OverflowQueue)_oflowqs.get(conn);
if (oqueue != null) {
oqueue.add(tup.right);
continue;
}
DownstreamMessage outmsg = (DownstreamMessage)tup.right;
try {
// write the message via the connection's object output
// stream (which is configured to write data to our
// framing output stream)
ObjectOutputStream oout = conn.getObjectOutputStream(_framer);
// Log.info("Sending " + outmsg + ".");
oout.writeObject(outmsg);
oout.flush();
// then write framed message to real output stream
try {
ByteBuffer buffer = _framer.frameAndReturnBuffer();
int wrote = conn.getChannel().write(buffer);
if (wrote != buffer.limit()) {
Log.warning("Aiya! Failed to write complete message " +
"[conn=" + conn + ", wrote=" + wrote +
", needed=" + buffer.limit() + "].");
// TODO: give them the boot
// } else {
// Log.info("Sent [msg=" + outmsg +
// ", size=" + wrote + "].");
}
int out = buffer.limit();
synchronized (this) {
_bytesOut += wrote;
_msgsOut++;
}
} finally {
_framer.resetFrame();
}
} catch (IOException ioe) {
// instruct the connection to deal with its failure
conn.handleFailure(ioe);
}
// otherwise write the message out to the client directly
writeMessage(conn, (DownstreamMessage)tup.right, _oflowHandler);
}
// check for connections that have completed authentication
@@ -390,6 +373,74 @@ public class ConnectionManager extends LoopingThread
ready.clear();
}
/**
* Writes a message out to a connection, passing the buck to the
* partial write handler if the entire message could not be written.
*
* @return true if the message was fully written, false if it was
* partially written (in which case the partial message handler will
* have been invoked).
*/
protected boolean writeMessage (
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
// legal, a user can logoff whenever they like, even if we still
// have things to tell them; such is life in a fully asynchronous
// distributed system
if (conn.isClosed()) {
return true;
}
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();
try {
// then write the framed message to the socket
ByteBuffer buffer = _framer.frameAndReturnBuffer();
int wrote = conn.getChannel().write(buffer);
noteWrite(1, wrote);
if (buffer.remaining() > 0) {
fully = false;
Log.info("Partial write [conn=" + conn +
", msg=" + outmsg + ", wrote=" + wrote +
", size=" + buffer.limit() + "].");
pwh.handlePartialWrite(conn, buffer);
} else if (wrote > 10000) {
Log.info("Big write [conn=" + conn + ", msg=" + outmsg +
", wrote=" + wrote + "].");
}
} finally {
_framer.resetFrame();
}
} catch (IOException ioe) {
// instruct the connection to deal with its failure
conn.handleFailure(ioe);
}
return fully;
}
/** Called by {@link #writeMessage} and friends when they write data
* over the network. */
protected final synchronized void noteWrite (int msgs, int bytes)
{
_msgsOut += msgs;
_bytesOut += bytes;
}
// documentation inherited
protected void handleIterateFailure (Exception e)
{
@@ -481,6 +532,7 @@ public class ConnectionManager extends LoopingThread
// remove this connection from our mapping (it is automatically
// removed from the Selector when the socket is closed)
_handlers.remove(conn.getSelectionKey());
_oflowqs.remove(conn);
// let our observers know what's up
notifyObservers(CONNECTION_FAILED, conn, ioe, null);
@@ -494,11 +546,103 @@ public class ConnectionManager extends LoopingThread
// remove this connection from our mapping (it is automatically
// removed from the Selector when the socket is closed)
_handlers.remove(conn.getSelectionKey());
_oflowqs.remove(conn);
// let our observers know what's up
notifyObservers(CONNECTION_CLOSED, conn, null, null);
}
/** Used to handle partial writes in {@link #writeMessage}. */
protected static interface PartialWriteHandler
{
public void handlePartialWrite (Connection conn, ByteBuffer buffer);
}
/**
* Used to handle messages for a client whose network buffer has
* filled up because their outgoing network buffer has filled up. This
* can happen if the client receives many messages in rapid succession
* or if they receive very large messages or if they become
* unresponsive and stop acknowledging network packets sent by the
* server. We want to accomodate the first to circumstances and
* recognize the third as quickly as possible so that we can
* disconnect the client and propagate that information up to the
* higher levels so that further messages are not queued up for the
* unresponsive client.
*/
protected class OverflowQueue extends ArrayList
implements PartialWriteHandler
{
/** The connection for which we're managing overflow. */
public Connection conn;
/**
* Creates a new overflow queue for the supplied connection and
* with the supplied initial partial message.
*/
public OverflowQueue (Connection conn, ByteBuffer message)
{
this.conn = conn;
// set up our initial _partial buffer
handlePartialWrite(conn, message);
}
/**
* Called each time through the {@link ConnectionManager#iterate}
* loop, this attempts to send any remaining partial message and
* all subsequent messages in the overflow queue.
*
* @return true if all messages in this queue were successfully
* sent, false if there remains data to be sent on the next loop.
*
* @throws IOException if an error occurs writing data to the
* connection or if we have been unable to write any data to the
* connection for ten seconds.
*/
public boolean writeOverflowMessages (long iterStamp)
throws IOException
{
// write any partial message if we have one
if (_partial != null) {
// write all we can of our partial buffer
int wrote = conn.getChannel().write(_partial);
noteWrite(0, wrote);
if (_partial.remaining() == 0) {
_partial = null;
} else {
Log.info("Still going [conn=" + conn + ", wrote=" + wrote +
", remain=" + _partial.remaining() + "].");
return false;
}
}
while (size() > 0) {
DownstreamMessage outmsg = (DownstreamMessage)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)) {
return false;
}
}
return true;
}
// documentation inherited
public void handlePartialWrite (Connection conn, ByteBuffer buffer)
{
// set up our _partial buffer
_partial = ByteBuffer.allocate(buffer.remaining());
_partial.put(buffer);
_partial.flip();
}
/** The remains of a message that was only partially written on
* its first attempt. */
protected ByteBuffer _partial;
}
protected int _port;
protected Authenticator _author;
protected Selector _selector;
@@ -513,6 +657,8 @@ public class ConnectionManager extends LoopingThread
protected Queue _outq = new Queue();
protected FramingOutputStream _framer;
protected HashMap _oflowqs = new HashMap();
protected ArrayList _observers = new ArrayList();
/** Bytes in and out in the last reporting period. */
@@ -521,6 +667,15 @@ public class ConnectionManager extends LoopingThread
/** Messages read and written in the last reporting period. */
protected int _msgsIn, _msgsOut;
/** Used to create an overflow queue on the first partial write. */
protected PartialWriteHandler _oflowHandler = new PartialWriteHandler() {
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
_oflowqs.put(conn, new OverflowQueue(conn, msgbuf));
}
};
/**
* How long we wait for network events before checking our running
* flag to see if we should still be running. We don't want to loop