From 52023608c8b5f5b78b2b4fe2f73a0e634f1047b8 Mon Sep 17 00:00:00 2001 From: Charlie Groves Date: Thu, 16 Jun 2011 03:37:08 +0000 Subject: [PATCH] Allow downstream and upstream messages to be compounded A client can collect a bunch of messages into a CompoundUpstreamMessage, which are unpacked and dispatched in order by the server. The server collects any messages sent while processing that compound message into a CompoundDownstreamMessage, which is unpacked and dispatched by the client. This allows the client to indicate that it's going to send enough messages that it'd blow out the throttle, but that it's doing so intentionally. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6659 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../com/threerings/presents/client/Client.as | 16 +++- .../presents/client/ClientDObjectMgr.as | 5 ++ .../presents/client/Communicator.as | 29 +++++++ .../presents/net/CompoundDownstreamMessage.as | 38 +++++++++ .../presents/net/CompoundUpstreamMessage.as | 41 +++++++++ .../threerings/presents/net/PongResponse.as | 5 -- .../presents/client/ClientDObjectMgr.java | 76 +++++++++-------- .../net/CompoundDownstreamMessage.java | 40 +++++++++ .../presents/net/CompoundUpstreamMessage.java | 42 ++++++++++ .../presents/server/PresentsSession.java | 84 ++++++++++++++++++- 10 files changed, 333 insertions(+), 43 deletions(-) create mode 100644 src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as create mode 100644 src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as create mode 100644 src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java create mode 100644 src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java diff --git a/src/main/as/com/threerings/presents/client/Client.as b/src/main/as/com/threerings/presents/client/Client.as index 18e48fd0b..9e342bb54 100644 --- a/src/main/as/com/threerings/presents/client/Client.as +++ b/src/main/as/com/threerings/presents/client/Client.as @@ -20,7 +20,6 @@ // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA package com.threerings.presents.client { - import flash.events.EventDispatcher; import flash.events.TimerEvent; @@ -280,6 +279,21 @@ public class Client extends EventDispatcher return true; } + /** + * Runs the given function while collecting any generated messages in a CompoundEvent, which + * will be sent when the function returns. + */ + public function inCompoundMessage (run :Function) :void + { + _comm.startCompoundMessage(); + try { + run(); + } finally { + _comm.finishCompoundMessage(); + } + + } + /** * Create the client property object to pass to {@link ObjectInputStream}. Subclasses may * add references useful during e.g. deserialization. diff --git a/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as b/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as index 674889911..5d863e63d 100644 --- a/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as +++ b/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as @@ -20,6 +20,7 @@ // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA package com.threerings.presents.client { +import com.threerings.presents.net.CompoundDownstreamMessage; import flash.events.Event; import flash.events.TimerEvent; @@ -175,6 +176,10 @@ public class ClientDObjectMgr } else if (msg is UpdateThrottleMessage) { _client.setOutgoingMessageThrottle((msg as UpdateThrottleMessage).messagesPerSec); + } else if (msg is CompoundDownstreamMessage) { + for each (var submsg :DownstreamMessage in CompoundDownstreamMessage(msg).msgs) { + processMessage(submsg); + } } } diff --git a/src/main/as/com/threerings/presents/client/Communicator.as b/src/main/as/com/threerings/presents/client/Communicator.as index e0636339b..88b968b55 100644 --- a/src/main/as/com/threerings/presents/client/Communicator.as +++ b/src/main/as/com/threerings/presents/client/Communicator.as @@ -20,6 +20,7 @@ // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA package com.threerings.presents.client { +import com.threerings.presents.net.CompoundUpstreamMessage; import flash.events.Event; import flash.events.IOErrorEvent; @@ -75,6 +76,10 @@ public class Communicator public function postMessage (msg :UpstreamMessage) :void { + if (_compound != null) { + _compound.msgs.push(msg); + return; + } _outq.push(msg); if (_writer != null) { sendPendingMessages(null); @@ -109,6 +114,23 @@ public class Communicator return false; } + public function startCompoundMessage () :void + { + if (_compound == null) { + _compound = new CompoundUpstreamMessage(); + } + _compoundDepth++; + } + + public function finishCompoundMessage () :void + { + if (--_compoundDepth == 0) { + var toSend :CompoundUpstreamMessage = _compound; + _compound = null; + postMessage(toSend); + } + } + /** * Attempts to logon on using the port at the specified index. */ @@ -383,6 +405,13 @@ public class Communicator protected var _writer :Timer; protected var _notedThrottle :Boolean = false; + protected var _compound :CompoundUpstreamMessage; + + /** + * The count of startCompoundMessage calls that have occurred without a finishCompoundMessage + */ + protected var _compoundDepth :int; + protected const log :Log = Log.getLog(this); /** The current port we'll try to connect to. */ diff --git a/src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as b/src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as new file mode 100644 index 000000000..10695fae2 --- /dev/null +++ b/src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as @@ -0,0 +1,38 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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.streamers.ArrayStreamer; +import com.threerings.io.TypedArray; +import com.threerings.io.ObjectInputStream; + +public class CompoundDownstreamMessage extends DownstreamMessage +{ + public var msgs :TypedArray = TypedArray.create(DownstreamMessage); + + override public function readObject (ins :ObjectInputStream) :void + { + super.readObject(ins); + + msgs = ins.readField(ArrayStreamer.INSTANCE); + } +} +} diff --git a/src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as b/src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as new file mode 100644 index 000000000..e083d33bb --- /dev/null +++ b/src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as @@ -0,0 +1,41 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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; +import com.threerings.io.ObjectOutputStream; +import com.threerings.io.TypedArray; +import com.threerings.io.streamers.ArrayStreamer; + +import com.threerings.presents.net.UpstreamMessage; + +public class CompoundUpstreamMessage extends UpstreamMessage +{ + public var msgs :TypedArray = TypedArray.create(UpstreamMessage); + + override public function writeObject (out :ObjectOutputStream) :void + { + super.writeObject(out); + out.writeField(msgs, ArrayStreamer.INSTANCE); + } +} +} diff --git a/src/main/as/com/threerings/presents/net/PongResponse.as b/src/main/as/com/threerings/presents/net/PongResponse.as index f6e760cd5..5d7f14af1 100644 --- a/src/main/as/com/threerings/presents/net/PongResponse.as +++ b/src/main/as/com/threerings/presents/net/PongResponse.as @@ -31,11 +31,6 @@ import com.threerings.io.Streamable; public class PongResponse extends DownstreamMessage { - public function PongResponse () - { - super(); - } - public function getPackStamp () :Long { return _packStamp; diff --git a/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java b/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java index f342f2cf6..1aa9a5e5f 100644 --- a/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java +++ b/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java @@ -47,6 +47,8 @@ import com.threerings.presents.dobj.ObjectDestroyedEvent; import com.threerings.presents.dobj.Subscriber; import com.threerings.presents.net.BootstrapData; import com.threerings.presents.net.BootstrapNotification; +import com.threerings.presents.net.CompoundDownstreamMessage; +import com.threerings.presents.net.DownstreamMessage; import com.threerings.presents.net.EventNotification; import com.threerings.presents.net.FailureResponse; import com.threerings.presents.net.ForwardEventRequest; @@ -184,44 +186,52 @@ public class ClientDObjectMgr // process the next event on our queue Object obj; if ((obj = _actions.getNonBlocking()) != null) { - // do the proper thing depending on the object - if (obj instanceof EventNotification) { - dispatchEvent(((EventNotification)obj).getEvent()); + dispatchAction(obj); + } + } - } else if (obj instanceof BootstrapNotification) { - BootstrapData data = ((BootstrapNotification)obj).getData(); - _client.gotBootstrap(data, this); + protected void dispatchAction (Object obj) + { + if (obj instanceof EventNotification) { + dispatchEvent(((EventNotification)obj).getEvent()); - } else if (obj instanceof ObjectResponse) { - registerObjectAndNotify((ObjectResponse)obj); + } else if (obj instanceof BootstrapNotification) { + BootstrapData data = ((BootstrapNotification)obj).getData(); + _client.gotBootstrap(data, this); - } else if (obj instanceof UnsubscribeResponse) { - int oid = ((UnsubscribeResponse)obj).getOid(); - if (_dead.remove(oid) == null) { - log.warning("Received unsub ACK from unknown object", "oid", oid); - } + } else if (obj instanceof ObjectResponse) { + registerObjectAndNotify((ObjectResponse)obj); - } else if (obj instanceof FailureResponse) { - notifyFailure(((FailureResponse)obj).getOid(), ((FailureResponse)obj).getMessage()); - - } else if (obj instanceof PongResponse) { - _client.gotPong((PongResponse)obj); - - } else if (obj instanceof UpdateThrottleMessage) { - UpdateThrottleMessage upmsg = (UpdateThrottleMessage)obj; - _client.setOutgoingMessageThrottle(upmsg.messagesPerSec); - - } else if (obj instanceof ObjectAction) { - ObjectAction act = (ObjectAction)obj; - if (act.subscribe) { - doSubscribe(act); - } else { - doUnsubscribe(act.oid, act.target); - } - - } else { - log.warning("Unknown action", "action", obj); + } else if (obj instanceof UnsubscribeResponse) { + int oid = ((UnsubscribeResponse)obj).getOid(); + if (_dead.remove(oid) == null) { + log.warning("Received unsub ACK from unknown object", "oid", oid); } + + } else if (obj instanceof FailureResponse) { + notifyFailure(((FailureResponse)obj).getOid(), ((FailureResponse)obj).getMessage()); + + } else if (obj instanceof PongResponse) { + _client.gotPong((PongResponse)obj); + + } else if (obj instanceof UpdateThrottleMessage) { + UpdateThrottleMessage upmsg = (UpdateThrottleMessage)obj; + _client.setOutgoingMessageThrottle(upmsg.messagesPerSec); + + } else if (obj instanceof ObjectAction) { + ObjectAction act = (ObjectAction)obj; + if (act.subscribe) { + doSubscribe(act); + } else { + doUnsubscribe(act.oid, act.target); + } + + } else if (obj instanceof CompoundDownstreamMessage) { + for (DownstreamMessage submsg : ((CompoundDownstreamMessage)obj).msgs) { + dispatchAction(submsg); + } + } else { + log.warning("Unknown action", "action", obj); } } diff --git a/src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java b/src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java new file mode 100644 index 000000000..036d3f6d6 --- /dev/null +++ b/src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java @@ -0,0 +1,40 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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 java.util.List; + +import com.google.common.collect.Lists; + +/** + * Groups messages to be dispatched without triggering the message throttle. + */ +public class CompoundDownstreamMessage extends DownstreamMessage +{ + public List msgs = Lists.newArrayList(); + + @Override + public String toString () + { + return "[type=COMPOUND, msgid=" + messageId + "]"; + } +} diff --git a/src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java b/src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java new file mode 100644 index 000000000..81bf3dc08 --- /dev/null +++ b/src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java @@ -0,0 +1,42 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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 java.util.List; + +import com.google.common.collect.Lists; + +/** + * Groups messages to be dispatched without triggering the message throttle. Any messages sent to + * a client while processing an upstream message will be sent back en masse in a + * CompoundDownstreamMessage. + */ +public class CompoundUpstreamMessage extends UpstreamMessage +{ + public List msgs = Lists.newArrayList(); + + @Override + public String toString () + { + return "[type=COMPOUND, msgid=" + messageId + "]"; + } +} diff --git a/src/main/java/com/threerings/presents/server/PresentsSession.java b/src/main/java/com/threerings/presents/server/PresentsSession.java index af081a6a8..324b5f3b2 100644 --- a/src/main/java/com/threerings/presents/server/PresentsSession.java +++ b/src/main/java/com/threerings/presents/server/PresentsSession.java @@ -54,6 +54,8 @@ import com.threerings.presents.dobj.ProxySubscriber; import com.threerings.presents.net.AuthRequest; import com.threerings.presents.net.BootstrapData; import com.threerings.presents.net.BootstrapNotification; +import com.threerings.presents.net.CompoundDownstreamMessage; +import com.threerings.presents.net.CompoundUpstreamMessage; import com.threerings.presents.net.Credentials; import com.threerings.presents.net.DownstreamMessage; import com.threerings.presents.net.EventNotification; @@ -70,6 +72,7 @@ import com.threerings.presents.net.TransmitDatagramsRequest; 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.PresentsConnection; import com.threerings.nio.conman.Connection; @@ -433,8 +436,6 @@ public class PresentsSession // from interface Connection.MessageHandler public void handleMessage (Message message) { - _messagesIn++; // count 'em up! - // if the client has been getting crazy with the cheeze whiz, stick a fork in them; the // first time through we end our session, subsequently _throttle is null and we just drop // any messages that come in until we've fully shutdown @@ -447,6 +448,16 @@ public class PresentsSession handleThrottleExceeded(); } + dispatchMessage(message); + } + + /** + * Processes a message without throttling. + */ + protected void dispatchMessage (Message message) + { + _messagesIn++; // count 'em up! + // we dispatch to a message dispatcher that is specialized for the particular class of // message that we received MessageDispatcher disp = _disps.get(message.getClass()); @@ -464,8 +475,7 @@ public class PresentsSession * applied to this client. The old client object will not yet have been destroyed, so any final * events can be sent along prior to the new object being put into effect. */ - protected void clientObjectWillChange ( - ClientObject oldClobj, ClientObject newClobj) + protected void clientObjectWillChange (ClientObject oldClobj, ClientObject newClobj) { } @@ -901,6 +911,31 @@ public class PresentsSession }); } + /** + * Collects downstream messages in a compound message until finishCompoundMessage is called. + */ + protected void startCompoundMessage () + { + if (_compound == null) { + _compound = new CompoundDownstreamMessage(); + } + _compoundDepth++; + } + + /** + * Sends the compound message created in startCompoundMessage. + */ + protected void finishCompoundMessage () + { + if (--_compoundDepth == 0) { + CompoundDownstreamMessage downstream = _compound; + _compound = null; + if (!downstream.msgs.isEmpty()) { + postMessage(downstream, null); + } + } + } + /** Queues a message for delivery to the client. */ protected boolean postMessage (DownstreamMessage msg, PresentsConnection expect) { @@ -914,6 +949,11 @@ public class PresentsSession return false; } + if (_compound != null) { + _compound.msgs.add(msg); + return true; + } + // make sure we have a connection at all if (conn != null) { conn.postMessage(msg); @@ -1124,6 +1164,29 @@ public class PresentsSession } } + /** + * Processes compound messages. + */ + protected static class CompoundDispatcher implements MessageDispatcher + { + public void dispatch (final PresentsSession client, Message msg) + { + // Compound downstream messages sent while dispatching the upstream compound message + client._omgr.postRunnable(new Runnable() { + public void run () { + client.startCompoundMessage(); + }}); + for (UpstreamMessage submsg : ((CompoundUpstreamMessage)msg).msgs) { + client.dispatchMessage(submsg); + } + // Send any messages produced en masse now that we've finished dispatching + client._omgr.postRunnable(new Runnable() { + public void run () { + client.finishCompoundMessage(); + }}); + } + } + /** * Processes unsubscribe requests. */ @@ -1236,6 +1299,18 @@ public class PresentsSession protected ClientObject _clobj; protected IntMap _subscrips = IntMaps.newHashIntMap(); + /** + * Message in which we're currently compounding messages to send, or null if we're sending them + * straight on. + */ + protected CompoundDownstreamMessage _compound; + + /** + * The count of startCompoundMessage calls that have occurred without a finishCompoundMessage. + * _compound won't be set until this reaches 0 again. + */ + protected int _compoundDepth; + /** The Oids of objects that have been destroyed while we were subscribed. */ protected HashSet _destroyedSubs = Sets.newHashSet(); protected ClassLoader _loader; @@ -1282,5 +1357,6 @@ public class PresentsSession _disps.put(TransmitDatagramsRequest.class, new TransmitDatagramsDispatcher()); _disps.put(ThrottleUpdatedMessage.class, new ThrottleUpdatedDispatcher()); _disps.put(LogoffRequest.class, new LogoffDispatcher()); + _disps.put(CompoundUpstreamMessage.class, new CompoundDispatcher()); } }