Track various peer stats.

git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5760 542714f4-19e9-0310-aa3c-eee0fc999fb1

src/java/com/threerings/presents/peer/server/PeerManager.java

@@ -23,6 +23,7 @@ package com.threerings.presents.peer.server;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -157,6 +158,43 @@ public abstract class PeerManager
         protected abstract void execute ();
     }
 
+    /** Returned by {@link #getStats}. */
+    public static class Stats implements Cloneable
+    {
+        /** The number of locks this node has acquired. */
+        public long locksAcquired;
+
+        /** The number of milliseconds spent waiting to acquire locks. */
+        public long lockAcquireWait;
+
+        /** The number of locks this node has released. */
+        public long locksReleased;
+
+        /** The number of locks this node has had hijacked. */
+        public long locksHijacked;
+
+        /** The number of lock requests that have timed out. */
+        public long lockTimeouts;
+
+        /** The number of node actions we've invoked. */
+        public long nodeActionsInvoked;
+
+        /** The total number of messages received from all of our peers. This is updated on the
+         * conmgr thread which is why it's atomic. */
+        public AtomicLong peerMessagesIn = new AtomicLong(0);
+
+        /** The total number of messages sent to all of our peers. */
+        public long peerMessagesOut;
+
+        @Override public Stats clone () {
+            try {
+                return (Stats)super.clone();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
     /**
      * Creates an uninitialized peer manager.
      */
@@ -359,6 +397,10 @@ public abstract class PeerManager
         if (!invoked && onDropped != null) {
             onDropped.run();
         }
+
+        if (invoked) {
+            _stats.nodeActionsInvoked++; // stats!
+        }
     }
 
     /**
@@ -497,8 +539,7 @@ public abstract class PeerManager
             public void requestCompleted (String result) {
                 if (result == null) {
                     if (_suboids.isEmpty()) {
-                        _nodeobj.addToLocks(lock);
+                        lockAcquired(lock, 0L, listener);
-                        listener.requestCompleted(_nodeName);
                     } else {
                         _locks.put(lock, new LockHandler(lock, true, listener));
                     }
@@ -521,8 +562,7 @@ public abstract class PeerManager
             public void requestCompleted (String result) {
                 if (_nodeName.equals(result)) {
                     if (_suboids.isEmpty()) {
-                        _nodeobj.removeFromLocks(lock);
+                        lockReleased(lock, listener);
-                        listener.requestCompleted(null);
                     } else {
                         _locks.put(lock, new LockHandler(lock, false, listener));
                     }
@@ -705,11 +745,20 @@ public abstract class PeerManager
         _nodeobj.setCacheData(new NodeObject.CacheData(cache, data));
     }
 
+    /**
+     * Returns a snapshot of runtime statistics tracked by the peer manager.
+     */
+    public Stats getStats ()
+    {
+        return _stats.clone();
+    }
+
     // from interface ShutdownManager.Shutdowner
     public void shutdown ()
     {
         if (_nodeName == null) { // sanity check
-            throw new IllegalStateException("Shutting down PeerManager that was never initialized.");
+            throw new IllegalStateException(
+                "Shutting down PeerManager that was never initialized.");
         }
 
         // clear out our invocation service
@@ -952,6 +1001,7 @@ public abstract class PeerManager
     protected void droppedLock (final NodeObject.Lock lock)
     {
         _nodeobj.removeFromLocks(lock);
+        _stats.locksHijacked++;
         _dropobs.apply(new ObserverList.ObserverOp<DroppedLockObserver>() {
             public boolean apply (DroppedLockObserver observer) {
                 observer.droppedLock(lock);
@@ -1021,6 +1071,8 @@ public abstract class PeerManager
         // this may be the first we've heard of this guy, so let's refresh our peers and
         // potentially connect right back to him
         refreshPeers();
+        // pass our stats record in so that it can count up messages in/out
+        session.setStats(_stats);
     }
 
     /**
@@ -1028,7 +1080,7 @@ public abstract class PeerManager
      */
     protected void peerEndedSession (PeerSession session)
     {
-        // nothing presently
+        // nada
     }
 
     /**
@@ -1083,6 +1135,21 @@ public abstract class PeerManager
         }
     }
 
+    protected void lockAcquired (NodeObject.Lock lock, long wait, ResultListener<String> listener)
+    {
+        _nodeobj.addToLocks(lock);
+        _stats.locksAcquired++;
+        _stats.lockAcquireWait += wait;
+        listener.requestCompleted(_nodeName);
+    }
+
+    protected void lockReleased (NodeObject.Lock lock, ResultListener<String> listener)
+    {
+        _nodeobj.removeFromLocks(lock);
+        _stats.locksReleased++;
+        listener.requestCompleted(null);
+    }
+
     /**
      * Handles a lock in a state of resolution.
      */
@@ -1118,6 +1185,7 @@ public abstract class PeerManager
                 public void expired () {
                     log.warning("Lock handler timed out, acting anyway", "lock", _lock,
                                 "acquire", _acquire);
+                    _stats.lockTimeouts++;
                     activate();
                 }
             }).schedule(LOCK_TIMEOUT);
@@ -1198,6 +1266,7 @@ public abstract class PeerManager
         {
             cancel();
             _locks.remove(_lock);
+            _stats.locksHijacked++;
             listeners.requestCompleted(nodeName);
         }
 
@@ -1264,11 +1333,9 @@ public abstract class PeerManager
         {
             _locks.remove(_lock);
             if (_acquire) {
-                _nodeobj.addToLocks(_lock);
+                lockAcquired(_lock, System.currentTimeMillis() - _startStamp, listeners);
-                listeners.requestCompleted(_nodeName);
             } else {
-                _nodeobj.removeFromLocks(_lock);
+                lockReleased(_lock, listeners);
-                listeners.requestCompleted(null);
             }
         }
 
@@ -1287,6 +1354,7 @@ public abstract class PeerManager
         protected boolean _acquire;
         protected ArrayIntSet _remoids;
         protected Interval _timeout;
+        protected long _startStamp = System.currentTimeMillis();
     }
 
     // (this need not use a runqueue as all it will do is post an invoker unit)
@@ -1319,6 +1387,9 @@ public abstract class PeerManager
     /** Locks in the process of resolution. */
     protected Map<NodeObject.Lock, LockHandler> _locks = Maps.newHashMap();
 
+    /** Used to track runtime statistics. */
+    protected Stats _stats = new Stats();
+
     // our service dependencies
     @Inject protected ConnectionManager _conmgr;
     @Inject protected ClientManager _clmgr;

src/java/com/threerings/presents/peer/server/PeerSession.java

@@ -27,6 +27,8 @@ import com.samskivert.util.Throttle;
 
 import com.threerings.presents.dobj.DObject;
 import com.threerings.presents.net.BootstrapData;
+import com.threerings.presents.net.DownstreamMessage;
+import com.threerings.presents.net.Message;
 import com.threerings.presents.peer.data.NodeObject;
 import com.threerings.presents.peer.net.PeerBootstrapData;
 import com.threerings.presents.server.PresentsSession;
@@ -47,6 +49,21 @@ public class PeerSession extends PresentsSession
         _peermgr = peermgr;
     }
 
+    /**
+     * Passes in a stats instance that this peer can use to note when it trafficks.
+     */
+    public void setStats (PeerManager.Stats stats)
+    {
+        _stats = stats;
+    }
+
+    @Override // from PresentsSession
+    public void handleMessage (Message message)
+    {
+        super.handleMessage(message);
+        _stats.peerMessagesIn.incrementAndGet();
+    }
+
     @Override // from PresentsSession
     protected BootstrapData createBootstrapData ()
     {
@@ -98,6 +115,16 @@ public class PeerSession extends PresentsSession
         _peermgr.peerEndedSession(this);
     }
 
+    @Override // from PresentsSession
+    protected final boolean postMessage (DownstreamMessage msg)
+    {
+        if (!super.postMessage(msg)) {
+            return false;
+        }
+        _stats.peerMessagesOut++;
+        return true;
+    }
+
     @Override // from PresentsSession
     protected void subscribedToObject (DObject object)
     {
@@ -134,6 +161,7 @@ public class PeerSession extends PresentsSession
     }
 
     protected PeerManager _peermgr;
+    protected PeerManager.Stats _stats;
     protected int _cloid;
     protected long _nextThrottleWarning;
 }