From 89f02fc72882dee124f416f6c1b8f4a86005e7c2 Mon Sep 17 00:00:00 2001 From: Michael Bayne Date: Fri, 7 Nov 2008 00:27:07 +0000 Subject: [PATCH] Created ServerCommunicator which does all I/O using the ConnectionManager's polled socket system. This much more cleanly and efficiently integrates peer (and other server to server) connections into the normal server I/O framework and should also eliminate for good the annoying server hangs that result when our old blocking client I/O threads got their pants wedged. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5510 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../presents/client/BlockingCommunicator.java | 89 ++--- .../threerings/presents/client/Client.java | 188 ++++------ .../presents/client/ClientDObjectMgr.java | 11 +- .../presents/client/Communicator.java | 68 +++- .../ObserverOps.java} | 34 +- .../presents/net/DownstreamMessage.java | 11 +- .../com/threerings/presents/net/Message.java | 3 + .../presents/net/UpstreamMessage.java | 28 +- .../presents/peer/server/PeerManager.java | 2 +- .../presents/peer/server/PeerNode.java | 29 +- .../presents/server/PresentsSession.java | 29 +- .../server/net/AuthingConnection.java | 73 ++-- .../presents/server/net/Connection.java | 332 +++++++++--------- .../server/net/ConnectionManager.java | 217 ++++++++---- .../server/net/RunningConnection.java | 61 ---- .../server/net/ServerCommunicator.java | 196 +++++++++++ 16 files changed, 777 insertions(+), 594 deletions(-) rename src/java/com/threerings/presents/{server/net/MessageHandler.java => client/ObserverOps.java} (51%) delete mode 100644 src/java/com/threerings/presents/server/net/RunningConnection.java create mode 100644 src/java/com/threerings/presents/server/net/ServerCommunicator.java diff --git a/src/java/com/threerings/presents/client/BlockingCommunicator.java b/src/java/com/threerings/presents/client/BlockingCommunicator.java index 6d8f4ec9b..e321f2ae0 100644 --- a/src/java/com/threerings/presents/client/BlockingCommunicator.java +++ b/src/java/com/threerings/presents/client/BlockingCommunicator.java @@ -183,17 +183,10 @@ public class BlockingCommunicator extends Communicator return _lastWrite; } - /** - * Callback called by the reader when the authentication process completes successfully. Here - * we extract the bootstrap information for the client and start up the writer thread to manage - * the other half of our bi-directional message stream. - */ + @Override // from Communicator protected synchronized void logonSucceeded (AuthResponseData data) { - log.debug("Logon succeeded: " + data); - - // create our distributed object manager - _omgr = new ClientDObjectMgr(this, _client); + super.logonSucceeded(data); // create a new writer thread and start it up if (_writer != null) { @@ -201,19 +194,13 @@ public class BlockingCommunicator extends Communicator } _writer = new Writer(); _writer.start(); - - // fill the auth data into the client's local field so that it can be requested by external - // entities - _client._authData = data; - - // wait for the bootstrap notification before we claim that we're actually logged on } /** * Callback called by the reader or writer thread when something goes awry with our socket * connection to the server. */ - protected synchronized void connectionFailed (IOException ioe) + protected synchronized void connectionFailed (final IOException ioe) { // make sure the socket isn't already closed down (meaning we've already dealt with the // failed connection) @@ -224,7 +211,11 @@ public class BlockingCommunicator extends Communicator log.info("Connection failed", ioe); // let the client know that things went south - _client.notifyObservers(Client.CLIENT_CONNECTION_FAILED, ioe); + notifyClientObservers(new ObserverOps.Client() { + protected void notify (ClientObserver obs) { + obs.clientConnectionFailed(_client, ioe); + } + }); // and request that we go through the motions of logging off logoff(); @@ -260,7 +251,7 @@ public class BlockingCommunicator extends Communicator closeChannel(); // let the client know when we finally go away - _client.cleanup(_logonError); + clientCleanup(_logonError); } log.debug("Reader thread exited."); @@ -276,7 +267,11 @@ public class BlockingCommunicator extends Communicator log.debug("Writer thread exited."); // let the client observers know that we're logged off - _client.notifyObservers(Client.CLIENT_DID_LOGOFF, null); + notifyClientObservers(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientDidLogoff(_client); + } + }); // now that the writer thread has gone away, we can safely close our socket and let the // client know that the logoff process has completed @@ -284,7 +279,7 @@ public class BlockingCommunicator extends Communicator // let the client know when we finally go away if (_reader == null) { - _client.cleanup(_logonError); + clientCleanup(_logonError); } } @@ -439,14 +434,6 @@ public class BlockingCommunicator extends Communicator _datagramChannel.write(buf); } - /** - * Makes a note of the time at which we last communicated with the server. - */ - protected synchronized void updateWriteStamp () - { - _lastWrite = System.currentTimeMillis(); - } - /** * Reads a new message from the socket (blocking until a message has arrived). */ @@ -504,16 +491,6 @@ public class BlockingCommunicator extends Communicator } } - /** - * Callback called by the reader thread when it has parsed a new message from the socket and - * wishes to have it processed. - */ - protected void processMessage (DownstreamMessage msg) - { - // post this message to the dobjmgr queue - _omgr.processMessage(msg); - } - protected void openChannel (InetAddress host) throws IOException { @@ -545,13 +522,17 @@ public class BlockingCommunicator extends Communicator @Override protected void willStart () { - // first we connect and authenticate with the server try { // connect to the server connect(); - // then authenticate - logon(); + // construct an auth request and send it + sendMessage(new AuthRequest(_client.getCredentials(), _client.getVersion(), + _client.getBootGroups())); + + // now wait for the auth response + log.debug("Waiting for auth response."); + gotAuthResponse((AuthResponse)receiveMessage()); } catch (Exception e) { log.debug("Logon failed: " + e); @@ -586,30 +567,6 @@ public class BlockingCommunicator extends Communicator _oout = new ObjectOutputStream(_fout); } - protected void logon () - throws IOException, LogonException - { - // construct an auth request and send it - AuthRequest req = new AuthRequest( - _client.getCredentials(), _client.getVersion(), _client.getBootGroups()); - sendMessage(req); - - // now wait for the auth response - log.debug("Waiting for auth response."); - AuthResponse rsp = (AuthResponse)receiveMessage(); - AuthResponseData data = rsp.getData(); - log.debug("Got auth response: " + data); - - // if the auth request failed, we want to let the communicator know by throwing a logon - // exception - if (!data.code.equals(AuthResponseData.SUCCESS)) { - throw new LogonException(data.code); - } - - // we're all clear. let the communicator know that we're in - logonSucceeded(data); - } - // now that we're authenticated, we manage the reading half of things by continuously // reading messages from the socket and processing them @Override @@ -961,7 +918,6 @@ public class BlockingCommunicator extends Communicator protected DatagramChannel _datagramChannel; protected Queue _dataq = new Queue(); - protected long _lastWrite; protected Exception _logonError; /** We use this to frame our upstream messages. */ @@ -983,7 +939,6 @@ public class BlockingCommunicator extends Communicator protected DatagramSequencer _sequencer; - protected ClientDObjectMgr _omgr; protected ClassLoader _loader; /** The number of times per port to try to establish a datagram "connection". */ diff --git a/src/java/com/threerings/presents/client/Client.java b/src/java/com/threerings/presents/client/Client.java index 8500ebc48..9c1d2be09 100644 --- a/src/java/com/threerings/presents/client/Client.java +++ b/src/java/com/threerings/presents/client/Client.java @@ -418,8 +418,12 @@ public class Client return false; } - // notify our observers - notifyObservers(CLIENT_WILL_LOGON, null); + // notify our observers immediately + _observers.apply(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientWillLogon(Client.this); + } + }); // otherwise create a new communicator instance and start it up. this will initiate the // logon process @@ -504,7 +508,15 @@ public class Client } // if the request is abortable, let's run it past the observers before we act upon it - if (abortable && notifyObservers(CLIENT_WILL_LOGOFF, null)) { + final boolean[] rejected = new boolean[] { false }; + _observers.apply(new ObserverOps.Client() { + protected void notify (ClientObserver obs) { + if (!obs.clientWillLogoff(Client.this)) { + rejected[0] = true; + } + } + }); + if (abortable && rejected[0]) { return false; } @@ -527,7 +539,12 @@ public class Client public String[] prepareStandaloneLogon () { _standalone = true; - notifyObservers(CLIENT_WILL_LOGON, null); + // notify our observers immediately + _observers.apply(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientWillLogon(Client.this); + } + }); return getBootGroups(); } @@ -548,7 +565,11 @@ public class Client */ public void standaloneLogoff () { - notifyObservers(CLIENT_DID_LOGOFF, null); + notifyObservers(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientDidLogoff(Client.this); + } + }); cleanup(null); // this will set _standalone to false } @@ -715,9 +736,9 @@ public class Client */ protected void reportLogonTribulations (final LogonException cause) { - _runQueue.postRunnable(new Runnable() { - public void run () { - notifyObservers(CLIENT_FAILED_TO_LOGON, cause); + notifyObservers(new ObserverOps.Client() { + protected void notify (ClientObserver obs) { + obs.clientFailedToLogon(Client.this, cause); } }); } @@ -732,16 +753,24 @@ public class Client _clobj = clobj; // let the client know that logon has now fully succeeded - notifyObservers(CLIENT_DID_LOGON, null); + notifyObservers(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientDidLogon(Client.this); + } + }); } /** * Called by the invocation director if it fails to subscribe to the client object after logon. */ - protected void getClientObjectFailed (Exception cause) + protected void getClientObjectFailed (final Exception cause) { // pass the buck onto the listeners - notifyObservers(CLIENT_FAILED_TO_LOGON, cause); + notifyObservers(new ObserverOps.Client() { + protected void notify (ClientObserver obs) { + obs.clientFailedToLogon(Client.this, cause); + } + }); } /** @@ -753,32 +782,23 @@ public class Client _cloid = _clobj.getOid(); // report to our observers - notifyObservers(CLIENT_OBJECT_CHANGED, null); + notifyObservers(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientObjectDidChange(Client.this); + } + }); } - protected boolean notifyObservers (int code, Exception cause) + protected void notifyObservers (final ObserverOps.Session op) { - final Notifier noty = new Notifier(code, cause); - Runnable unit = new Runnable() { - public void run () { - synchronized (_observers) { - _observers.apply(noty); - } - } - }; - - // we need to run immediately if this is WILL_LOGON, WILL_LOGOFF or if we have no RunQueue - // (which currently only happens in some really obscure circumstances where we're using a - // Client instance on the server so that we can sort of pretend to be a real client) - if (code == CLIENT_WILL_LOGON || code == CLIENT_WILL_LOGOFF || _runQueue == null) { - unit.run(); - return noty.getRejected(); - + if (_runQueue == null) { + _observers.apply(op); } else { - // otherwise we can queue this notification up with our RunQueue and ensure that it's - // run on the proper thread - _runQueue.postRunnable(unit); - return false; + _runQueue.postRunnable(new Runnable() { + public void run () { + _observers.apply(op); + } + }); } } @@ -809,11 +829,15 @@ public class Client // now that the communicator is cleaned up; this allows a logon failure listener to // immediately try another logon (hopefully with something changed like the server // or port) - if (logonError != null) { - notifyObservers(CLIENT_FAILED_TO_LOGON, logonError); - } else { - notifyObservers(CLIENT_DID_CLEAR, null); - } + notifyObservers(new ObserverOps.Client() { + protected void notify (ClientObserver obs) { + if (logonError != null) { + obs.clientFailedToLogon(Client.this, logonError); + } else { + obs.clientDidClear(Client.this); + } + } + }); } }); } @@ -844,81 +868,6 @@ public class Client return false; } - /** - * Used to notify client observers of events. - */ - protected class Notifier - implements ObserverList.ObserverOp - { - public Notifier (int code, Exception cause) - { - _code = code; - _cause = cause; - _rejected = false; - } - - public boolean getRejected () - { - return _rejected; - } - - public boolean apply (SessionObserver obs) - { - switch (_code) { - case CLIENT_WILL_LOGON: - obs.clientWillLogon(Client.this); - break; - - case CLIENT_DID_LOGON: - obs.clientDidLogon(Client.this); - break; - - case CLIENT_FAILED_TO_LOGON: - if (obs instanceof ClientObserver) { - ((ClientObserver)obs).clientFailedToLogon(Client.this, _cause); - } - break; - - case CLIENT_OBJECT_CHANGED: - obs.clientObjectDidChange(Client.this); - break; - - case CLIENT_CONNECTION_FAILED: - if (obs instanceof ClientObserver) { - ((ClientObserver)obs).clientConnectionFailed(Client.this, _cause); - } - break; - - case CLIENT_WILL_LOGOFF: - if (obs instanceof ClientObserver) { - if (!((ClientObserver)obs).clientWillLogoff(Client.this)) { - _rejected = true; - } - } - break; - - case CLIENT_DID_LOGOFF: - obs.clientDidLogoff(Client.this); - break; - - case CLIENT_DID_CLEAR: - if (obs instanceof ClientObserver) { - ((ClientObserver)obs).clientDidClear(Client.this); - } - break; - - default: - throw new RuntimeException("Invalid code supplied to notifyObservers: " + _code); - } - - return true; - } - - protected int _code; - protected Exception _cause; - protected boolean _rejected; - } - /** Handles the process of switching between servers. * See {@link Client#moveToServer(String,int[],int[],ConfirmListener)}. */ protected class ServerSwitcher extends ClientAdapter @@ -1037,8 +986,7 @@ public class Client protected int[] _datagramPorts; /** Our list of client observers. */ - protected ObserverList _observers = - new ObserverList(ObserverList.SAFE_IN_ORDER_NOTIFY); + protected ObserverList _observers = ObserverList.newSafeInOrder(); /** The entity that manages our network communications. */ protected Communicator _comm; @@ -1075,14 +1023,4 @@ public class Client /** How often we recompute our time offset from the server. */ protected static final long CLOCK_SYNC_INTERVAL = 600 * 1000L; - - // client observer codes - static final int CLIENT_WILL_LOGON = 0; - static final int CLIENT_DID_LOGON = 1; - static final int CLIENT_FAILED_TO_LOGON = 2; - static final int CLIENT_OBJECT_CHANGED = 3; - static final int CLIENT_CONNECTION_FAILED = 4; - static final int CLIENT_WILL_LOGOFF = 5; - static final int CLIENT_DID_LOGOFF = 6; - static final int CLIENT_DID_CLEAR = 7; } diff --git a/src/java/com/threerings/presents/client/ClientDObjectMgr.java b/src/java/com/threerings/presents/client/ClientDObjectMgr.java index 45ea11ede..58daba5f8 100644 --- a/src/java/com/threerings/presents/client/ClientDObjectMgr.java +++ b/src/java/com/threerings/presents/client/ClientDObjectMgr.java @@ -44,10 +44,10 @@ import com.threerings.presents.dobj.ObjectDestroyedEvent; import com.threerings.presents.dobj.Subscriber; import com.threerings.presents.net.BootstrapData; import com.threerings.presents.net.BootstrapNotification; -import com.threerings.presents.net.DownstreamMessage; import com.threerings.presents.net.EventNotification; import com.threerings.presents.net.FailureResponse; import com.threerings.presents.net.ForwardEventRequest; +import com.threerings.presents.net.Message; import com.threerings.presents.net.ObjectResponse; import com.threerings.presents.net.PongResponse; import com.threerings.presents.net.SubscribeRequest; @@ -157,10 +157,10 @@ public class ClientDObjectMgr } /** - * Called by the communicator when a downstream message arrives from the network layer. We - * queue it up for processing and request some processing time on the main thread. + * Called by the communicator when a message arrives from the network layer. We queue it up for + * processing and request some processing time on the main thread. */ - public void processMessage (DownstreamMessage msg) + public void processMessage (Message msg) { // append it to our queue _actions.append(msg); @@ -211,6 +211,9 @@ public class ClientDObjectMgr } else { doUnsubscribe(act.oid, act.target); } + + } else { + log.warning("Unknown action", "action", obj); } } } diff --git a/src/java/com/threerings/presents/client/Communicator.java b/src/java/com/threerings/presents/client/Communicator.java index 20c852148..888bbe426 100644 --- a/src/java/com/threerings/presents/client/Communicator.java +++ b/src/java/com/threerings/presents/client/Communicator.java @@ -21,6 +21,9 @@ package com.threerings.presents.client; +import com.threerings.presents.net.AuthResponse; +import com.threerings.presents.net.AuthResponseData; +import com.threerings.presents.net.Message; import com.threerings.presents.net.UpstreamMessage; /** @@ -66,7 +69,70 @@ public abstract class Communicator /** * Returns the time at which we last sent a packet to the server. */ - public abstract long getLastWrite (); + public long getLastWrite () + { + return _lastWrite; + } + + /** + * Makes a note of the time at which we last communicated with the server. + */ + protected synchronized void updateWriteStamp () + { + _lastWrite = System.currentTimeMillis(); + } + + /** + * Subclasses must call this method when they receive the authentication response. + */ + protected void gotAuthResponse (AuthResponse rsp) + throws LogonException + { + AuthResponseData data = rsp.getData(); + if (!data.code.equals(AuthResponseData.SUCCESS)) { + throw new LogonException(data.code); + } + logonSucceeded(data); + } + + /** + * Called when the authentication process completes successfully. Derived classes can override + * this method and complete any initialization that need wait for authentication success. + */ + protected synchronized void logonSucceeded (AuthResponseData data) + { + // create our distributed object manager + _omgr = new ClientDObjectMgr(this, _client); + + // fill the auth data into the client's local field so that it can be requested by external + // entities + _client._authData = data; + + // wait for the bootstrap notification before we claim that we're actually logged on + } + + /** + * Callback called by the reader thread when it has parsed a new message from the socket and + * wishes to have it processed. + */ + protected void processMessage (Message msg) + { + // post this message to the dobjmgr queue + _omgr.processMessage(msg); + } + + protected void notifyClientObservers (ObserverOps.Session op) + { + _client.notifyObservers(op); + } + + protected void clientCleanup (Exception logonError) + { + _client.cleanup(logonError); + _client = null; // prevent any post-cleanup tomfoolery + } protected Client _client; + protected ClientDObjectMgr _omgr; + protected long _lastWrite; } diff --git a/src/java/com/threerings/presents/server/net/MessageHandler.java b/src/java/com/threerings/presents/client/ObserverOps.java similarity index 51% rename from src/java/com/threerings/presents/server/net/MessageHandler.java rename to src/java/com/threerings/presents/client/ObserverOps.java index 738278df6..19d5fc41b 100644 --- a/src/java/com/threerings/presents/server/net/MessageHandler.java +++ b/src/java/com/threerings/presents/client/ObserverOps.java @@ -2,7 +2,7 @@ // $Id$ // // Narya library - tools for developing networked games -// Copyright (C) 2002-2007 Three Rings Design, Inc., All Rights Reserved +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved // http://www.threerings.net/code/narya/ // // This library is free software; you can redistribute it and/or modify it @@ -19,19 +19,31 @@ // License along with this library; if not, write to the Free Software // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -package com.threerings.presents.server.net; +package com.threerings.presents.client; -import com.threerings.presents.net.UpstreamMessage; +import com.samskivert.util.ObserverList; /** - * After the connection object has parsed an entire upstream message, it - * passes it on to its message handler. + * Used to notify session and client observers. */ -public interface MessageHandler +public class ObserverOps { - /** - * Called when a complete message has been parsed from incoming - * network data. - */ - void handleMessage (UpstreamMessage message); + public abstract static class Session implements ObserverList.ObserverOp + { + public boolean apply (SessionObserver obs) { + notify(obs); + return true; + } + protected abstract void notify (SessionObserver obs); + } + + public abstract static class Client extends Session + { + @Override public void notify (SessionObserver obs) { + if (obs instanceof ClientObserver) { + notify((ClientObserver)obs); + } + } + protected abstract void notify (ClientObserver obs); + } } diff --git a/src/java/com/threerings/presents/net/DownstreamMessage.java b/src/java/com/threerings/presents/net/DownstreamMessage.java index 6928e8390..b6415d452 100644 --- a/src/java/com/threerings/presents/net/DownstreamMessage.java +++ b/src/java/com/threerings/presents/net/DownstreamMessage.java @@ -22,16 +22,15 @@ package com.threerings.presents.net; /** - * This class encapsulates a message in the distributed object protocol - * that flows from the server to the client. Downstream messages include - * object subscription, event forwarding and session management. + * This class encapsulates a message in the distributed object protocol that flows from the server + * to the client. Downstream messages include object subscription, event forwarding and session + * management. */ public abstract class DownstreamMessage extends Message { /** - * The message id of the upstream message with which this downstream - * message is associated (or -1 if it is not associated with any - * upstream message). + * The message id of the upstream message with which this downstream message is associated (or + * -1 if it is not associated with any upstream message). */ public short messageId = -1; diff --git a/src/java/com/threerings/presents/net/Message.java b/src/java/com/threerings/presents/net/Message.java index 6699399e9..98d5eb985 100644 --- a/src/java/com/threerings/presents/net/Message.java +++ b/src/java/com/threerings/presents/net/Message.java @@ -28,6 +28,9 @@ import com.threerings.io.SimpleStreamableObject; */ public abstract class Message extends SimpleStreamableObject { + /** A timestamp indicating when this message was received from the network. */ + public transient long received; + /** * Sets the message transport parameters. For messages received over the network, these * describe the mode of transport over which the message was received. When sending messages, diff --git a/src/java/com/threerings/presents/net/UpstreamMessage.java b/src/java/com/threerings/presents/net/UpstreamMessage.java index 2e609fb0d..a02baf617 100644 --- a/src/java/com/threerings/presents/net/UpstreamMessage.java +++ b/src/java/com/threerings/presents/net/UpstreamMessage.java @@ -22,27 +22,22 @@ package com.threerings.presents.net; /** - * This class encapsulates a message in the distributed object protocol - * that flows from the client to the server. Upstream messages include - * object subscription, event forwarding and session management. + * This class encapsulates a message in the distributed object protocol that flows from the client + * to the server. Upstream messages include object subscription, event forwarding and session + * management. */ public abstract class UpstreamMessage extends Message { /** - * This is a unique (within the context of a reasonable period of - * time) identifier assigned to each upstream message. The message ids - * are used to correlate a downstream response message to the - * appropriate upstream request message. + * This is a unique (within the context of a reasonable period of time) identifier assigned to + * each upstream message. The message ids are used to correlate a downstream response message + * to the appropriate upstream request message. */ public short messageId; - /** A timestamp indicating when this upstream message was received. */ - public transient long received; - /** - * Each upstream message derived class must provide a zero argument - * constructor so that it can be unserialized when read from the - * network. + * Each upstream message derived class must provide a zero argument constructor so that it can + * be unserialized when read from the network. */ public UpstreamMessage () { @@ -59,8 +54,7 @@ public abstract class UpstreamMessage extends Message } /** - * Returns the next message id suitable for use by an upstream - * message. + * Returns the next message id suitable for use by an upstream message. */ protected static synchronized short nextMessageId () { @@ -69,8 +63,8 @@ public abstract class UpstreamMessage extends Message } /** - * This is used to generate monotonically increasing message ids on - * the client as new messages are generated. + * This is used to generate monotonically increasing message ids on the client as new messages + * are generated. */ protected static short _nextMessageId; } diff --git a/src/java/com/threerings/presents/peer/server/PeerManager.java b/src/java/com/threerings/presents/peer/server/PeerManager.java index 89e9eeb03..7a5fde4ad 100644 --- a/src/java/com/threerings/presents/peer/server/PeerManager.java +++ b/src/java/com/threerings/presents/peer/server/PeerManager.java @@ -861,7 +861,7 @@ public abstract class PeerManager PeerNode peer = _peers.get(record.nodeName); if (peer == null) { _peers.put(record.nodeName, peer = createPeerNode()); - peer.init(this, _omgr, record); + peer.init(this, _omgr, _conmgr, record); } peer.refresh(record); } diff --git a/src/java/com/threerings/presents/peer/server/PeerNode.java b/src/java/com/threerings/presents/peer/server/PeerNode.java index 55a1930e2..bcf798c39 100644 --- a/src/java/com/threerings/presents/peer/server/PeerNode.java +++ b/src/java/com/threerings/presents/peer/server/PeerNode.java @@ -27,7 +27,6 @@ import java.util.Date; import com.samskivert.util.ResultListenerList; -import com.threerings.presents.client.BlockingCommunicator; import com.threerings.presents.client.Client; import com.threerings.presents.client.ClientObserver; import com.threerings.presents.client.Communicator; @@ -41,6 +40,8 @@ import com.threerings.presents.peer.data.NodeObject; import com.threerings.presents.peer.net.PeerBootstrapData; import com.threerings.presents.peer.server.persist.NodeRecord; import com.threerings.presents.server.PresentsDObjectMgr; +import com.threerings.presents.server.net.ConnectionManager; +import com.threerings.presents.server.net.ServerCommunicator; import static com.threerings.presents.Log.log; @@ -53,14 +54,17 @@ public class PeerNode /** This peer's node object. */ public NodeObject nodeobj; - public void init (PeerManager peermgr, PresentsDObjectMgr omgr, NodeRecord record) + /** + * Initializes this peer node and creates its internal client. + */ + public void init (PeerManager peermgr, final PresentsDObjectMgr omgr, + final ConnectionManager conmgr, NodeRecord record) { _peermgr = peermgr; _omgr = omgr; _record = record; _client = new Client(null, _omgr) { - @Override - protected void convertFromRemote (DObject target, DEvent event) { + @Override protected void convertFromRemote (DObject target, DEvent event) { super.convertFromRemote(target, event); // rewrite the event's target oid using the oid currently configured on the // distributed object (we will have it mapped in our remote server's oid space, @@ -70,11 +74,8 @@ public class PeerNode // properly deal with it event.eventId = PeerNode.this._omgr.getNextEventId(true); } - @Override - protected Communicator createCommunicator () { - // TODO: make a custom communicator that uses the ClientManager NIO system to do - // its I/O instead of using two threads and blocking socket I/O - return new BlockingCommunicator(this); + @Override protected Communicator createCommunicator () { + return new ServerCommunicator(this, conmgr, omgr); } }; _client.addClientObserver(this); @@ -102,10 +103,8 @@ public class PeerNode public void refresh (NodeRecord record) { - // if the hostname of this node changed, kill our existing client connection and connect - // anew - if (!record.hostName.equals(_record.hostName) && - _client.isActive()) { + // if the hostname of this node changed, kill our existing client and connect anew + if (!record.hostName.equals(_record.hostName) && _client.isActive()) { _client.logoff(false); } @@ -120,8 +119,8 @@ public class PeerNode // if our client hasn't updated its record since we last tried to logon, then just // chill if ((_lastConnectStamp - _record.lastUpdated.getTime()) > STALE_INTERVAL) { - log.debug("Not reconnecting to stale client [record=" + _record + - ", lastTry=" + new Date(_lastConnectStamp) + "]."); + log.debug("Not reconnecting to stale client", "record", _record, + "lastTry", new Date(_lastConnectStamp)); return; } diff --git a/src/java/com/threerings/presents/server/PresentsSession.java b/src/java/com/threerings/presents/server/PresentsSession.java index fe54d70be..4b5440fd4 100644 --- a/src/java/com/threerings/presents/server/PresentsSession.java +++ b/src/java/com/threerings/presents/server/PresentsSession.java @@ -57,6 +57,7 @@ import com.threerings.presents.net.EventNotification; import com.threerings.presents.net.FailureResponse; import com.threerings.presents.net.ForwardEventRequest; import com.threerings.presents.net.LogoffRequest; +import com.threerings.presents.net.Message; import com.threerings.presents.net.ObjectResponse; import com.threerings.presents.net.PingRequest; import com.threerings.presents.net.PongResponse; @@ -65,10 +66,8 @@ import com.threerings.presents.net.ThrottleUpdatedMessage; import com.threerings.presents.net.UnsubscribeRequest; import com.threerings.presents.net.UnsubscribeResponse; import com.threerings.presents.net.UpdateThrottleMessage; -import com.threerings.presents.net.UpstreamMessage; import com.threerings.presents.server.net.Connection; import com.threerings.presents.server.net.ConnectionManager; -import com.threerings.presents.server.net.MessageHandler; import static com.threerings.presents.Log.log; @@ -84,7 +83,7 @@ import static com.threerings.presents.Log.log; * from the conmgr thread and therefore also need not be synchronized. */ public class PresentsSession - implements MessageHandler, ClientResolutionListener + implements Connection.MessageHandler, ClientResolutionListener { /** Used by {@link PresentsSession#setUsername} to report success or failure. */ public static interface UserChangeListener @@ -396,8 +395,8 @@ public class PresentsSession endSession(); } - // from interface MessageHandler - public void handleMessage (UpstreamMessage message) + // from interface Connection.MessageHandler + public void handleMessage (Message message) { _messagesIn++; // count 'em up! @@ -957,16 +956,16 @@ public class PresentsSession } /** - * Message dispatchers are used to dispatch each different type of upstream message. We can - * look the dispatcher up in a table and invoke it through an overloaded member which is faster - * (so we think) than doing a bunch of instanceofs. + * Message dispatchers are used to dispatch each different type of message. We can look the + * dispatcher up in a table and invoke it through an overloaded member which is faster (so we + * think) than doing a bunch of instanceofs. */ protected static interface MessageDispatcher { /** * Dispatch the supplied message for the specified client. */ - void dispatch (PresentsSession client, UpstreamMessage mge); + void dispatch (PresentsSession client, Message mge); } /** @@ -974,7 +973,7 @@ public class PresentsSession */ protected static class SubscribeDispatcher implements MessageDispatcher { - public void dispatch (PresentsSession client, UpstreamMessage msg) + public void dispatch (PresentsSession client, Message msg) { SubscribeRequest req = (SubscribeRequest)msg; // log.info("Subscribing [client=" + client + ", oid=" + req.getOid() + "]."); @@ -989,7 +988,7 @@ public class PresentsSession */ protected static class UnsubscribeDispatcher implements MessageDispatcher { - public void dispatch (PresentsSession client, UpstreamMessage msg) + public void dispatch (PresentsSession client, Message msg) { UnsubscribeRequest req = (UnsubscribeRequest)msg; int oid = req.getOid(); @@ -1009,7 +1008,7 @@ public class PresentsSession */ protected static class ForwardEventDispatcher implements MessageDispatcher { - public void dispatch (PresentsSession client, UpstreamMessage msg) + public void dispatch (PresentsSession client, Message msg) { ForwardEventRequest req = (ForwardEventRequest)msg; DEvent fevt = req.getEvent(); @@ -1036,7 +1035,7 @@ public class PresentsSession */ protected static class PingDispatcher implements MessageDispatcher { - public void dispatch (PresentsSession client, UpstreamMessage msg) + public void dispatch (PresentsSession client, Message msg) { // send a pong response using the transport with which the message was received PingRequest req = (PingRequest)msg; @@ -1049,7 +1048,7 @@ public class PresentsSession */ protected static class ThrottleUpdatedDispatcher implements MessageDispatcher { - public void dispatch (final PresentsSession client, UpstreamMessage msg) + public void dispatch (final PresentsSession client, Message msg) { log.debug("Client ACKed throttle update", "client", client); client.throttleUpdated(); @@ -1061,7 +1060,7 @@ public class PresentsSession */ protected static class LogoffDispatcher implements MessageDispatcher { - public void dispatch (final PresentsSession client, UpstreamMessage msg) + public void dispatch (final PresentsSession client, Message msg) { log.debug("Client requested logoff", "client", client); client.safeEndSession(); diff --git a/src/java/com/threerings/presents/server/net/AuthingConnection.java b/src/java/com/threerings/presents/server/net/AuthingConnection.java index a18c5faf9..dccd653e2 100644 --- a/src/java/com/threerings/presents/server/net/AuthingConnection.java +++ b/src/java/com/threerings/presents/server/net/AuthingConnection.java @@ -21,62 +21,37 @@ package com.threerings.presents.server.net; -import java.io.IOException; - -import java.nio.channels.SelectionKey; -import java.nio.channels.SocketChannel; - -import com.samskivert.util.StringUtil; - import com.threerings.presents.net.AuthRequest; import com.threerings.presents.net.AuthResponse; -import com.threerings.presents.net.UpstreamMessage; +import com.threerings.presents.net.Message; import static com.threerings.presents.Log.log; /** - * The authing connection manages the client connection until - * authentication has completed (for better or for worse). + * The authing connection manages the client connection until authentication has completed (for + * better or for worse). */ public class AuthingConnection extends Connection - implements MessageHandler { - /** - * Creates a new authing connection object that will manage the - * authentication process for the suppled client socket. - */ - public AuthingConnection (ConnectionManager cmgr, SelectionKey selkey, - SocketChannel channel) - throws IOException + public AuthingConnection () { - super(cmgr, selkey, channel, System.currentTimeMillis()); - - // we are our own message handler - setMessageHandler(this); + setMessageHandler(new MessageHandler() { + public void handleMessage (Message msg) { + try { + // keep a handle on our auth request + _authreq = (AuthRequest)msg; + // post ourselves for processing by the authmgr + _cmgr.authenticateConnection(AuthingConnection.this); + } catch (ClassCastException cce) { + log.warning("Received non-authreq message during authentication process", + "conn", AuthingConnection.this, "msg", msg); + } + } + }); } /** - * Called when a new message has arrived from the client. - */ - public void handleMessage (UpstreamMessage msg) - { - try { - // keep a handle on our auth request - _authreq = (AuthRequest)msg; - - // post ourselves for processing by the authmgr - _cmgr.authenticateConnection(this); - - } catch (ClassCastException cce) { - log.warning("Received non-authreq message during " + - "authentication process [conn=" + this + - ", msg=" + msg + "]."); - } - } - - /** - * Returns a reference to the auth request currently being processed - * by this authing connection. + * Returns a reference to the auth request currently being processed. */ public AuthRequest getAuthRequest () { @@ -84,8 +59,8 @@ public class AuthingConnection extends Connection } /** - * Returns the auth response delivered to the client (only valid after - * the auth request has been processed. + * Returns the auth response delivered to the client (only valid after the auth request has + * been processed. */ public AuthResponse getAuthResponse () { @@ -93,9 +68,8 @@ public class AuthingConnection extends Connection } /** - * Stores a reference to the auth response delivered to this - * connection. This is called by the auth manager after delivering the - * auth response to the client. + * Stores a reference to the auth response delivered to this connection. This is called by the + * auth manager after delivering the auth response to the client. */ public void setAuthResponse (AuthResponse authrsp) { @@ -105,8 +79,7 @@ public class AuthingConnection extends Connection @Override public String toString () { - return "[mode=AUTHING, addr=" + - StringUtil.toString(_channel.socket().getInetAddress()) + "]"; + return "[mode=AUTHING, addr=" + getInetAddress() + "]"; } protected AuthRequest _authreq; diff --git a/src/java/com/threerings/presents/server/net/Connection.java b/src/java/com/threerings/presents/server/net/Connection.java index e35eb444f..307c53bb1 100644 --- a/src/java/com/threerings/presents/server/net/Connection.java +++ b/src/java/com/threerings/presents/server/net/Connection.java @@ -43,34 +43,39 @@ import com.threerings.io.ObjectOutputStream; import com.threerings.io.UnreliableObjectInputStream; import com.threerings.io.UnreliableObjectOutputStream; -import com.threerings.presents.net.DownstreamMessage; +import com.threerings.presents.net.Message; import com.threerings.presents.net.PingRequest; -import com.threerings.presents.net.UpstreamMessage; import com.threerings.presents.util.DatagramSequencer; import static com.threerings.presents.Log.log; /** * The base connection class implements the net event handler interface and processes raw incoming - * network data into a stream of parsed UpstreamMessage objects. It also provides the - * means to send messages to the client and facilities for checking delinquency. + * network data into a stream of parsed {@link Message} objects. It also provides the means to send + * messages to the client and facilities for checking delinquency. */ -public abstract class Connection implements NetEventHandler +public class Connection implements NetEventHandler { + /** Used with {@link #setMessageHandler}. */ + public static interface MessageHandler { + /** Called when a complete message has been parsed from incoming network data. */ + void handleMessage (Message message); + } + + /** The key used by the NIO code to track this connection. */ + public SelectionKey selkey; + /** - * Constructs a connection object that is associated with the supplied socket. + * Initializes a connection object with a socket and related info. * * @param cmgr The connection manager with which this connection is associated. - * @param selkey the key used by the NIO code to track this connection. * @param channel The socket channel from which we'll be reading messages. * @param createStamp The time at which this connection was created. */ - public Connection (ConnectionManager cmgr, SelectionKey selkey, SocketChannel channel, - long createStamp) + public void init (ConnectionManager cmgr, SocketChannel channel, long createStamp) throws IOException { _cmgr = cmgr; - _selkey = selkey; _channel = channel; _lastEvent = createStamp; _connectionId = ++_lastConnectionId; @@ -104,14 +109,6 @@ public abstract class Connection implements NetEventHandler return _connectionId; } - /** - * Returns the selection key associated with our socket channel. - */ - public SelectionKey getSelectionKey () - { - return _selkey; - } - /** * Returns the non-blocking socket object used to construct this connection. */ @@ -178,10 +175,39 @@ public abstract class Connection implements NetEventHandler } /** - * Called when there is a failure reading or writing on this connection. We notify the + * Queues up a request to have this connection closed by the connection manager once all + * messages in its queue have been written to its target. + */ + public void asyncClose () + { + _cmgr.postAsyncClose(this); + } + + /** + * Posts a message for delivery to this connection. The message will be delivered by the conmgr + * thread as soon as it gets to it. + */ + public void postMessage (Message msg) + { + // pass this along to the connection manager + _cmgr.postMessage(this, msg); + } + + /** + * Called when an outgoing socket experiences a connect failure. The connection manager will + * have cleaned up the partial registration needed during the connect process, so we are only + * responsible for closing our socket. + */ + public void connectFailure (IOException ioe) + { + closeSocket(); + } + + /** + * Called when there is a failure reading or writing to this connection. We notify the * connection manager and close ourselves down. */ - public void handleFailure (IOException ioe) + public void networkFailure (IOException ioe) { // if we're already closed, then something is seriously funny if (isClosed()) { @@ -196,6 +222,135 @@ public abstract class Connection implements NetEventHandler closeSocket(); } + /** + * Processes a datagram sent to this connection. + */ + public void handleDatagram (InetSocketAddress source, ByteBuffer buf, long when) + { + // lazily create our various bits and bobs + if (_digest == null) { + try { + _digest = MessageDigest.getInstance("MD5"); + } catch (NoSuchAlgorithmException nsae) { + log.warning("Missing MD5 algorithm."); + return; + } + ByteBufferInputStream bin = new ByteBufferInputStream(buf); + _sequencer = new DatagramSequencer( + new UnreliableObjectInputStream(bin), + new UnreliableObjectOutputStream(_cmgr.getFlattener())); + } + + // verify the hash + buf.position(12); + _digest.update(buf); + byte[] hash = _digest.digest(_datagramSecret); + buf.position(4); + for (int ii = 0; ii < 8; ii++) { + if (hash[ii] != buf.get()) { + log.warning("Datagram failed hash check [connectionId=" + _connectionId + + ", source=" + source + "]."); + return; + } + } + + // update our target address + _datagramAddress = source; + + // read the contents through the sequencer + try { + Message msg = _sequencer.readDatagram(); + if (msg == null) { + return; // received out of order + } + msg.received = when; + _handler.handleMessage(msg); + + } catch (ClassNotFoundException cnfe) { + log.warning("Error reading datagram [error=" + cnfe + "]."); + + } catch (IOException ioe) { + log.warning("Error reading datagram [error=" + ioe + "]."); + } + } + + // from interface NetEventHandler + public int handleEvent (long when) + { + // make a note that we received an event as of this time + _lastEvent = when; + + int bytesIn = 0; + try { + // we're lazy about creating our input streams because we may be inheriting them from + // our authing connection and we don't want to unnecessarily create them in that case + if (_fin == null) { + _fin = new FramedInputStream(); + _oin = new ObjectInputStream(_fin); + if (_loader != null) { + _oin.setClassLoader(_loader); + } + } + + // there may be more than one frame in the buffer, so we keep reading them until we run + // out of data + while (_fin.readFrame(_channel)) { + // make a note of how many bytes are in this frame (including the frame length + // bytes which aren't reported in available()) + bytesIn = _fin.available() + 4; + // parse the message and pass it on + Message msg = (Message)_oin.readObject(); + msg.received = when; +// Log.info("Read message " + msg + "."); + _handler.handleMessage(msg); + } + + } catch (EOFException eofe) { + // close down the socket gracefully + close(); + + } catch (ClassNotFoundException cnfe) { + log.warning("Error reading message from socket [channel=" + + StringUtil.safeToString(_channel) + ", error=" + cnfe + "]."); + // deal with the failure + String errmsg = "Unable to decode incoming message."; + networkFailure((IOException) new IOException(errmsg).initCause(cnfe)); + + } catch (IOException ioe) { + // don't log a warning for the ever-popular "the client dropped the connection" failure + String msg = ioe.getMessage(); + if (msg == null || msg.indexOf("reset by peer") == -1) { + log.warning("Error reading message from socket [channel=" + + StringUtil.safeToString(_channel) + ", error=" + ioe + "]."); + } + // deal with the failure + networkFailure(ioe); + } + + return bytesIn; + } + + // documentation inherited from interface + public boolean checkIdle (long now) + { + long idleMillis = now - _lastEvent; + if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) { + return false; + } + if (isClosed()) { + return true; + } + log.info("Disconnecting non-communicative client [conn=" + this + + ", idle=" + idleMillis + "ms]."); + return true; + } + + @Override // from Object + public String toString () + { + return "[id=" + (hashCode() % 1000) + ", addr=" + getInetAddress() + "]"; + } + /** * Returns the object input stream associated with this connection. This should only be used * by the connection manager. @@ -270,147 +425,10 @@ public abstract class Connection implements NetEventHandler } // clear out our references to prevent repeat closings - _selkey = null; _channel = null; } - /** - * Called when our client socket has data available for reading. - */ - public int handleEvent (long when) - { - // make a note that we received an event as of this time - _lastEvent = when; - - int bytesIn = 0; - try { - // we're lazy about creating our input streams because we may be inheriting them from - // our authing connection and we don't want to unnecessarily create them in that case - if (_fin == null) { - _fin = new FramedInputStream(); - _oin = new ObjectInputStream(_fin); - if (_loader != null) { - _oin.setClassLoader(_loader); - } - } - - // there may be more than one frame in the buffer, so we keep reading them until we run - // out of data - while (_fin.readFrame(_channel)) { - // make a note of how many bytes are in this frame (including the frame length - // bytes which aren't reported in available()) - bytesIn = _fin.available() + 4; - // parse the message and pass it on - UpstreamMessage msg = (UpstreamMessage)_oin.readObject(); - msg.received = when; -// Log.info("Read message " + msg + "."); - _handler.handleMessage(msg); - } - - } catch (EOFException eofe) { - // close down the socket gracefully - close(); - - } catch (ClassNotFoundException cnfe) { - log.warning("Error reading message from socket [channel=" + - StringUtil.safeToString(_channel) + ", error=" + cnfe + "]."); - // deal with the failure - String errmsg = "Unable to decode incoming message."; - handleFailure((IOException) new IOException(errmsg).initCause(cnfe)); - - } catch (IOException ioe) { - // don't log a warning for the ever-popular "the client dropped the connection" failure - String msg = ioe.getMessage(); - if (msg == null || msg.indexOf("reset by peer") == -1) { - log.warning("Error reading message from socket [channel=" + - StringUtil.safeToString(_channel) + ", error=" + ioe + "]."); - } - // deal with the failure - handleFailure(ioe); - } - - return bytesIn; - } - - /** - * Processes a datagram sent to this connection. - */ - public void handleDatagram (InetSocketAddress source, ByteBuffer buf, long when) - { - // lazily create our various bits and bobs - if (_digest == null) { - try { - _digest = MessageDigest.getInstance("MD5"); - } catch (NoSuchAlgorithmException nsae) { - log.warning("Missing MD5 algorithm."); - return; - } - ByteBufferInputStream bin = new ByteBufferInputStream(buf); - _sequencer = new DatagramSequencer( - new UnreliableObjectInputStream(bin), - new UnreliableObjectOutputStream(_cmgr.getFlattener())); - } - - // verify the hash - buf.position(12); - _digest.update(buf); - byte[] hash = _digest.digest(_datagramSecret); - buf.position(4); - for (int ii = 0; ii < 8; ii++) { - if (hash[ii] != buf.get()) { - log.warning("Datagram failed hash check [connectionId=" + _connectionId + - ", source=" + source + "]."); - return; - } - } - - // update our target address - _datagramAddress = source; - - // read the contents through the sequencer - try { - UpstreamMessage msg = (UpstreamMessage)_sequencer.readDatagram(); - if (msg == null) { - return; // received out of order - } - msg.received = when; - _handler.handleMessage(msg); - - } catch (ClassNotFoundException cnfe) { - log.warning("Error reading datagram [error=" + cnfe + "]."); - - } catch (IOException ioe) { - log.warning("Error reading datagram [error=" + ioe + "]."); - } - } - - // documentation inherited from interface - public boolean checkIdle (long now) - { - long idleMillis = now - _lastEvent; - if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) { - return false; - } - if (isClosed()) { - return true; - } - log.info("Disconnecting non-communicative client [conn=" + this + - ", idle=" + idleMillis + "ms]."); - return true; - } - - /** - * Posts a downstream message for delivery to this connection. The message will be delivered by - * the conmgr thread as soon as it gets to it. - */ - public void postMessage (DownstreamMessage msg) - { - // pass this along to the connection manager - _cmgr.postMessage(this, msg); - } - protected ConnectionManager _cmgr; - protected SelectionKey _selkey; protected SocketChannel _channel; protected long _lastEvent; diff --git a/src/java/com/threerings/presents/server/net/ConnectionManager.java b/src/java/com/threerings/presents/server/net/ConnectionManager.java index e039f0cb8..6f8be8824 100644 --- a/src/java/com/threerings/presents/server/net/ConnectionManager.java +++ b/src/java/com/threerings/presents/server/net/ConnectionManager.java @@ -62,7 +62,7 @@ import com.threerings.presents.client.Client; import com.threerings.presents.data.ConMgrStats; import com.threerings.presents.net.AuthRequest; import com.threerings.presents.net.AuthResponse; -import com.threerings.presents.net.DownstreamMessage; +import com.threerings.presents.net.Message; import com.threerings.presents.server.Authenticator; import com.threerings.presents.server.ChainedAuthenticator; import com.threerings.presents.server.DummyAuthenticator; @@ -193,6 +193,74 @@ public class ConnectionManager extends LoopingThread } } + /** + * Opens an outgoing connection to the supplied address. The connection will be opened in a + * non-blocking manner and added to the connection manager's select set. Messages posted to the + * connection prior to it being actually connected to its destination will remain in the queue. + * If the connection fails those messages will be dropped. + * + * @param conn the connection to be initialized and opened. Callers may want to provide a + * {@link Connection} derived class so that they may intercept calldown methods. + * @param hostname the hostname of the server to which to connect. + * @param port the port on which to connect to the server. + * + * @exception IOException thrown if an error occurs opening a connection to the supplied + * server. When this method returns the connection may or may not be actually connected to the + * server. If the asynchronous connection attempt fails, the Connection will be notified via + * {@link Connection#networkFailure}. + */ + public void openOutgoingConnection (final Connection conn, String hostname, int port) + throws IOException + { + // create a non-blocking socket channel to use for this connection + final SocketChannel sockchan = SocketChannel.open(); + sockchan.configureBlocking(false); + + // create our connection instance + conn.init(this, sockchan, System.currentTimeMillis()); + // and register our channel with the selector (if this fails, we abandon ship immediately) + conn.selkey = sockchan.register(_selector, SelectionKey.OP_CONNECT); + + // start our connection process (now if we fail we need to clean things up) + try { + NetEventHandler handler; + if (sockchan.connect(new InetSocketAddress(hostname, port))) { + // it is possible even for a non-blocking socket to connect immediately, in which + // case we stick the connection in as its event handler immediately + handler = conn; + + } else { + // 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 + handler = new NetEventHandler() { + 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); + } + } catch (IOException ioe) { + _handlers.remove(conn.selkey); + _oflowqs.remove(conn); + conn.connectFailure(ioe); + } + return 0; + } + public boolean checkIdle (long now) { + return conn.checkIdle(now); + } + }; + } + _handlers.put(conn.selkey, handler); + + } catch (IOException ioe) { + log.warning("Failed to initiate connection for " + sockchan + ".", ioe); + conn.connectFailure(ioe); // nothing else to clean up + throw ioe; + } + } + /** * Queues a connection up to be closed on the conmgr thread. */ @@ -201,25 +269,8 @@ public class ConnectionManager extends LoopingThread _deathq.append(conn); } - /** - * Performs the authentication process on the specified connection. This is called by {@link - * AuthingConnection} itself once it receives its auth request. - */ - public void authenticateConnection (AuthingConnection conn) - { - _author.authenticateConnection(_authInvoker, conn, new ResultListener() { - public void requestCompleted (AuthingConnection conn) { - _authq.append(conn); - } - public void requestFailed (Exception cause) { - // this never happens - } - }); - } - - // documentation inherited from interface ReportManager.Reporter - public void appendReport ( - StringBuilder report, long now, long sinceLast, boolean reset) + // from interface ReportManager.Reporter + public void appendReport (StringBuilder report, long now, long sinceLast, boolean reset) { ConMgrStats stats = getStats(); int connects = stats.connects - _lastStats.connects; @@ -263,6 +314,22 @@ public class ConnectionManager extends LoopingThread return super.isRunning() || _omgr.isRunning(); } + /** + * Performs the authentication process on the specified connection. This is called by {@link + * AuthingConnection} itself once it receives its auth request. + */ + protected void authenticateConnection (AuthingConnection conn) + { + _author.authenticateConnection(_authInvoker, conn, new ResultListener() { + public void requestCompleted (AuthingConnection conn) { + _authq.append(conn); + } + public void requestFailed (Exception cause) { + // this never happens + } + }); + } + /** * Notifies the connection observers of a connection event. Used internally. */ @@ -363,10 +430,6 @@ public class ConnectionManager extends LoopingThread } } - // we'll use these for sending messages to clients - _framer = new FramingOutputStream(); - _flattener = new ByteArrayOutputStream(); - // notify our startup listener, if we have one if (_startlist != null) { _startlist.requestCompleted(null); @@ -461,21 +524,20 @@ public class ConnectionManager extends LoopingThread try { // construct a new running connection to handle this connections network traffic // from here on out - SelectionKey selkey = conn.getSelectionKey(); - RunningConnection rconn = new RunningConnection( - this, selkey, conn.getChannel(), iterStamp); + Connection rconn = new Connection(); + rconn.init(this, conn.getChannel(), iterStamp); + rconn.selkey = conn.selkey; // we need to keep using the same object input and output streams from the // beginning of the session because they have context that needs to be preserved rconn.inheritStreams(conn); // replace the mapping in the handlers table from the old conn with the new one - _handlers.put(selkey, rconn); + _handlers.put(rconn.selkey, rconn); // add a mapping for the connection id and set the datagram secret _connections.put(rconn.getConnectionId(), rconn); - rconn.setDatagramSecret( - conn.getAuthRequest().getCredentials().getDatagramSecret()); + rconn.setDatagramSecret(conn.getAuthRequest().getCredentials().getDatagramSecret()); // transfer any overflow queue for that connection OverflowQueue oflowHandler = _oflowqs.remove(conn); @@ -494,9 +556,10 @@ public class ConnectionManager extends LoopingThread Set ready = null; try { - // check for incoming network events // log.debug("Selecting from " + StringUtil.toString(_selector.keys()) + " (" + // SELECT_LOOP_TIME + ")."); + + // check for incoming network events int ecount = _selector.select(SELECT_LOOP_TIME); ready = _selector.selectedKeys(); if (ecount == 0) { @@ -538,8 +601,7 @@ public class ConnectionManager extends LoopingThread try { handler = _handlers.get(selkey); if (handler == null) { - log.warning("Received network event but have no registered handler " + - "[selkey=" + selkey + "]."); + log.warning("Received network event for unknown handler", "key", selkey); // request that this key be removed from our selection set, which normally // happens automatically but for some reason didn't selkey.cancel(); @@ -592,7 +654,7 @@ public class ConnectionManager extends LoopingThread } } catch (IOException ioe) { - oq.conn.handleFailure(ioe); + oq.conn.networkFailure(ioe); } } } @@ -641,10 +703,15 @@ public class ConnectionManager extends LoopingThread return true; } + // if this is an asynchronous close request, queue the connection up for death + if (data == ASYNC_CLOSE_REQUEST) { + closeConnection(conn); + 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 + "]."); + log.warning("Refusing to write very large message", "conn", conn, "size", data.length); return true; } @@ -664,26 +731,26 @@ public class ConnectionManager extends LoopingThread _outbuf.put(data); _outbuf.flip(); + // if the connection to which we're writing is not yet ready, the whole message is + // "leftover", so we pass it to the partial write handler + SocketChannel sochan = conn.getChannel(); + if (!sochan.isConnected()) { + pwh.handlePartialWrite(conn, _outbuf); + return false; + } + // then write the data to the socket - int wrote = conn.getChannel().write(_outbuf); + int wrote = sochan.write(_outbuf); noteWrite(1, wrote); + // if we didn't write our entire message, deal with the leftover bytes if (_outbuf.remaining() > 0) { fully = false; -// log.info("Partial write [conn=" + conn + -// ", msg=" + StringUtil.shortClassName(outmsg) + ", wrote=" + wrote + -// ", size=" + buffer.limit() + "]."); pwh.handlePartialWrite(conn, _outbuf); - -// } else if (wrote > 10000) { -// log.info("Big write [conn=" + conn + -// ", msg=" + StringUtil.shortClassName(outmsg) + -// ", wrote=" + wrote + "]."); } } catch (IOException ioe) { - // instruct the connection to deal with its failure - conn.handleFailure(ioe); + conn.networkFailure(ioe); // instruct the connection to deal with its failure } finally { _outbuf.clear(); @@ -790,8 +857,10 @@ public class ConnectionManager extends LoopingThread // connection and register it with our selection set SelectableChannel selchan = channel; selchan.configureBlocking(false); - SelectionKey selkey = selchan.register(_selector, SelectionKey.OP_READ); - _handlers.put(selkey, new AuthingConnection(this, selkey, channel)); + AuthingConnection aconn = new AuthingConnection(); + aconn.selkey = selchan.register(_selector, SelectionKey.OP_READ); + aconn.init(this, channel, System.currentTimeMillis()); + _handlers.put(aconn.selkey, aconn); synchronized (this) { _stats.connects++; } @@ -863,11 +932,11 @@ public class ConnectionManager extends LoopingThread * which happens when forwarding an event to a client and at the completion of authentication, * both of which must happen only on the distributed object thread. */ - void postMessage (Connection conn, DownstreamMessage msg) + void postMessage (Connection conn, Message msg) { if (!isRunning()) { - log.warning( - "Posting message to inactive connection manager", "msg", msg, new Exception()); + log.warning("Posting message to inactive connection manager", + "msg", msg, new Exception()); } // sanity check @@ -901,22 +970,20 @@ public class ConnectionManager extends LoopingThread ByteBuffer buffer = _framer.frameAndReturnBuffer(); byte[] data = new byte[buffer.limit()]; buffer.get(data); - -// log.info("Flattened " + msg + " into " + data.length + " bytes."); + // log.info("Flattened " + msg + " into " + data.length + " bytes."); // and slap both on the queue - _outq.append(new Tuple(conn, data)); + _outq.append(Tuple.create(conn, data)); } catch (Exception e) { - log.warning("Failure flattening message [conn=" + conn + - ", msg=" + msg + "].", e); + log.warning("Failure flattening message", "conn", conn, "msg", msg, e); } } /** * Helper function for {@link #postMessage}; handles posting the message as a datagram. */ - void postDatagram (Connection conn, DownstreamMessage msg) + void postDatagram (Connection conn, Message msg) throws Exception { _flattener.reset(); @@ -929,7 +996,20 @@ public class ConnectionManager extends LoopingThread byte[] data = _flattener.toByteArray(); // slap it on the queue - _dataq.append(new Tuple(conn, data)); + _dataq.append(Tuple.create(conn, data)); + } + + /** + * Posts a fake message to this connection's outgoing message queue that will cause the + * connection to be closed when this message is reached. This is only used by outgoing + * connections to ensure that they finish sending their queued outgoing messages before closing + * their connection. Incoming connections tend only to be closed at the request of the client + * or in case of delinquincy. In neither circumstance do we need to flush the client's outgoing + * queue before closing. + */ + void postAsyncClose (Connection conn) + { + _outq.append(Tuple.create(conn, ASYNC_CLOSE_REQUEST)); } /** @@ -939,7 +1019,7 @@ public class ConnectionManager extends LoopingThread { // remove this connection from our mappings (it is automatically removed from the Selector // when the socket is closed) - _handlers.remove(conn.getSelectionKey()); + _handlers.remove(conn.selkey); _connections.remove(conn.getConnectionId()); _oflowqs.remove(conn); synchronized (this) { @@ -957,7 +1037,7 @@ public class ConnectionManager extends LoopingThread { // remove this connection from our mappings (it is automatically removed from the Selector // when the socket is closed) - _handlers.remove(conn.getSelectionKey()); + _handlers.remove(conn.selkey); _connections.remove(conn.getConnectionId()); _oflowqs.remove(conn); @@ -1012,8 +1092,14 @@ public class ConnectionManager extends LoopingThread { // write any partial message if we have one if (_partial != null) { + // if our outgoing channel is still not ready, then bail immediately + SocketChannel sochan = conn.getChannel(); + if (!sochan.isConnected()) { + return false; + } + // write all we can of our partial buffer - int wrote = conn.getChannel().write(_partial); + int wrote = sochan.write(_partial); noteWrite(0, wrote); if (_partial.remaining() == 0) { @@ -1095,8 +1181,8 @@ public class ConnectionManager extends LoopingThread protected Queue> _outq = new Queue>(); protected Queue> _dataq = new Queue>(); - protected FramingOutputStream _framer; - protected ByteArrayOutputStream _flattener; + protected FramingOutputStream _framer = new FramingOutputStream(); + protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream(); protected ByteBuffer _outbuf = ByteBuffer.allocateDirect(64 * 1024); protected ByteBuffer _databuf = ByteBuffer.allocateDirect(Client.MAX_DATAGRAM_SIZE); @@ -1135,4 +1221,7 @@ public class ConnectionManager extends LoopingThread protected static final int CONNECTION_ESTABLISHED = 0; protected static final int CONNECTION_FAILED = 1; protected static final int CONNECTION_CLOSED = 2; + + /** Used to denote asynchronous close requests. */ + protected static final byte[] ASYNC_CLOSE_REQUEST = new byte[0]; } diff --git a/src/java/com/threerings/presents/server/net/RunningConnection.java b/src/java/com/threerings/presents/server/net/RunningConnection.java deleted file mode 100644 index d1ab8b37a..000000000 --- a/src/java/com/threerings/presents/server/net/RunningConnection.java +++ /dev/null @@ -1,61 +0,0 @@ -// -// $Id$ -// -// Narya library - tools for developing networked games -// Copyright (C) 2002-2007 Three Rings Design, Inc., All Rights Reserved -// http://www.threerings.net/code/narya/ -// -// This library is free software; you can redistribute it and/or modify it -// under the terms of the GNU Lesser General Public License as published -// by the Free Software Foundation; either version 2.1 of the License, or -// (at your option) any later version. -// -// This library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -// Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public -// License along with this library; if not, write to the Free Software -// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - -package com.threerings.presents.server.net; - -import java.io.IOException; - -import java.nio.channels.SelectionKey; -import java.nio.channels.SocketChannel; - -import com.threerings.presents.net.UpstreamMessage; - -/** - * A running connection object represents a fully operational client - * connection to the server. - */ -public class RunningConnection extends Connection -{ - /** - * Constructs a new running connection object to manage the supplied - * client socket. - */ - public RunningConnection (ConnectionManager cmgr, SelectionKey selkey, - SocketChannel channel, long createStamp) - throws IOException - { - super(cmgr, selkey, channel, createStamp); - } - - /** - * Called when a new message has arrived from the client. - */ - public void handleMessage (UpstreamMessage msg) - { - } - - @Override - public String toString () - { - return "[mode=RUNNING, id=" + (hashCode() % 1000) + - ", addr=" + getInetAddress() + "]"; - } -} diff --git a/src/java/com/threerings/presents/server/net/ServerCommunicator.java b/src/java/com/threerings/presents/server/net/ServerCommunicator.java new file mode 100644 index 000000000..5c54b1550 --- /dev/null +++ b/src/java/com/threerings/presents/server/net/ServerCommunicator.java @@ -0,0 +1,196 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/narya/ +// +// This library is free software; you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published +// by the Free Software Foundation; either version 2.1 of the License, or +// (at your option) any later version. +// +// This library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +// Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public +// License along with this library; if not, write to the Free Software +// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + +package com.threerings.presents.server.net; + +import java.io.IOException; + +import com.threerings.presents.client.Client; +import com.threerings.presents.client.ClientDObjectMgr; +import com.threerings.presents.client.ClientObserver; +import com.threerings.presents.client.Communicator; +import com.threerings.presents.client.ObserverOps; +import com.threerings.presents.client.SessionObserver; +import com.threerings.presents.dobj.RootDObjectManager; +import com.threerings.presents.net.AuthRequest; +import com.threerings.presents.net.AuthResponse; +import com.threerings.presents.net.AuthResponseData; +import com.threerings.presents.net.LogoffRequest; +import com.threerings.presents.net.Message; +import com.threerings.presents.net.UpstreamMessage; + +import static com.threerings.presents.Log.log; + +/** + * Provides Client {@link Communicator} services using non-blocking I/O and the connection manager. + */ +public class ServerCommunicator extends Communicator +{ + public ServerCommunicator (Client client, ConnectionManager conmgr, RootDObjectManager rootmgr) + { + super(client); + _conmgr = conmgr; + _rootmgr = rootmgr; + _omgr = new ClientDObjectMgr(this, client); + } + + @Override // from Communicator + public void logon () + { + // make sure things are copacetic + if (_conn != null) { + throw new RuntimeException("Communicator already started."); + } + + // we assume server entities have no firewall issues and can connect on the first port + try { + Connection conn = new Connection() { + @Override public void connectFailure (IOException ioe) { + _logonError = ioe; // report this as a logon failure + super.connectFailure(ioe); + } + + @Override public void networkFailure (final IOException ioe) { + notifyClientObservers(new ObserverOps.Client() { + protected void notify (ClientObserver obs) { + obs.clientConnectionFailed(_client, ioe); + } + }); + super.networkFailure(ioe); + } + + @Override protected void closeSocket () { + super.closeSocket(); + shutdown(); + } + }; + conn.setMessageHandler(new Connection.MessageHandler() { + public void handleMessage (Message message) { + try { + // our first message will always be an auth response + gotAuthResponse((AuthResponse)message); + } catch (Exception e) { + _logonError = e; + shutdown(); + } + } + }); + _conmgr.openOutgoingConnection(conn, _client.getHostname(), _client.getPorts()[0]); + _conn = conn; + if (_loader != null) { + _conn.setClassLoader(_loader); + } + + // now send our auth request + postMessage(new AuthRequest(_client.getCredentials(), _client.getVersion(), + _client.getBootGroups())); + + } catch (IOException ioe) { + _logonError = ioe; + shutdown(); + } + } + + @Override // from Communicator + public void logoff () + { + if (_conn != null) { + _conn.postMessage(new LogoffRequest()); + _conn.asyncClose(); + _conn = null; + } + } + + @Override // from Communicator + public void gotBootstrap () + { + // nothing needed + } + + @Override // from Communicator + public void postMessage (final UpstreamMessage msg) + { + // if we're not on the main dobjmgr thread, we need to get there + if (!_rootmgr.isDispatchThread()) { + _rootmgr.postRunnable(new Runnable() { + public void run () { + postMessage(msg); + } + }); + return; + } + + if (_conn == null) { + log.info("Dropping message for lack of connection.", "client", _client, "msg", msg); + return; + } + + // pass this message along to our connection + _conn.postMessage(msg); + + // we cheat a bit and claim that we "wrote" when we post our messages so that we don't have + // to modify the connection manager to call a method on Connection every time it writes a + // message from the queue + updateWriteStamp(); + } + + @Override // from Communicator + public void setClassLoader (ClassLoader loader) + { + _loader = loader; + if (_conn != null) { + _conn.setClassLoader(loader); + } + } + + @Override // from Communicator + protected synchronized void logonSucceeded (AuthResponseData data) + { + super.logonSucceeded(data); + + // now we can route all messages to the ClientDObjectMgr + _conn.setMessageHandler(new Connection.MessageHandler() { + public void handleMessage (Message message) { + _omgr.processMessage(message); + } + }); + } + + protected void shutdown () + { + if (_logonError == null) { + // we were logged on successfully, so report didLogoff first + notifyClientObservers(new ObserverOps.Session() { + protected void notify (SessionObserver obs) { + obs.clientDidLogoff(_client); + } + }); + } + clientCleanup(_logonError); + } + + protected ConnectionManager _conmgr; + protected RootDObjectManager _rootmgr; + protected Connection _conn; + protected ClientDObjectMgr _omgr; + protected ClassLoader _loader; + protected Exception _logonError; +}