diff --git a/src/java/com/threerings/presents/client/BlockingCommunicator.java b/src/java/com/threerings/presents/client/BlockingCommunicator.java index e70d71f19..6d8f4ec9b 100644 --- a/src/java/com/threerings/presents/client/BlockingCommunicator.java +++ b/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); diff --git a/src/java/com/threerings/presents/client/Client.java b/src/java/com/threerings/presents/client/Client.java index b8f4a3c6e..737a46da6 100644 --- a/src/java/com/threerings/presents/client/Client.java +++ b/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; diff --git a/src/java/com/threerings/presents/client/ClientDObjectMgr.java b/src/java/com/threerings/presents/client/ClientDObjectMgr.java index 89be03361..c60e471fc 100644 --- a/src/java/com/threerings/presents/client/ClientDObjectMgr.java +++ b/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) { diff --git a/src/java/com/threerings/presents/net/ThrottleUpdatedMessage.java b/src/java/com/threerings/presents/net/ThrottleUpdatedMessage.java new file mode 100644 index 000000000..c7c63b2ee --- /dev/null +++ b/src/java/com/threerings/presents/net/ThrottleUpdatedMessage.java @@ -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 {@link UpdateThrottleMessage}. + */ +public class ThrottleUpdatedMessage extends UpstreamMessage +{ + // nothing needed +} diff --git a/src/java/com/threerings/presents/net/UpdateThrottleMessage.java b/src/java/com/threerings/presents/net/UpdateThrottleMessage.java new file mode 100644 index 000000000..b99c2dbdf --- /dev/null +++ b/src/java/com/threerings/presents/net/UpdateThrottleMessage.java @@ -0,0 +1,49 @@ +// +// $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 client that its message throttle has been updated. + */ +public class UpdateThrottleMessage extends DownstreamMessage +{ + /** The number of messages allowed per throttle period. */ + public final int messages; + + /** The throttle period. */ + public final long period; + + /** + * Zero argument constructor used when unserializing an instance. + */ + public UpdateThrottleMessage () + { + this.messages = 0; + this.period = 0; + } + + public UpdateThrottleMessage (int messages, long period) + { + this.messages = messages; + this.period = period; + } +} diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java index 0ab1035a9..b9c2e0a4c 100644 --- a/src/java/com/threerings/presents/server/PresentsClient.java +++ b/src/java/com/threerings/presents/server/PresentsClient.java @@ -23,7 +23,7 @@ package com.threerings.presents.server; import java.net.InetAddress; -import java.util.LinkedList; +import java.util.List; import java.util.Map; import java.util.TimeZone; @@ -35,7 +35,6 @@ import com.google.inject.Inject; import com.samskivert.util.IntMap; import com.samskivert.util.IntMaps; -import com.samskivert.util.Interval; import com.samskivert.util.ResultListener; import com.samskivert.util.Throttle; @@ -43,6 +42,7 @@ import com.threerings.util.Name; import com.threerings.presents.annotation.AnyThread; import com.threerings.presents.annotation.EventThread; +import com.threerings.presents.client.Client; import com.threerings.presents.data.ClientObject; import com.threerings.presents.dobj.DEvent; import com.threerings.presents.dobj.DObject; @@ -61,8 +61,10 @@ import com.threerings.presents.net.ObjectResponse; import com.threerings.presents.net.PingRequest; import com.threerings.presents.net.PongResponse; import com.threerings.presents.net.SubscribeRequest; +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; @@ -174,6 +176,32 @@ public class PresentsClient } } + /** + * Configures the rate at which incoming messages are throttled for this client. This will + * communicate the new limit to the client and begin enforcing the limit when the client has + * acknowledged the new limit. + * + *

Note: this means that a hacked client can refuse to ACK message rate reductions + * and continue to use the most generous rate ever assigned to it. Don't increase the throttle + * beyond the default for untrusted clients. This mechanism exists so that trusted clients can + * have their throttle relaxed in a robust manner which will not result in disconnects if the + * client happens to be at or near the throttle limit when the throttle is reduced. + * + * @param messages the number of messages allowed in the period. + * @param period the throttle period (in milliseconds). + */ + @EventThread + public void setIncomingMessageThrottle (int messages, long period) + { + // we do a couple of things to make sure we don't accidentally hit our throttle: we use 2x + // messages/period so that if the client sends all of its messages in 1ms and then tries to + // send another full batch in the final ms after the throttle lets up, we don't freak out + // if we disagree about that ms and we reduce the client's message count by one + _pendingThrottles.add(new int[] { 2*messages, (int)(2*period) }); + postMessage(new UpdateThrottleMessage(messages-1, period)); + // when we get a ThrottleUpdatedMessage from the client, we'll apply the new throttle + } + /** * Danger: this method is not for general consumption. This changes the username of * the client, but should only be done very early in a user's session, when you know that no @@ -194,8 +222,7 @@ public class PresentsClient public void setUsername (Name username, final UserChangeListener ucl) { ClientResolutionListener clr = new ClientResolutionListener() { - public void clientResolved (final Name username, final ClientObject clobj) - { + public void clientResolved (final Name username, final ClientObject clobj) { // if they old client object is gone by now, they ended their session while we were // switching, so freak out if (_clobj == null) { @@ -228,8 +255,7 @@ public class PresentsClient /** * Finish the final phase of the switch. */ - protected void finishResolved (Name username, ClientObject clobj) - { + protected void finishResolved (Name username, ClientObject clobj) { // let the client know that the rug has been yanked out from under their ass Object[] args = new Object[] { Integer.valueOf(clobj.getOid()) }; _clobj.postMessage(ClientObject.CLOBJ_CHANGED, args); @@ -325,11 +351,6 @@ public class PresentsClient // clear out the client object so that we know the session is over _clobj = null; - - // Cancel overflow processing - if (_overflow != null) { - _overflow.stop(); - } } /** @@ -393,62 +414,10 @@ public class PresentsClient // ", msg=" + message + "]."); return; - } else if (_overflow != null && _overflow.size() > 0) { - // We've got some overflow, the new message has to dispatch after those - if (_overflow.size() >= _overflow.limit) { - // Can't take any more overflow, bail out - handleThrottleExceeded(); - - } else { - _overflow.enqueue(message); - _overflow.start(); - } - return; - } else if (_throttle.throttleOp(message.received)) { - // We're throttled, check if overflow is allowed - if (_overflow != null && _overflow.limit > 0) { - // Enter overflow mode and save message for later - _overflow.enqueue(message); - _overflow.start(); - return; - } - handleThrottleExceeded(); } - dispatch(message); - } - - /** - * Process the overflowed messages one at a time as long as the throttle will let us. - */ - protected void handleOverflowMessages () - { - // Special case, we've already reached our limit since overflow was kicked off - if (_throttle == null) { - _overflow.stop(); - return; - } - - long now = System.currentTimeMillis(); - while (true) { - // We're done, the next message will be business as usual - if (_overflow.size() == 0) { - _overflow.stop(); - return; - } - - // Dispatch the next message if it isn't throttled - if (_throttle.throttleOp(now)) { - return; - } - dispatch(_overflow.dequeue()); - } - } - - protected void dispatch (UpstreamMessage message) - { // we dispatch to a message dispatcher that is specialized for the particular class of // message that we received MessageDispatcher disp = _disps.get(message.getClass()); @@ -587,13 +556,13 @@ public class PresentsClient } /** - * Creates our incoming message throttle. Subclasses can override this to customize the message - * rate limit. + * Creates our incoming message throttle. Use {@link #setIncomingMessageThrottle} to adjust the + * throttle for running clients. */ protected Throttle createIncomingMessageThrottle () { - // more than 100 messages in 10 seconds and you're audi like 5000 - return new Throttle(DEFAULT_THROTTLE_MESSAGE_LIMIT, DEFAULT_THROTTLE_TIME_LIMIT); + // see setIncomingMessageThrottle for more details on all of this + return new Throttle(2*Client.DEFAULT_MAX_MSG_RATE[0], 2*Client.DEFAULT_MAX_MSG_RATE[1]); } /** @@ -601,26 +570,12 @@ public class PresentsClient */ protected void handleThrottleExceeded () { - log.warning( - "Client exceeded message limit, disconnecting", "client", this, "throttle", _throttle); + log.warning("Client exceeded incoming message throttle, disconnecting", + "client", this, "throttle", _throttle); safeEndSession(); _throttle = null; } - /** - * Allow throttled messages to hang around until permitted to send. - */ - protected void setOverflowLimit (int limit) - { - if (_overflow == null) { - if (limit == 0) { - return; - } - _overflow = new Overflow(); - } - _overflow.limit = limit; - } - /** * Makes a note that this client is no longer subscribed to this object. The subscription map * is used to clean up after the client when it goes away. @@ -884,8 +839,7 @@ public class PresentsClient // don't log dropped messages unless we're dropping a lot of them (meaning something is // still queueing messages up for this dead client even though it shouldn't be) if (++_messagesDropped % 50 == 0) { - log.warning("Dropping many messages? [client=" + this + - ", count=" + _messagesDropped + "]."); + log.warning("Dropping many messages?", "client", this, "count", _messagesDropped); } // make darned sure we don't have any remaining subscriptions @@ -897,6 +851,26 @@ public class PresentsClient return false; } + /** + * Notifies this client that its throttle was updated. + */ + protected void throttleUpdated () + { + _omgr.postRunnable(new Runnable() { + public void run () { + if (_pendingThrottles.size() == 0) { + log.warning("Received throttleUpdated but have no pending throttles", + "client", this); + return; + } + int[] data = _pendingThrottles.remove(0); + log.info("Applying updated throttle", "client", this, + "msgs", data[0], "period", data[1]); + _throttle.reinit(data[0], data[1]); + } + }); + } + @Override public String toString () { @@ -1071,6 +1045,18 @@ public class PresentsClient } } + /** + * Processes throttle updated messages. + */ + protected static class ThrottleUpdatedDispatcher implements MessageDispatcher + { + public void dispatch (final PresentsClient client, UpstreamMessage msg) + { + log.debug("Client ACKed throttle update", "client", client); + client.throttleUpdated(); + } + } + /** * Processes logoff requests. */ @@ -1078,76 +1064,11 @@ public class PresentsClient { public void dispatch (final PresentsClient client, UpstreamMessage msg) { - log.debug("Client requested logoff " + client + "."); + log.debug("Client requested logoff", "client", client); client.safeEndSession(); } } - /** - * Contains information about message overflow. - */ - protected class Overflow - { - /** Maxmimum size of queue before disconnection. */ - public int limit; - - /** - * Starts processing the overflowed messages every so often. - */ - public void start () - { - if (_interval == null) { - _interval = new Interval(_omgr) { - @Override public void expired () { - handleOverflowMessages(); - } - }; - _interval.schedule(OVERFLOW_CHECK_INTERVAL, true); - } - } - - /** - * Stops processing overflowed messages. - */ - public void stop () - { - if (_interval != null) { - _interval.cancel(); - _interval = null; - } - } - - /** - * Gets the number of overflowed messages. - */ - public int size () - { - return _queue.size(); - } - - /** - * Adds a message to the end of the queue. - */ - public void enqueue (UpstreamMessage message) - { - _queue.add(message); - } - - /** - * Removes and returns the least recently added message. - */ - public UpstreamMessage dequeue () - { - return _queue.remove(); - } - - /** Interval for checking the queue. */ - protected Interval _interval; - - /** Previously throttled messages waiting to be sent. */ - protected LinkedList _queue = new LinkedList(); - } - @Inject protected ClientManager _clmgr; @Inject protected ConnectionManager _conmgr; @Inject protected PresentsDObjectMgr _omgr; @@ -1174,8 +1095,8 @@ public class PresentsClient /** Prevent the client from sending too many messages too frequently. */ protected Throttle _throttle = createIncomingMessageThrottle(); - /** Overflow data, null unless active. */ - protected Overflow _overflow; + /** Used to keep throttles around until we know the client is ready for us to apply them. */ + protected List _pendingThrottles = Lists.newArrayList(); // keep these for kicks and giggles protected int _messagesIn; @@ -1189,15 +1110,6 @@ public class PresentsClient * ended. */ protected static final long FLUSH_TIME = 7 * 60 * 1000L; - /** Maximum number of messages in allowed in a time period. */ - protected static final int DEFAULT_THROTTLE_MESSAGE_LIMIT = 100; - - /** Time period over which the message limit applies. */ - protected static final long DEFAULT_THROTTLE_TIME_LIMIT = 10*1000; - - /** Time between checks of the overflow queue. */ - protected static final int OVERFLOW_CHECK_INTERVAL = 250; - // register our message dispatchers static { _disps.put(SubscribeRequest.class, new SubscribeDispatcher()); diff --git a/src/java/com/threerings/presents/server/VariableThrottle.java b/src/java/com/threerings/presents/server/VariableThrottle.java deleted file mode 100644 index a8a43684d..000000000 --- a/src/java/com/threerings/presents/server/VariableThrottle.java +++ /dev/null @@ -1,70 +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; - -import com.samskivert.util.Throttle; - -/** - * Throttle subclass that allows changing the operation limit. - */ -public class VariableThrottle extends Throttle -{ - /** - * Creates a new updateable throttle. - */ - public VariableThrottle (int operations, long period) - { - super(operations, 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 - */ - public void updateOpLimit (int operations) - { - long[] ops = new long[operations]; - if (operations > _ops.length) { - // Copy to a larger buffer, leaving zeroes at the beginning - int lastOp = _lastOp + operations - _ops.length; - System.arraycopy(_ops, 0, ops, 0, _lastOp); - System.arraycopy(_ops, _lastOp, ops, lastOp, _ops.length - _lastOp); - - } else if (operations < _ops.length) { - // Copy to a smaller buffer, truncating older operations - int lastOp = (_lastOp + _ops.length - operations) % _ops.length; - int endCount = Math.min(_ops.length - lastOp, operations); - System.arraycopy(_ops, lastOp, ops, 0, endCount); - System.arraycopy(_ops, 0, ops, endCount, operations - endCount); - _lastOp = 0; - } - _ops = ops; - } - - @Override - public String toString () - { - return "VariableThrottle [" + _ops.length + " per " + _period + "ms]"; - } -} diff --git a/tests/src/java/com/threerings/presents/client/TestClient.java b/tests/src/java/com/threerings/presents/client/TestClient.java index add984c19..55e884446 100644 --- a/tests/src/java/com/threerings/presents/client/TestClient.java +++ b/tests/src/java/com/threerings/presents/client/TestClient.java @@ -118,7 +118,12 @@ public class TestClient { object.addListener(this); log.info("Object available: " + object); - object.setBar("lawl!"); + object.postMessage("lawl!"); + + // try blowing through our message limit + for (int ii = 0; ii < 2*Client.DEFAULT_MAX_MSG_RATE[0]+5; ii++) { + object.postMessage("ZOMG!", new Integer(ii)); + } } public void requestFailed (int oid, ObjectAccessException cause) @@ -133,8 +138,8 @@ public class TestClient { log.info("Got event [event=" + event + "]."); - // request that we log off - _client.logoff(true); +// // request that we log off +// _client.logoff(true); } // documentation inherited from interface diff --git a/tests/src/java/com/threerings/presents/server/TestVariableThrottle.java b/tests/src/java/com/threerings/presents/server/TestVariableThrottle.java deleted file mode 100644 index 23b906658..000000000 --- a/tests/src/java/com/threerings/presents/server/TestVariableThrottle.java +++ /dev/null @@ -1,71 +0,0 @@ -// $Id$ - -package com.threerings.presents.server; - -/** - * Test class to exercise the code in {@link VariableThrottle}. - */ -public class TestVariableThrottle -{ - /** - * Exercise the code in {@link VariableThrottle}. - */ - public static void main (String[] args) - { - testUpdate(4); - testUpdate(5); - testUpdate(6); - testUpdate(7); - testUpdate(8); - } - - /** - * Calls the {@link VariableThrottle#updateOpLimit} function a few times, interspersed with - * calls to add operations. Prints the operations contained in the throttle after each change. - */ - public static void testUpdate (int opCount) - { - // set up a throttle for 5 ops per millisecond - TestThrottle throttle = new TestThrottle(5, 1); - System.out.println("Testing updates with " + opCount + " operations"); - long time = 0; - for (int ii = 0; ii < opCount; ++ii) { - throttle.throttleOp(time += 1); - } - System.out.println(" " + throttle.opsToString()); - throttle.updateOpLimit(10); - for (int ii = 0; ii < opCount; ++ii) { - throttle.throttleOp(time += 1); - } - System.out.println(" " + throttle.opsToString()); - throttle.updateOpLimit(5); - System.out.println(" " + throttle.opsToString()); - for (int ii = 0; ii < opCount; ++ii) { - throttle.throttleOp(time += 1); - } - System.out.println(" " + throttle.opsToString()); - throttle.updateOpLimit(10); - System.out.println(" " + throttle.opsToString()); - } - - /** - * Constructs a string representation of all operation time stamps, starting from the oldest. - */ - public static class TestThrottle extends VariableThrottle - { - public TestThrottle (int maxOps, long period) - { - super(maxOps, period); - } - - public String opsToString () - { - String hist = String.valueOf(_ops[_lastOp]); - for (int ii = 1; ii < _ops.length; ++ii) { - long tn = _ops[(_lastOp + ii) % _ops.length]; - hist += ", " + String.valueOf(tn); - } - return hist; - } - } -}