Added support for DObject transactions which are collections of events

that are dispatched over the network at once and processed all at once.


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

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

@@ -1,9 +1,11 @@
 //
-// $Id: ClientDObjectMgr.java,v 1.11 2002/02/07 00:24:36 shaper Exp $
+// $Id: ClientDObjectMgr.java,v 1.12 2002/02/09 07:50:37 mdb Exp $
 
 package com.threerings.presents.client;
 
 import java.util.ArrayList;
+import java.util.List;
+
 import com.samskivert.util.HashIntMap;
 import com.samskivert.util.Queue;
 import com.samskivert.util.StringUtil;
@@ -153,6 +155,17 @@ public class ClientDObjectMgr
     {
         // Log.info("Dispatch event: " + event);
 
+        // if this is a compound event, we need to process its contained
+        // events in order
+        if (event instanceof CompoundEvent) {
+            List events = ((CompoundEvent)event).getEvents();
+            int ecount = events.size();
+            for (int i = 0; i < ecount; i++) {
+                dispatchEvent((DEvent)events.get(i));
+            }
+            return;
+        }
+
         // look up the object on which we're dispatching this event
         DObject target = (DObject)_ocache.get(event.getTargetOid());
         if (target == null) {

src/java/com/threerings/presents/dobj/CompoundEvent.java

@@ -0,0 +1,168 @@
+//
+// $Id: CompoundEvent.java,v 1.1 2002/02/09 07:50:37 mdb Exp $
+
+package com.threerings.presents.dobj;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.threerings.presents.io.TypedObjectFactory;
+
+/**
+ * Used to manage and submit groups of events on a collection of
+ * distributed objects in a single transaction.
+ *
+ * @see DObject#startTransaction
+ */
+public class CompoundEvent extends TypedEvent
+{
+    /** The typed object code for this event. */
+    public static final short TYPE = TYPE_BASE + 50;
+
+    /**
+     * Constructs a compound event and prepares it for operation.
+     */
+    public CompoundEvent (DObjectManager omgr)
+    {
+        super(0); // we don't have a single target object oid
+        _omgr = omgr;
+    }
+
+    /**
+     * Lets the event know that this dobject is participating in their
+     * transaction. The supplied dobject will have their transaction
+     * cleared when this event is committed or cancelled.
+     */
+    public void addObject (DObject object)
+    {
+        if (_participants == null) {
+            _participants = new ArrayList();
+        }
+        _participants.add(object);
+    }
+
+    /**
+     * Posts an event to this transaction. The event will be delivered as
+     * part of the entire transaction if it is committed or discarded if
+     * the transaction is cancelled.
+     */
+    public void postEvent (TypedEvent event)
+    {
+        _events.add(event);
+    }
+
+    /**
+     * Returns the list of events contained within this compound event.
+     */
+    public List getEvents ()
+    {
+        return _events;
+    }
+
+    /**
+     * Commits this transaction by posting this event to the distributed
+     * object event queue. All participating dobjects will have their
+     * transaction references cleared and will go back to normal
+     * operation.
+     */
+    public void commit ()
+    {
+        // first clear our participants
+        clearParticipants();
+
+        // then post this event onto the queue
+        _omgr.postEvent(this);
+    }
+
+    /**
+     * Cancels this transaction. All events posted to this transaction
+     * will be discarded.
+     */
+    public void cancel ()
+    {
+        // clear our participants
+        clearParticipants();
+        // clear our event queue in case someone holds onto us
+        _events.clear();
+    }
+
+    /**
+     * Nothing to apply here.
+     */
+    public boolean applyToObject (DObject target)
+        throws ObjectAccessException
+    {
+        return false;
+    }
+
+    // documentation inherited
+    public short getType ()
+    {
+        return TYPE;
+    }
+
+    // documentation inherited
+    public void writeTo (DataOutputStream out)
+        throws IOException
+    {
+        super.writeTo(out);
+        int ecount = _events.size();
+        for (int i = 0; i < ecount; i++) {
+            TypedEvent event = (TypedEvent)_events.get(i);
+            TypedObjectFactory.writeTo(out, event);
+        }
+    }
+
+    // documentation inherited
+    public void readFrom (DataInputStream in)
+        throws IOException
+    {
+        super.readFrom(in);
+        int ecount = in.readInt();
+        for (int i = 0; i < ecount; i++) {
+            _events.add(TypedObjectFactory.readFrom(in));
+        }
+    }
+
+    /**
+     * Calls out to all of the participating dobjects, clearing their
+     * transaction reference.
+     */
+    protected void clearParticipants ()
+    {
+        if (_participants != null) {
+            int psize = _participants.size();
+            for (int i = 0; i < psize; i++) {
+                DObject obj = (DObject)_participants.get(i);
+                obj.clearTransaction();
+            }
+            _participants = null;
+        }
+    }
+
+    // documentation inherited
+    protected void toString (StringBuffer buf)
+    {
+        buf.append("COMPOUND:");
+        super.toString(buf);
+        for (int i = 0; i < _events.size(); i++) {
+            buf.append(", ").append(_events.get(i));
+        }
+    }
+
+    /** The object manager that we'll post ourselves to when we're
+     * committed. */
+    protected DObjectManager _omgr;
+
+    /** A list of the events associated with this compound event. */
+    protected ArrayList _events = new ArrayList();
+
+    /** A list of the dobject participants in this transaction. They will
+     * be notified when we are committed or cancelled so that they can
+     * stop posting their events to us. */
+    protected ArrayList _participants;
+}

src/java/com/threerings/presents/dobj/DObject.java

@@ -1,5 +1,5 @@
 //
-// $Id: DObject.java,v 1.38 2002/02/06 22:47:28 mdb Exp $
+// $Id: DObject.java,v 1.39 2002/02/09 07:50:37 mdb Exp $
 
 package com.threerings.presents.dobj;
 
@@ -130,8 +130,8 @@ public class DObject
             // the last subscriber from our list; we also want to be sure
             // that we're still active otherwise there's no need to notify
             // our objmgr because we don't have one
-            if (--_scount == 0 && _mgr != null) {
+            if (--_scount == 0 && _omgr != null) {
-                _mgr.removedLastSubscriber(this);
+                _omgr.removedLastSubscriber(this);
             }
         }
     }
@@ -226,8 +226,7 @@ public class DObject
     public void releaseLock (String name)
     {
         // queue up a release lock event
-        ReleaseLockEvent event = new ReleaseLockEvent(_oid, name);
+        postEvent(new ReleaseLockEvent(_oid, name));
-        _mgr.postEvent(event);
     }
 
     /**
@@ -257,7 +256,7 @@ public class DObject
      */
     public void destroy ()
     {
-        _mgr.postEvent(new ObjectDestroyedEvent(_oid));
+        postEvent(new ObjectDestroyedEvent(_oid));
     }
 
     /**
@@ -410,7 +409,7 @@ public class DObject
      */
     public boolean isActive ()
     {
-        return _mgr != null;
+        return _omgr != null;
     }
 
     /**
@@ -421,9 +420,9 @@ public class DObject
      *
      * @see DObjectManager#createObject
      */
-    public void setManager (DObjectManager mgr)
+    public void setManager (DObjectManager omgr)
     {
-        _mgr = mgr;
+        _omgr = omgr;
     }
 
     /**
@@ -448,16 +447,115 @@ public class DObject
         return buf.append("]").toString();
     }
 
+    /**
+     * Begins a transaction on this distributed object. In some
+     * situations, it is desirable to cause multiple changes to
+     * distributed object fields in one unified operation. Starting a
+     * transaction causes all subsequent field modifications to be stored
+     * in a single compound event which can then be committed, dispatching
+     * and applying all included events in a single group. Additionally,
+     * the events are dispatched over the network in a single unit which
+     * can significantly enhance network efficiency.
+     *
+     * <p> When the transaction is complete, the caller must call {@link
+     * #commitTransaction} or {@link CompoundEvent#commit} to commit the
+     * transaction and release all involved objects back to their normal
+     * non-transacting state. If the caller decides not to commit their
+     * transaction, they must call {@link #cancelTransaction} or {@link
+     * CompoundEvent#cancel} to cancel the transaction and release all
+     * involved objects. Failure to do so will cause the pooch to be
+     * totally screwed.
+     *
+     * <p> Note: like all other distributed object operations,
+     * transactions are not thread safe. It is expected that a single
+     * thread will handle all distributed object operations and that
+     * thread will begin and complete a transaction before giving up
+     * control to unknown code which might try to operate on the
+     * transacting distributed object (or objects).
+     *
+     * @return the compound event that encapsulates the transaction. This
+     * can be ignored if the transaction will be limited to this object,
+     * but if it is desired that other objects be involved in the
+     * transaction, the caller will need to pass the {@link CompoundEvent}
+     * returned by this method to a call to {@link #joinTransaction} on
+     * the other objects that are involved.
+     */
+    public CompoundEvent startTransaction ()
+    {
+        if (_tevent != null) {
+            String errmsg = "Cannot start transaction on dobject that " +
+                "is already transacting [dobj=" + this + "]";
+            throw new IllegalStateException(errmsg);
+        }
+        _tevent = new CompoundEvent(_omgr);
+        _tevent.addObject(this);
+        return _tevent;
+    }
+
+    /**
+     * Causes this object to join the supplied transaction. See {@link
+     * #startTransaction} for more information on transactions.
+     */
+    public void joinTransaction (CompoundEvent event)
+    {
+        if (_tevent != null) {
+            String errmsg = "Cannot join transaction while already " +
+                "transacting [dobj=" + this + "]";
+            throw new IllegalStateException(errmsg);
+        }
+        _tevent = event;
+        _tevent.addObject(this);
+    }
+
+    /**
+     * Commits the transaction in which this distributed object is
+     * involved.
+     *
+     * @see CompoundEvent#commit
+     */
+    public void commitTransaction ()
+    {
+        if (_tevent == null) {
+            String errmsg = "Cannot commit: not involved in a transaction " +
+                "[dobj=" + this + "]";
+            throw new IllegalStateException(errmsg);
+        }
+        _tevent.commit();
+    }        
+
+    /**
+     * Cancels the transaction in which this distributed object is
+     * involved.
+     *
+     * @see CompoundEvent#cancel
+     */
+    public void cancelTransaction ()
+    {
+        if (_tevent == null) {
+            String errmsg = "Cannot cancel: not involved in a transaction " +
+                "[dobj=" + this + "]";
+            throw new IllegalStateException(errmsg);
+        }
+        _tevent.cancel();
+    }        
+
+    /**
+     * Removes this object from participation in any transaction in which
+     * it might be taking part.
+     */
+    protected void clearTransaction ()
+    {
+        _tevent = null;
+    }
+
     /**
      * Called by derived instances when an attribute setter method was
      * called.
      */
     protected void requestAttributeChange (String name, Object value)
     {
-        // generate an attribute changed event
+        // dispatch an attribute changed event
-        DEvent event = new AttributeChangedEvent(_oid, name, value);
+        postEvent(new AttributeChangedEvent(_oid, name, value));
-        // and dispatch it to our dobjmgr
-        _mgr.postEvent(event);
     }
 
     /**
@@ -465,10 +563,8 @@ public class DObject
      */
     protected void requestOidAdd (String name, int oid)
     {
-        // generate an object added event
+        // dispatch an object added event
-        DEvent event = new ObjectAddedEvent(_oid, name, oid);
+        postEvent(new ObjectAddedEvent(_oid, name, oid));
-        // and dispatch it to our dobjmgr
-        _mgr.postEvent(event);
     }
 
     /**
@@ -476,10 +572,8 @@ public class DObject
      */
     protected void requestOidRemove (String name, int oid)
     {
-        // generate an object removed event
+        // dispatch an object removed event
-        DEvent event = new ObjectRemovedEvent(_oid, name, oid);
+        postEvent(new ObjectRemovedEvent(_oid, name, oid));
-        // and dispatch it to our dobjmgr
-        _mgr.postEvent(event);
     }
 
     /**
@@ -489,11 +583,9 @@ public class DObject
     {
         try {
             DSet set = (DSet)getAttribute(name);
-            // generate an element added event
+            // dispatch an element added event
-            DEvent event = new ElementAddedEvent(
+            postEvent(new ElementAddedEvent(
-                _oid, name, elem, !set.homogenous());
+                          _oid, name, elem, !set.homogenous()));
-            // and dispatch it to our dobjmgr
-            _mgr.postEvent(event);
 
         } catch (ObjectAccessException oae) {
             Log.warning("Unable to request elementAdd [name=" + name +
@@ -506,10 +598,8 @@ public class DObject
      */
     protected void requestElementRemove (String name, Object key)
     {
-        // generate an element removed event
+        // dispatch an element removed event
-        DEvent event = new ElementRemovedEvent(_oid, name, key);
+        postEvent(new ElementRemovedEvent(_oid, name, key));
-        // and dispatch it to our dobjmgr
-        _mgr.postEvent(event);
     }
 
     /**
@@ -519,11 +609,9 @@ public class DObject
     {
         try {
             DSet set = (DSet)getAttribute(name);
-            // generate an element updated event
+            // dispatch an element updated event
-            DEvent event = new ElementUpdatedEvent(
+            postEvent(new ElementUpdatedEvent(
-                _oid, name, elem, !set.homogenous());
+                          _oid, name, elem, !set.homogenous()));
-            // and dispatch it to our dobjmgr
-            _mgr.postEvent(event);
 
         } catch (ObjectAccessException oae) {
             Log.warning("Unable to request elementUpdate [name=" + name +
@@ -531,11 +619,24 @@ public class DObject
         }
     }
 
+    /**
+     * Posts the specified event either to our dobject manager or to the
+     * compound event for which we are currently transacting.
+     */
+    protected void postEvent (TypedEvent event)
+    {
+        if (_tevent != null) {
+            _tevent.postEvent(event);
+        } else {
+            _omgr.postEvent(event);
+        }
+    }
+
     /** Our object id. */
     protected int _oid;
 
     /** A reference to our object manager. */
-    protected DObjectManager _mgr;
+    protected DObjectManager _omgr;
 
     /** A list of outstanding locks. */
     protected Object[] _locks;
@@ -548,4 +649,8 @@ public class DObject
 
     /** Our subscriber count. */
     protected int _scount;
+
+    /** The compound event associated with our transaction, if we're
+     * currently in a transaction. */
+    protected CompoundEvent _tevent;
 }

src/java/com/threerings/presents/server/PresentsDObjectMgr.java

@@ -1,10 +1,11 @@
 //
-// $Id: PresentsDObjectMgr.java,v 1.22 2002/02/02 09:53:00 mdb Exp $
+// $Id: PresentsDObjectMgr.java,v 1.23 2002/02/09 07:50:37 mdb Exp $
 
 package com.threerings.presents.server;
 
 import java.lang.reflect.*;
 import java.util.HashMap;
+import java.util.List;
 
 import com.samskivert.util.HashIntMap;
 import com.samskivert.util.Queue;
@@ -132,62 +133,79 @@ public class PresentsDObjectMgr implements RootDObjectManager
                     Log.warning("Execution unit failed [unit=" + unit + "].");
                     Log.logStackTrace(e);
                 }
-                continue;
-            }
 
-            // otherwise it's an event, so we do more complicated
+            } else if (unit instanceof CompoundEvent) {
-            // processing
+                // if this is a compound event, we need to apply each
-            DEvent event = (DEvent)unit;
+                // event indivdually
-
+                CompoundEvent event = (CompoundEvent)unit;
-            // look up the target object
+                List events = event.getEvents();
-            DObject target = (DObject)_objects.get(event.getTargetOid());
+                int ecount = events.size();
-            if (target == null) {
+                for (int i = 0; i < ecount; i++) {
-                Log.warning("Event target no longer exists " +
+                    processEvent((DEvent)events.get(i));
-                            "[event=" + event + "].");
-                continue;
-            }
-
-            // check the event's permissions
-            if (!target.checkPermissions(event)) {
-                Log.warning("Event failed permissions check " +
-                            "[event=" + event + ", target=" + target + "].");
-                continue;
-            }
-
-            try {
-                // do any internal management necessary based on this
-                // event
-                Method helper = (Method)_helpers.get(event.getClass());
-                if (helper != null) {
-                    // invoke the helper method
-                    Object rv =
-                        helper.invoke(this, new Object[] { event, target });
-                    // if helper returns false, we abort event processing
-                    if (!((Boolean)rv).booleanValue()) {
-                        continue;
-                    }
                 }
 
-                // everything's good so far, apply the event to the object
+            } else {
-                boolean notify = event.applyToObject(target);
+                // otherwise it's a regular event, so do the standard
-
+                // processing
-                // if the event returns false from applyToObject, this
+                processEvent((DEvent)unit);
-                // means it's a silent event and we shouldn't notify the
-                // listeners
-                if (notify) {
-                    target.notifyListeners(event);
-                }
-
-            } catch (Exception e) {
-                Log.warning("Failure processing event [event=" + event +
-                            ", target=" + target + "].");
-                Log.logStackTrace(e);
             }
         }
 
         Log.info("DOMGR exited.");
     }
 
+    /**
+     * Performs the processing associated with an event, notifying
+     * subscribers and the like.
+     */
+    protected void processEvent (DEvent event)
+    {
+        // look up the target object
+        DObject target = (DObject)_objects.get(event.getTargetOid());
+        if (target == null) {
+            Log.warning("Event target no longer exists " +
+                        "[event=" + event + "].");
+            return;
+        }
+
+        // check the event's permissions
+        if (!target.checkPermissions(event)) {
+            Log.warning("Event failed permissions check " +
+                        "[event=" + event + ", target=" + target + "].");
+            return;
+        }
+
+        try {
+            // do any internal management necessary based on this
+            // event
+            Method helper = (Method)_helpers.get(event.getClass());
+            if (helper != null) {
+                // invoke the helper method
+                Object rv =
+                    helper.invoke(this, new Object[] { event, target });
+                // if helper returns false, we abort event processing
+                if (!((Boolean)rv).booleanValue()) {
+                    return;
+                }
+            }
+
+            // everything's good so far, apply the event to the object
+            boolean notify = event.applyToObject(target);
+
+            // if the event returns false from applyToObject, this
+            // means it's a silent event and we shouldn't notify the
+            // listeners
+            if (notify) {
+                target.notifyListeners(event);
+            }
+
+        } catch (Exception e) {
+            Log.warning("Failure processing event [event=" + event +
+                        ", target=" + target + "].");
+            Log.logStackTrace(e);
+        }
+    }
+
     /**
      * Requests that the dobjmgr shut itself down. It will exit the event
      * processing loop which cause <code>run()</code> to return.