From 9f2682a5e456bbf6be18d9706a2738cc648b20b5 Mon Sep 17 00:00:00 2001 From: Michael Bayne Date: Thu, 9 Oct 2008 20:18:12 +0000 Subject: [PATCH] All of the ActionScript side of adjustable outgoing message rate throttling except the actual throttling part. Need to think about a good way to do this without blocking the whole client when we hit the throttle. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5428 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../com/threerings/presents/client/Client.as | 66 +++---- .../presents/client/ClientDObjectMgr.as | 4 + .../presents/net/ThrottleUpdatedMessage.as | 30 ++++ .../presents/net/UpdateThrottleMessage.as | 45 +++++ src/as/com/threerings/util/Throttle.as | 168 ++++++++++++++++++ 5 files changed, 282 insertions(+), 31 deletions(-) create mode 100644 src/as/com/threerings/presents/net/ThrottleUpdatedMessage.as create mode 100644 src/as/com/threerings/presents/net/UpdateThrottleMessage.as create mode 100644 src/as/com/threerings/util/Throttle.as diff --git a/src/as/com/threerings/presents/client/Client.as b/src/as/com/threerings/presents/client/Client.as index 3c6393968..4331cfa01 100644 --- a/src/as/com/threerings/presents/client/Client.as +++ b/src/as/com/threerings/presents/client/Client.as @@ -28,6 +28,7 @@ import flash.utils.Timer; import com.threerings.util.Log; import com.threerings.util.MethodQueue; +import com.threerings.util.Throttle; import com.threerings.presents.client.InvocationService_ConfirmListener; import com.threerings.presents.data.ClientObject; @@ -41,12 +42,16 @@ 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; public class Client extends EventDispatcher { /** The default port on which the server listens for client connections. */ public static const DEFAULT_SERVER_PORTS :Array = [ 47624 ]; + /** Our default maximum outgoing message rate in messages per second. */ + public static const DEFAULT_MSGS_PER_SECOND :int = 10; + private static const log :Log = Log.getLog(Client); public function Client (creds :Credentials) @@ -390,6 +395,34 @@ public class Client extends EventDispatcher notifyObservers(ClientEvent.CLIENT_OBJECT_CHANGED); } + /** + * Convenience method to dispatch a client event to any listeners and return the result of + * dispatchEvent. + */ + public function notifyObservers (evtCode :String, cause :Error = null) :Boolean + { + return dispatchEvent(new ClientEvent(evtCode, this, cause)); + } + + /** + * Called by the omgr when we receive a pong packet. + */ + internal function gotPong (pong :PongResponse) :void + { + // TODO: compute time delta? + } + + internal function setOutgoingMessageThrottle (messagesPerSec :int) :void + { + _outThrottle.reinit(messagesPerSec, 1000); + _comm.postMessage(new ThrottleUpdatedMessage()); + } + + internal function getOutgoingMessageThrottle () :Throttle + { + return _outThrottle; + } + internal function cleanup (logonError :Error) :void { // clear out our references @@ -413,24 +446,6 @@ public class Client extends EventDispatcher _switcher = null; } - /** - * Called by the omgr when we receive a pong packet. - */ - public function gotPong (pong :PongResponse) :void - { - // TODO: compute time delta? - } - - /** - * Convenience method to dispatch a client event to any listeners - * and return the result of dispatchEvent. - */ - public function notifyObservers (evtCode :String, cause :Error = null) - :Boolean - { - return dispatchEvent(new ClientEvent(evtCode, this, cause)); - } - /** The credentials we used to authenticate with the server. */ protected var _creds :Credentials; @@ -458,9 +473,8 @@ public class Client extends EventDispatcher /** The entity that manages our network communications. */ protected var _comm :Communicator; - /** Our list of client observers. */ -// protected var _observers :ObserverList = -// new ObserverList(ObserverList.SAFE_IN_ORDER_NOTIFY); + /** Our outgoing message throttle. */ + protected var _outThrottle :Throttle = new Throttle(DEFAULT_MSGS_PER_SECOND, 1000); /** The set of bootstrap service groups this client cares about. */ protected var _bootGroups :Array = new Array(InvocationCodes.GLOBAL_GROUP); @@ -477,15 +491,5 @@ public class Client extends EventDispatcher /** Used to temporarily track our server switcher so that we can tell when we're logging off * whether or not we're switching servers or actually ending our session. */ protected var _switcher :ServerSwitcher; - - // client observer constants - /* - internal static const CLIENT_DID_LOGON :int = 0; - internal static const CLIENT_FAILED_TO_LOGON :int = 1; - internal static const CLIENT_OBJECT_CHANGED :int = 2; - internal static const CLIENT_CONNECTION_FAILED :int = 3; - internal static const CLIENT_WILL_LOGOFF :int = 4; - internal static const CLIENT_DID_LOGOFF :int = 5; - */ } } diff --git a/src/as/com/threerings/presents/client/ClientDObjectMgr.as b/src/as/com/threerings/presents/client/ClientDObjectMgr.as index 8c37dba1a..7ce44e0b6 100644 --- a/src/as/com/threerings/presents/client/ClientDObjectMgr.as +++ b/src/as/com/threerings/presents/client/ClientDObjectMgr.as @@ -48,6 +48,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; /** * The client distributed object manager manages a set of proxy objects @@ -170,6 +171,9 @@ public class ClientDObjectMgr } else if (msg is PongResponse) { _client.gotPong(msg as PongResponse); + + } else if (msg is UpdateThrottleMessage) { + _client.setOutgoingMessageThrottle((msg as UpdateThrottleMessage).messagesPerSec); } } diff --git a/src/as/com/threerings/presents/net/ThrottleUpdatedMessage.as b/src/as/com/threerings/presents/net/ThrottleUpdatedMessage.as new file mode 100644 index 000000000..654fcffe8 --- /dev/null +++ b/src/as/com/threerings/presents/net/ThrottleUpdatedMessage.as @@ -0,0 +1,30 @@ +// +// $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.net { + +/** + * Notifies the server that the client has received its UpdateThrottleMessage. + */ +public class ThrottleUpdatedMessage extends UpstreamMessage +{ +} +} diff --git a/src/as/com/threerings/presents/net/UpdateThrottleMessage.as b/src/as/com/threerings/presents/net/UpdateThrottleMessage.as new file mode 100644 index 000000000..8dca3130e --- /dev/null +++ b/src/as/com/threerings/presents/net/UpdateThrottleMessage.as @@ -0,0 +1,45 @@ +// +// $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.net { + +import com.threerings.io.ObjectInputStream; + +/** + * Notifies the client that its message throttle has been updated. + */ +public class UpdateThrottleMessage extends DownstreamMessage +{ + /** The number of messages allowed per second. */ + public var messagesPerSec :int; + + public function UpdateThrottleMessage (messagesPerSec :int = 0) + { + this.messagesPerSec = messagesPerSec; + } + + override public function readObject (ins :ObjectInputStream) :void + { + super.readObject(ins); + messagesPerSec = ins.readInt(); + } +} +} diff --git a/src/as/com/threerings/util/Throttle.as b/src/as/com/threerings/util/Throttle.as new file mode 100644 index 000000000..310f10447 --- /dev/null +++ b/src/as/com/threerings/util/Throttle.as @@ -0,0 +1,168 @@ +// +// $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.util { + +import flash.utils.getTimer; // function import + +/** + * A throttle is used to prevent code from attempting a particular operation too often. Often it is + * desirable to retry an operation under failure conditions, but simplistic approaches to retrying + * operations can lead to large numbers of spurious attempts to do something that will obviously + * fail. The throttle class provides a mechanism for limiting such attempts by measuring whether or + * not an activity has been performed N times in the last M seconds. The user of the class decides + * the appropriate throttle parameters and then simply calls through to throttle to determine + * whether or not to go ahead with the operation. + * + *

For example: + * + * 

+ * protected Throttle _throttle = new Throttle(5, 60000L);
+ *
+ * public void performOp ()
+ *     throws UnavailableException
+ * {
+ *     if (_throttle.throttleOp()) {
+ *         throw new UnavailableException();
+ *     }
+ *
+ *     // perform operation
+ * }
+ *
+ */ +public class Throttle +{ + /** + * Constructs a new throttle instance that will allow the specified number of operations to + * proceed within the specified period (the period is measured in milliseconds). + * + *

As operations and period define a ratio, use the smallest value possible for + * operations as an array is created to track the time at which each operation was + * performed (e.g. use 6 ops per 10 seconds rather than 60 ops per 100 seconds if + * possible). However, note that you may not always want to reduce the ratio as much as + * possible if you wish to allow bursts of operations up to some large value. + */ + public function Throttle (operations :int, period :int) + { + _ops = new Array(operations); + _period = period; + } + + /** + * Updates the number of operations for this throttle to a new maximum, retaining the current + * history of operations if the limit is being increased and truncating the oldest operations + * if the limit is decreased. + * + * @param operations the new maximum number of operations. + * @param period the new period. + */ + public function reinit (operations :int, period :int) :void + { + var ops :Array = new Array(operations); + // copy up to 'operations' of our most recent operations into our new array + for (var ii :int = 0; ii < operations; ii++) { + ops[operations-ii-1] = _ops[(_oldestOp + _ops.length - ii - 1) % _ops.length]; + } + _oldestOp = Math.max(0, operations-_ops.length); + _ops = ops; + _period = period; + } + + /** + * Registers an attempt at an operation and returns true if the operation should be performed + * or false if it should be throttled (meaning N operations have already been performed in the + * last M seconds). + * + * @return true if the throttle is activated, false if the operation can proceed. + */ + public function throttleOp () :Boolean + { + return throttleOpAt(getTimer()); + } + + /** + * Registers an attempt at an operation and returns true if the operation should be performed + * or false if it should be throttled (meaning N operations have already been performed in the + * last M seconds). + * + * @param timeStamp the timestamp at which this operation is being attempted. + * + * @return true if the throttle is activated, false if the operation can proceed. + */ + public function throttleOpAt (timeStamp :int) :Boolean + { + if (wouldThrottle(timeStamp)) { + return true; + } + + noteOp(timeStamp); + return false; + } + + /** + * Check to see if we would throttle an operation occuring at the specified timestamp. + * Typically used in conjunction with {@link #noteOp}. + */ + public function wouldThrottle (timeStamp :int) :Boolean + { + // if the oldest operation was performed less than _period ago, we need to throttle + var elapsed :int = timeStamp - _ops[_oldestOp]; + // if negative time elapsed, we must be running on windows; let's just cope by not + // throttling + return (elapsed >= 0 && elapsed < _period); + } + + /** + * Note that an operation occurred at the specified timestamp. This method should be used with + * {@link #wouldThrottle} to note an operation that has already been cleared to + * occur. Typically this is used if there is another limiting factor besides the throttle that + * determines whether the operation can occur. You are responsible for calling this method in a + * safe and timely manner after using wouldThrottle. + */ + public function noteOp (timeStamp :int) :void + { + // overwrite the oldest operation with the current time and move the oldest operation + // pointer to the second oldest operation (which is now the oldest as we overwrote the + // oldest) + _ops[_oldestOp] = timeStamp; + _oldestOp = (_oldestOp + 1) % _ops.length; + } + + /** + * Returns the timestamp of the most recently recorded operation. + */ + public function getLatestOperation () :int + { + return _ops[(_oldestOp + _ops.length - 1) % _ops.length]; + } + + // from Object + public function toString () :String + { + var oldest :int = getTimer() - _ops[_oldestOp]; + return _ops.length + " ops per " + _period + "ms (oldest " + oldest + ")"; + } + + protected var _ops :Array; + protected var _oldestOp :int; + protected var _period :int; +} +}