Don't make a copy of all the events every tick; instead, queue up expiring, failed outgoing connections and then clean them up as part of the tick.
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6761 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
@@ -32,7 +32,6 @@ import java.nio.channels.SelectionKey;
|
|||||||
import java.nio.channels.Selector;
|
import java.nio.channels.Selector;
|
||||||
import java.nio.channels.SocketChannel;
|
import java.nio.channels.SocketChannel;
|
||||||
|
|
||||||
import com.google.common.collect.Lists;
|
|
||||||
import com.google.common.collect.Maps;
|
import com.google.common.collect.Maps;
|
||||||
import com.google.inject.Inject;
|
import com.google.inject.Inject;
|
||||||
import com.google.inject.name.Named;
|
import com.google.inject.name.Named;
|
||||||
@@ -172,7 +171,7 @@ public abstract class ConnectionManager extends LoopingThread
|
|||||||
|
|
||||||
// close connections that have had no network traffic for too long
|
// close connections that have had no network traffic for too long
|
||||||
long idleStamp = iterStamp - _idleTime;
|
long idleStamp = iterStamp - _idleTime;
|
||||||
for (NetEventHandler handler : Lists.newArrayList(_handlers.values())) {
|
for (NetEventHandler handler : _handlers.values()) {
|
||||||
if (handler.checkIdle(idleStamp)) {
|
if (handler.checkIdle(idleStamp)) {
|
||||||
// this will queue the connection for closure on our next tick
|
// this will queue the connection for closure on our next tick
|
||||||
handler.becameIdle();
|
handler.becameIdle();
|
||||||
|
|||||||
@@ -367,32 +367,7 @@ public class PresentsConnectionManager extends ConnectionManager
|
|||||||
} else {
|
} else {
|
||||||
// otherwise we wire up a special event handler that will wait for our socket to
|
// otherwise we wire up a special event handler that will wait for our socket to
|
||||||
// finish the connection process and then wire things up fully
|
// finish the connection process and then wire things up fully
|
||||||
handler = new NetEventHandler() {
|
handler = new OutgoingConnectionHandler(conn);
|
||||||
public int handleEvent (long when) {
|
|
||||||
try {
|
|
||||||
if (sockchan.finishConnect()) {
|
|
||||||
// great, we're ready to roll, wire up the connection
|
|
||||||
conn.selkey = sockchan.register(_selector, SelectionKey.OP_READ);
|
|
||||||
_handlers.put(conn.selkey, conn);
|
|
||||||
log.info("Outgoing connection ready", "conn", conn);
|
|
||||||
}
|
|
||||||
} catch (IOException ioe) {
|
|
||||||
handleError(ioe);
|
|
||||||
}
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
public boolean checkIdle (long idleStamp) {
|
|
||||||
return conn.checkIdle(idleStamp);
|
|
||||||
}
|
|
||||||
public void becameIdle () {
|
|
||||||
handleError(new IOException("Pending connection became idle."));
|
|
||||||
}
|
|
||||||
protected void handleError (IOException ioe) {
|
|
||||||
_handlers.remove(conn.selkey);
|
|
||||||
_oflowqs.remove(conn);
|
|
||||||
conn.connectFailure(ioe);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
_handlers.put(conn.selkey, handler);
|
_handlers.put(conn.selkey, handler);
|
||||||
|
|
||||||
@@ -402,6 +377,18 @@ public class PresentsConnectionManager extends ConnectionManager
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override // from LoopingThread
|
||||||
|
protected void iterate ()
|
||||||
|
{
|
||||||
|
super.iterate();
|
||||||
|
|
||||||
|
// reap any outgoing connection handlers that failed to connect due to idleness
|
||||||
|
OutgoingConnectionHandler handler;
|
||||||
|
while ((handler = _outfailq.getNonBlocking()) != null) {
|
||||||
|
handler.handleError(new IOException("Pending connection became idle."));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@Override // from LoopingThread
|
@Override // from LoopingThread
|
||||||
public boolean isRunning ()
|
public boolean isRunning ()
|
||||||
{
|
{
|
||||||
@@ -552,6 +539,50 @@ public class PresentsConnectionManager extends ConnectionManager
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
protected class OutgoingConnectionHandler implements NetEventHandler
|
||||||
|
{
|
||||||
|
public OutgoingConnectionHandler (Connection conn)
|
||||||
|
{
|
||||||
|
_conn = conn;
|
||||||
|
}
|
||||||
|
|
||||||
|
public int handleEvent (long when)
|
||||||
|
{
|
||||||
|
SocketChannel sockchan = _conn.getChannel();
|
||||||
|
try {
|
||||||
|
if (sockchan.finishConnect()) {
|
||||||
|
// great, we're ready to roll, wire up the connection
|
||||||
|
_conn.selkey = sockchan.register(_selector, SelectionKey.OP_READ);
|
||||||
|
_handlers.put(_conn.selkey, _conn);
|
||||||
|
log.info("Outgoing connection ready", "conn", _conn);
|
||||||
|
}
|
||||||
|
} catch (IOException ioe) {
|
||||||
|
handleError(ioe);
|
||||||
|
}
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
public boolean checkIdle (long idleStamp)
|
||||||
|
{
|
||||||
|
return _conn.checkIdle(idleStamp);
|
||||||
|
}
|
||||||
|
|
||||||
|
public void becameIdle ()
|
||||||
|
{
|
||||||
|
// this failed connection will be cleaned up in the next iterate() tick
|
||||||
|
_outfailq.append(this);
|
||||||
|
}
|
||||||
|
|
||||||
|
protected void handleError (IOException ioe)
|
||||||
|
{
|
||||||
|
_handlers.remove(_conn.selkey);
|
||||||
|
_oflowqs.remove(_conn);
|
||||||
|
_conn.connectFailure(ioe);
|
||||||
|
}
|
||||||
|
|
||||||
|
protected final Connection _conn;
|
||||||
|
}
|
||||||
|
|
||||||
/** Handles client authentication. The base authenticator is injected but optional services
|
/** Handles client authentication. The base authenticator is injected but optional services
|
||||||
* like the PeerManager may replace this authenticator with one that intercepts certain types
|
* like the PeerManager may replace this authenticator with one that intercepts certain types
|
||||||
* of authentication and then passes normal authentications through. */
|
* of authentication and then passes normal authentications through. */
|
||||||
@@ -562,6 +593,9 @@ public class PresentsConnectionManager extends ConnectionManager
|
|||||||
protected Queue<AuthingConnection> _authq = Queue.newQueue();
|
protected Queue<AuthingConnection> _authq = Queue.newQueue();
|
||||||
protected Queue<Tuple<Connection, InetSocketAddress>> _connectq = Queue.newQueue();
|
protected Queue<Tuple<Connection, InetSocketAddress>> _connectq = Queue.newQueue();
|
||||||
|
|
||||||
|
/** failed (idled out) outgoing connections that need to be cleaned up */
|
||||||
|
protected Queue<OutgoingConnectionHandler> _outfailq = Queue.newQueue();
|
||||||
|
|
||||||
protected FramingOutputStream _framer = new FramingOutputStream();
|
protected FramingOutputStream _framer = new FramingOutputStream();
|
||||||
protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream();
|
protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream();
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user