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

_usermap is referenced by both the conmgr and dobjmgr threads, so access to it

Browse Source 
must be synchronized.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4681 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2007-05-04 02:25:03 +00:00
parent 69485a88ee
commit 4d455eb6e6
3 changed files with 123 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/Log.java
+1 -2
View File
@@ -30,8 +30,7 @@ import java.util.logging.Logger;
public class Log public class Log
{ {
/** We dispatch our log messages through this logger. */ /** We dispatch our log messages through this logger. */
public static Logger log = public static Logger log = Logger.getLogger("com.threerings.narya.presents");
Logger.getLogger("com.threerings.narya.presents");
/** Convenience function. */ /** Convenience function. */
public static void debug (String message) public static void debug (String message)
src/java/com/threerings/presents/server/ClientManager.java
+58 -49
View File
@@ -26,6 +26,7 @@ import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
import java.util.Iterator; import java.util.Iterator;
import java.util.logging.Level;
import com.samskivert.util.Interval; import com.samskivert.util.Interval;
import com.samskivert.util.ObserverList; import com.samskivert.util.ObserverList;
@@ -33,13 +34,14 @@ import com.samskivert.util.StringUtil;
import com.threerings.util.Name; import com.threerings.util.Name;
import com.threerings.presents.Log;
import com.threerings.presents.data.ClientObject; import com.threerings.presents.data.ClientObject;
import com.threerings.presents.net.AuthRequest; import com.threerings.presents.net.AuthRequest;
import com.threerings.presents.net.AuthResponse; import com.threerings.presents.net.AuthResponse;
import com.threerings.presents.net.Credentials; import com.threerings.presents.net.Credentials;
import com.threerings.presents.server.net.*; import com.threerings.presents.server.net.*;
import static com.threerings.presents.Log.log;
/** /**
* The client manager is responsible for managing the clients (surprise, surprise) which are * The client manager is responsible for managing the clients (surprise, surprise) which are
* slightly more than just connections. Clients persist in the absence of connections in case a * slightly more than just connections. Clients persist in the absence of connections in case a
@@ -110,17 +112,17 @@ public class ClientManager
// documentation inherited from interface // documentation inherited from interface
public void shutdown () public void shutdown ()
{ {
Log.info("Client manager shutting down [ccount=" + _usermap.size() + "]."); log.info("Client manager shutting down [ccount=" + _usermap.size() + "].");
// inform all of our clients that they are being shut down // inform all of our clients that they are being shut down
for (Iterator iter = _usermap.values().iterator(); iter.hasNext(); ) { synchronized (_usermap) {
PresentsClient pc = (PresentsClient)iter.next(); for (PresentsClient pc : _usermap.values()) {
try { try {
pc.shutdown(); pc.shutdown();
} catch (Exception e) { } catch (Exception e) {
Log.warning( log.log(Level.WARNING, "Client choked in shutdown() [client=" +
"Client choked in shutdown() [client=" + StringUtil.safeToString(pc) + "]."); StringUtil.safeToString(pc) + "].", e);
Log.logStackTrace(e); }
} }
} }
} }
@@ -155,7 +157,9 @@ public class ClientManager
*/ */
public int getClientCount () public int getClientCount ()
{ {
return _usermap.size(); synchronized (_usermap) {
return _usermap.size();
}
} }
/** /**
@@ -196,7 +200,9 @@ public class ClientManager
*/ */
public PresentsClient getClient (Name authUsername) public PresentsClient getClient (Name authUsername)
{ {
return _usermap.get(authUsername); synchronized (_usermap) {
return _usermap.get(authUsername);
}
} }
/** /**
@@ -269,7 +275,7 @@ public class ClientManager
{ {
ClientObject clobj = _objmap.get(username); ClientObject clobj = _objmap.get(username);
if (clobj == null) { if (clobj == null) {
Log.warning("Requested to release unmapped client object [username=" + username + "]."); log.warning("Requested to release unmapped client object [username=" + username + "].");
Thread.dumpStack(); Thread.dumpStack();
return; return;
} }
@@ -279,7 +285,7 @@ public class ClientManager
return; return;
} }
Log.debug("Destroying client " + clobj.who() + "."); log.fine("Destroying client " + clobj.who() + ".");
// we're all clear to go; remove the mapping // we're all clear to go; remove the mapping
_objmap.remove(username); _objmap.remove(username);
@@ -311,7 +317,7 @@ public class ClientManager
_penders.remove(username); _penders.remove(username);
} }
// documentation inherited // from interface ConnectionObserver
public synchronized void connectionEstablished ( public synchronized void connectionEstablished (
Connection conn, AuthRequest req, AuthResponse rsp) Connection conn, AuthRequest req, AuthResponse rsp)
{ {
@@ -322,30 +328,32 @@ public class ClientManager
PresentsClient client = getClient(username); PresentsClient client = getClient(username);
if (client != null) { if (client != null) {
Log.info("Resuming session [username=" + username + ", conn=" + conn + "]."); log.info("Resuming session [username=" + username + ", conn=" + conn + "].");
client.resumeSession(req, conn); client.resumeSession(req, conn);
} else { } else {
Log.info("Session initiated [username=" + username + ", conn=" + conn + "]."); log.info("Session initiated [username=" + username + ", conn=" + conn + "].");
// create a new client and stick'em in the table // create a new client and stick'em in the table
client = _factory.createClient(req); client = _factory.createClient(req);
client.startSession(this, req, conn, rsp.authdata); client.startSession(this, req, conn, rsp.authdata);
// map their client instance // map their client instance
_usermap.put(username, client); synchronized (_usermap) {
_usermap.put(username, client);
}
} }
// map this connection to this client // map this connection to this client
_conmap.put(conn, client); _conmap.put(conn, client);
} }
// documentation inherited // from interface ConnectionObserver
public synchronized void connectionFailed (Connection conn, IOException fault) public synchronized void connectionFailed (Connection conn, IOException fault)
{ {
// remove the client from the connection map // remove the client from the connection map
PresentsClient client = _conmap.remove(conn); PresentsClient client = _conmap.remove(conn);
if (client != null) { if (client != null) {
Log.info("Unmapped failed client [client=" + client + ", conn=" + conn + log.info("Unmapped failed client [client=" + client + ", conn=" + conn +
", fault=" + fault + "]."); ", fault=" + fault + "].");
// let the client know the connection went away // let the client know the connection went away
client.wasUnmapped(); client.wasUnmapped();
@@ -353,23 +361,23 @@ public class ClientManager
client.connectionFailed(fault); client.connectionFailed(fault);
} else if (!(conn instanceof AuthingConnection)) { } else if (!(conn instanceof AuthingConnection)) {
Log.info("Unmapped connection failed? [conn=" + conn + ", fault=" + fault + "]."); log.info("Unmapped connection failed? [conn=" + conn + ", fault=" + fault + "].");
Thread.dumpStack(); Thread.dumpStack();
} }
} }
// documentation inherited // from interface ConnectionObserver
public synchronized void connectionClosed (Connection conn) public synchronized void connectionClosed (Connection conn)
{ {
// remove the client from the connection map // remove the client from the connection map
PresentsClient client = _conmap.remove(conn); PresentsClient client = _conmap.remove(conn);
if (client != null) { if (client != null) {
Log.debug("Unmapped client [client=" + client + ", conn=" + conn + "]."); log.fine("Unmapped client [client=" + client + ", conn=" + conn + "].");
// let the client know the connection went away // let the client know the connection went away
client.wasUnmapped(); client.wasUnmapped();
} else { } else {
Log.info("Closed unmapped connection '" + conn + "'. " + log.info("Closed unmapped connection '" + conn + "'. " +
"Client probably not yet authenticated."); "Client probably not yet authenticated.");
} }
} }
@@ -379,7 +387,9 @@ public class ClientManager
{ {
report.append("* presents.ClientManager:\n"); report.append("* presents.ClientManager:\n");
report.append("- Sessions: "); report.append("- Sessions: ");
report.append(_usermap.size()).append(" total, "); synchronized (_usermap) {
report.append(_usermap.size()).append(" total, ");
}
report.append(_conmap.size()).append(" connected, "); report.append(_conmap.size()).append(" connected, ");
report.append(_penders.size()).append(" pending\n"); report.append(_penders.size()).append(" pending\n");
report.append("- Mapped users: ").append(_objmap.size()).append("\n"); report.append("- Mapped users: ").append(_objmap.size()).append("\n");
@@ -407,17 +417,20 @@ public class ClientManager
protected void clientSessionDidEnd (final PresentsClient client) protected void clientSessionDidEnd (final PresentsClient client)
{ {
// remove the client from the username map // remove the client from the username map
PresentsClient rc = _usermap.remove(client.getCredentials().getUsername()); PresentsClient rc;
synchronized (_usermap) {
rc = _usermap.remove(client.getCredentials().getUsername());
}
// sanity check just because we can // sanity check just because we can
if (rc == null) { if (rc == null) {
Log.warning("Unregistered client ended session " + client + "."); log.warning("Unregistered client ended session " + client + ".");
Thread.dumpStack(); Thread.dumpStack();
} else if (rc != client) { } else if (rc != client) {
Log.warning("Different clients with same username!? " + log.warning("Different clients with same username!? " +
"[c1=" + rc + ", c2=" + client + "]."); "[c1=" + rc + ", c2=" + client + "].");
} else { } else {
Log.info("Ending session " + client + "."); log.info("Ending session " + client + ".");
} }
// notify the observers that the session is ended // notify the observers that the session is ended
@@ -435,30 +448,26 @@ public class ClientManager
*/ */
protected void flushClients () protected void flushClients ()
{ {
ArrayList<PresentsClient> victims = null; ArrayList<PresentsClient> victims = new ArrayList<PresentsClient>();
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();
// first build a list of our victims (we can't flush clients directly while iterating due // first build a list of our victims
// to risk of a ConcurrentModificationException) synchronized (_usermap) {
for (PresentsClient client : _usermap.values()) { for (PresentsClient client : _usermap.values()) {
if (client.checkExpired(now)) { if (client.checkExpired(now)) {
if (victims == null) { victims.add(client);
victims = new ArrayList<PresentsClient>();
} }
victims.add(client);
} }
} }
if (victims != null) { // now end their sessions
for (PresentsClient client : victims) { for (PresentsClient client : victims) {
try { try {
Log.info("Client expired, ending session [client=" + client + log.info("Client expired, ending session [client=" + client +
", dtime=" + (now-client.getNetworkStamp()) + "ms]."); ", dtime=" + (now-client.getNetworkStamp()) + "ms].");
client.endSession(); client.endSession();
} catch (Exception e) { } catch (Exception e) {
Log.warning("Choke while flushing client [victim=" + client + "]."); log.log(Level.WARNING, "Choke while flushing client [victim=" + client + "].", e);
Log.logStackTrace(e);
}
} }
} }
} }
@@ -476,8 +485,8 @@ public class ClientManager
_clop.apply(clobj); _clop.apply(clobj);
} catch (Exception e) { } catch (Exception e) {
Log.warning("Client op failed [username=" + username + ", clop=" + _clop + "]."); log.log(Level.WARNING, "Client op failed [username=" + username +
Log.logStackTrace(e); ", clop=" + _clop + "].", e);
} finally { } finally {
releaseClientObject(username); releaseClientObject(username);
src/java/com/threerings/presents/server/net/Connection.java
+64 -91
View File
@@ -28,6 +28,8 @@ import java.net.InetAddress;
import java.nio.channels.SelectionKey; import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel; import java.nio.channels.SocketChannel;
import com.samskivert.util.StringUtil;
import com.threerings.io.FramedInputStream; import com.threerings.io.FramedInputStream;
import com.threerings.io.FramingOutputStream; import com.threerings.io.FramingOutputStream;
import com.threerings.io.ObjectInputStream; import com.threerings.io.ObjectInputStream;
@@ -39,27 +41,22 @@ import com.threerings.presents.net.PingRequest;
import com.threerings.presents.net.UpstreamMessage; import com.threerings.presents.net.UpstreamMessage;
/** /**
* The base connection class implements the net event handler interface * The base connection class implements the net event handler interface and processes raw incoming
* and processes raw incoming network data into a stream of parsed * network data into a stream of parsed <code>UpstreamMessage</code> objects. It also provides the
* <code>UpstreamMessage</code> objects. It also provides the means to * means to send messages to the client and facilities for checking delinquency.
* send messages to the client and facilities for checking delinquency.
*/ */
public abstract class Connection implements NetEventHandler public abstract class Connection implements NetEventHandler
{ {
/** /**
* Constructs a connection object that is associated with the supplied * Constructs a connection object that is associated with the supplied socket.
* socket.
* *
* @param cmgr The connection manager with which this connection is * @param cmgr The connection manager with which this connection is associated.
* associated. * @param selkey the key used by the NIO code to track this connection.
* @param selkey the key used by the NIO code to track this * @param channel The socket channel from which we'll be reading messages.
* connection.
* @param channel The socket channel from which we'll be reading
* messages.
* @param createStamp The time at which this connection was created. * @param createStamp The time at which this connection was created.
*/ */
public Connection (ConnectionManager cmgr, SelectionKey selkey, public Connection (ConnectionManager cmgr, SelectionKey selkey, SocketChannel channel,
SocketChannel channel, long createStamp) long createStamp)
throws IOException throws IOException
{ {
_cmgr = cmgr; _cmgr = cmgr;
@@ -69,9 +66,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Instructs the connection to pass parsed messages on to this handler * Instructs the connection to pass parsed messages on to this handler for processing. This
* for processing. This should be done before the connection is turned * should be done before the connection is turned loose to process messages.
* loose to process messages.
*/ */
public void setMessageHandler (MessageHandler handler) public void setMessageHandler (MessageHandler handler)
{ {
@@ -98,8 +94,7 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Returns the non-blocking socket object used to construct this * Returns the non-blocking socket object used to construct this connection.
* connection.
*/ */
public SocketChannel getChannel () public SocketChannel getChannel ()
{ {
@@ -107,8 +102,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Returns the address associated with this connection or null if it * Returns the address associated with this connection or null if it has no underlying socket
* has no underlying socket channel. * channel.
*/ */
public InetAddress getInetAddress () public InetAddress getInetAddress ()
{ {
@@ -124,8 +119,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Closes this connection and unregisters it from the connection * Closes this connection and unregisters it from the connection manager. This should only be
* manager. This should only be called from the conmgr thread. * called from the conmgr thread.
*/ */
public void close () public void close ()
{ {
@@ -144,9 +139,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Called when there is a failure reading or writing on this * Called when there is a failure reading or writing on this connection. We notify the
* connection. We notify the connection manager and close ourselves * connection manager and close ourselves down.
* down.
*/ */
public void handleFailure (IOException ioe) public void handleFailure (IOException ioe)
{ {
@@ -165,8 +159,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Returns the object input stream associated with this connection. * Returns the object input stream associated with this connection. This should only be used
* This should only be used by the connection manager. * by the connection manager.
*/ */
protected ObjectInputStream getObjectInputStream () protected ObjectInputStream getObjectInputStream ()
{ {
@@ -174,10 +168,9 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Instructs this connection to inherit its streams from the supplied * Instructs this connection to inherit its streams from the supplied connection object. This
* connection object. This is called by the connection manager when * is called by the connection manager when the time comes to pass streams from the authing
* the time comes to pass streams from the authing connection to the * connection to the running connection.
* running connection.
*/ */
protected void inheritStreams (Connection other) protected void inheritStreams (Connection other)
{ {
@@ -190,16 +183,13 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Returns the object output stream associated with this connection * Returns the object output stream associated with this connection (creating it if
* (creating it if necessary). This should only be used by the * necessary). This should only be used by the connection manager.
* connection manager.
*/ */
protected ObjectOutputStream getObjectOutputStream ( protected ObjectOutputStream getObjectOutputStream (FramingOutputStream fout)
FramingOutputStream fout)
{ {
// we're lazy about creating our output stream because we may be // we're lazy about creating our output stream because we may be inheriting it from our
// inheriting it from our authing connection and we don't want to // authing connection and we don't want to unnecessarily create it in that case
// unnecessarily create it in that case
if (_oout == null) { if (_oout == null) {
_oout = new ObjectOutputStream(fout); _oout = new ObjectOutputStream(fout);
} }
@@ -207,8 +197,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Sets the object output stream used by this connection. This should * Sets the object output stream used by this connection. This should only be called by the
* only be called by the connection manager. * connection manager.
*/ */
protected void setObjectOutputStream (ObjectOutputStream oout) protected void setObjectOutputStream (ObjectOutputStream oout)
{ {
@@ -216,26 +206,25 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Closes the socket associated with this connection. This happens * Closes the socket associated with this connection. This happens when we receive EOF, are
* when we receive EOF, are requested to close down or when our * requested to close down or when our connection fails.
* connection fails.
*/ */
protected void closeSocket () protected void closeSocket ()
{ {
if (_channel != null) { if (_channel == null) {
Log.debug("Closing channel " + this + "."); return;
try {
_channel.close();
} catch (IOException ioe) {
Log.warning("Error closing connection [conn=" + this +
", error=" + ioe + "].");
}
// clear out our references to prevent repeat closings
_selkey = null;
_channel = null;
} }
Log.debug("Closing channel " + this + ".");
try {
_channel.close();
} catch (IOException ioe) {
Log.warning("Error closing connection [conn=" + this + ", error=" + ioe + "].");
}
// clear out our references to prevent repeat closings
_selkey = null;
_channel = null;
} }
/** /**
@@ -248,9 +237,8 @@ public abstract class Connection implements NetEventHandler
int bytesIn = 0; int bytesIn = 0;
try { try {
// we're lazy about creating our input streams because we may // we're lazy about creating our input streams because we may be inheriting them from
// be inheriting them from our authing connection and we don't // our authing connection and we don't want to unnecessarily create them in that case
// want to unnecessarily create them in that case
if (_fin == null) { if (_fin == null) {
_fin = new FramedInputStream(); _fin = new FramedInputStream();
_oin = new ObjectInputStream(_fin); _oin = new ObjectInputStream(_fin);
@@ -259,12 +247,11 @@ public abstract class Connection implements NetEventHandler
} }
} }
// there may be more than one frame in the buffer, so we keep // there may be more than one frame in the buffer, so we keep reading them until we run
// reading them until we run out of data // out of data
while (_fin.readFrame(_channel)) { while (_fin.readFrame(_channel)) {
// make a note of how many bytes are in this frame // make a note of how many bytes are in this frame (including the frame length
// (including the frame length bytes which aren't reported // bytes which aren't reported in available())
// in available())
bytesIn = _fin.available() + 4; bytesIn = _fin.available() + 4;
// parse the message and pass it on // parse the message and pass it on
UpstreamMessage msg = (UpstreamMessage)_oin.readObject(); UpstreamMessage msg = (UpstreamMessage)_oin.readObject();
@@ -278,30 +265,18 @@ public abstract class Connection implements NetEventHandler
close(); close();
} catch (ClassNotFoundException cnfe) { } catch (ClassNotFoundException cnfe) {
try { Log.warning("Error reading message from socket [channel=" +
Log.warning("Error reading message from socket " + StringUtil.safeToString(_channel) + ", error=" + cnfe + "].");
"[channel=" + _channel + ", error=" + cnfe + "].");
} catch (Error err) {
Log.warning("Error reading message from socket " +
"and error printing channel [error=" + cnfe + "].");
}
// deal with the failure // deal with the failure
handleFailure((IOException) new IOException( String errmsg = "Unable to decode incoming message.";
"Unable to decode incoming message.").initCause(cnfe)); handleFailure((IOException) new IOException(errmsg).initCause(cnfe));
} catch (IOException ioe) { } catch (IOException ioe) {
// don't log a warning for the ever-popular "the client // don't log a warning for the ever-popular "the client dropped the connection" failure
// dropped the connection" failure
String msg = ioe.getMessage(); String msg = ioe.getMessage();
if (msg == null || msg.indexOf("reset by peer") == -1) { if (msg == null || msg.indexOf("reset by peer") == -1) {
try { Log.warning("Error reading message from socket [channel=" +
Log.warning("Error reading message from socket " + StringUtil.safeToString(_channel) + ", error=" + ioe + "].");
"[channel=" + _channel +
", error=" + ioe + "].");
} catch (Error err) {
Log.warning("Error reading message from socket " +
"and error printing channel [error=" + ioe + "].");
}
} }
// deal with the failure // deal with the failure
handleFailure(ioe); handleFailure(ioe);
@@ -326,9 +301,8 @@ public abstract class Connection implements NetEventHandler
} }
/** /**
* Posts a downstream message for delivery to this connection. The * Posts a downstream message for delivery to this connection. The message will be delivered by
* message will be delivered by the conmgr thread as soon as it gets * the conmgr thread as soon as it gets to it.
* to it.
*/ */
public void postMessage (DownstreamMessage msg) public void postMessage (DownstreamMessage msg)
{ {
@@ -349,8 +323,7 @@ public abstract class Connection implements NetEventHandler
protected MessageHandler _handler; protected MessageHandler _handler;
protected ClassLoader _loader; protected ClassLoader _loader;
/** The number of milliseconds beyond the ping interval that we allow /** The number of milliseconds beyond the ping interval that we allow a client's network
* a client's network connection to be idle before we forcibly * connection to be idle before we forcibly disconnect them. */
* disconnect them. */
protected static final long LATENCY_GRACE = 30 * 1000L; protected static final long LATENCY_GRACE = 30 * 1000L;
} }