Client side message throttling. The server is configured with a hard limit, it

communicates that limit to the client which queues outgoing messages so that it
does not exceed the limit. When the throttle is changed, the server does not
apply the new throttle until the client has ACKed it to avoid edge cases where
the client is up against the limit when the throttle changes.

ActionScript side of all of this forthcoming as well as more testing.


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

src/java/com/threerings/presents/client/BlockingCommunicator.java

@@ -704,6 +704,15 @@ public class BlockingCommunicator extends Communicator
                 return;
             }
 
+            // make sure we're not exceeding our outgoing throttle rate
+            while (_client.getOutgoingMessageThrottle().throttleOp()) {
+                try {
+                    Thread.sleep(2);
+                } catch (InterruptedException ie) {
+                    // no problem
+                }
+            }
+
             try {
                 // write the message out the socket
                 sendMessage(msg);

src/java/com/threerings/presents/client/Client.java

@@ -28,6 +28,7 @@ import com.samskivert.util.ObserverList;
 import com.samskivert.util.RunAnywhere;
 import com.samskivert.util.RunQueue;
 import com.samskivert.util.StringUtil;
+import com.samskivert.util.Throttle;
 
 import com.threerings.presents.client.InvocationService.ConfirmListener;
 import com.threerings.presents.data.AuthCodes;
@@ -41,6 +42,7 @@ import com.threerings.presents.net.BootstrapData;
 import com.threerings.presents.net.Credentials;
 import com.threerings.presents.net.PingRequest;
 import com.threerings.presents.net.PongResponse;
+import com.threerings.presents.net.ThrottleUpdatedMessage;
 
 import static com.threerings.presents.Log.log;
 
@@ -60,6 +62,9 @@ public class Client
     /** The maximum size of a datagram. */
     public static final int MAX_DATAGRAM_SIZE = 1500;
 
+    /** Our default maximum outgoing message rate as { messages, milliseconds }. */
+    public static final int[] DEFAULT_MAX_MSG_RATE = { 50, 5*1000 };
+
     /**
      * Constructs a client object with the supplied credentials and RunQueue. The creds will be
      * used to authenticate with any server to which this client attempts to connect. The RunQueue
@@ -77,6 +82,10 @@ public class Client
     {
         _creds = creds;
         _runQueue = runQueue;
+
+        // initialize our default throttle; we use a slightly lowered rate from the default to
+        // avoid edge cases where we and the server disagree about a single millisecond
+        _outThrottle = new Throttle(DEFAULT_MAX_MSG_RATE[0]-1, DEFAULT_MAX_MSG_RATE[1]);
     }
 
     /**
@@ -631,6 +640,25 @@ public class Client
         return new ClientCommunicator(this);
     }
 
+    /**
+     * Configures the outgoing message throttle. This is done when the server informs us that a new
+     * rate is in effect.
+     */
+    protected synchronized void setOutgoingMessageThrottle (int msgs, long period)
+    {
+        log.info("Updating outgoing message throttle", "msgs", msgs, "period", period);
+        _outThrottle.reinit(msgs, period);
+        _comm.postMessage(new ThrottleUpdatedMessage());
+    }
+
+    /**
+     * Returns our outgoing message throttle. Used by the communicator's writer.
+     */
+    protected synchronized Throttle getOutgoingMessageThrottle ()
+    {
+        return _outThrottle;
+    }
+
     /**
      * Called every five seconds; ensures that we ping the server if we haven't communicated in a
      * long while and periodically resyncs the client and server clock deltas.
@@ -1046,6 +1074,9 @@ public class Client
     /** Our tick interval id. */
     protected Interval _tickInterval;
 
+    /** Our outgoing message throttle. */
+    protected Throttle _outThrottle = new Throttle(DEFAULT_MAX_MSG_RATE[0], DEFAULT_MAX_MSG_RATE[1]);
+
     /** How often we recompute our time offset from the server. */
     protected static final long CLOCK_SYNC_INTERVAL = 600 * 1000L;
 

src/java/com/threerings/presents/client/ClientDObjectMgr.java

@@ -53,6 +53,7 @@ import com.threerings.presents.net.PongResponse;
 import com.threerings.presents.net.SubscribeRequest;
 import com.threerings.presents.net.UnsubscribeRequest;
 import com.threerings.presents.net.UnsubscribeResponse;
+import com.threerings.presents.net.UpdateThrottleMessage;
 
 import static com.threerings.presents.Log.log;
 
@@ -199,6 +200,10 @@ public class ClientDObjectMgr
             } else if (obj instanceof PongResponse) {
                 _client.gotPong((PongResponse)obj);
 
+            } else if (obj instanceof UpdateThrottleMessage) {
+                UpdateThrottleMessage upmsg = (UpdateThrottleMessage)obj;
+                _client.setOutgoingMessageThrottle(upmsg.messages, upmsg.period);
+
             } else if (obj instanceof ObjectAction<?>) {
                 ObjectAction<?> act = (ObjectAction<?>)obj;
                 if (act.subscribe) {