diff --git a/src/java/com/threerings/presents/peer/data/Lock.java b/src/java/com/threerings/presents/peer/data/Lock.java new file mode 100644 index 000000000..e58198f1e --- /dev/null +++ b/src/java/com/threerings/presents/peer/data/Lock.java @@ -0,0 +1,139 @@ +// +// $Id$ + +package com.threerings.presents.peer.data; + +import com.samskivert.util.ObjectUtil; + +import com.threerings.io.SimpleStreamableObject; + +import com.threerings.presents.dobj.DSet; + +/** + * Describes a lock held by one of the peers. + */ +public class Lock extends SimpleStreamableObject + implements DSet.Entry +{ + /** + * Identifies a locked resource by its type and type-specific identifier. + */ + public static class Name extends SimpleStreamableObject + implements Comparable + { + /** + * Creates a new lock name. + * + * @param type the type of resource. Only names with the same type will have their ids + * compared + * @param id the resource instance identifier. Can be null for resources with + * only one instance + */ + public Name (String type, Comparable id) + { + _type = type; + _id = id; + } + + public Name () + { + } + + /** + * Returns the resource type. + */ + public String getType () + { + return _type; + } + + /** + * Returns the resource instance identifier. + */ + public Comparable getId () + { + return _id; + } + + // documentation inherited from interface Comparable + public int compareTo (Object other) + { + Name oname = (Name)other; + int v1 = _type.compareTo(oname._type); + if (v1 != 0 || _id == null) { + return v1; + } + @SuppressWarnings("unchecked") int v2 = _id.compareTo(oname._id); + return v2; + } + + @Override // documentation inherited + public int hashCode () + { + return _type.hashCode() + (_id == null ? 0 : _id.hashCode()); + } + + @Override // documentation inherited + public boolean equals (Object other) + { + Name oname = (Name)other; + return _type.equals(oname._type) && + ObjectUtil.equals(_id, oname._id); + } + + protected String _type; + protected Comparable _id; + } + + /** + * Creates a newly acquired lock. + */ + public Lock (Name name, String owner) + { + _name = name; + _owner = owner; + } + + /** + * No-arg constructor for deserialization. + */ + public Lock () + { + } + + /** + * Returns the name of the locked resource. + */ + public Name getName () + { + return _name; + } + + /** + * Returns the owner of the lock. + */ + public String getOwner () + { + return _owner; + } + + // documentation inherited from interface DSet.Entry + public Comparable getKey () + { + return _name; + } + + @Override // documentation inherited + public boolean equals (Object other) + { + Lock olock = (Lock)other; + return olock._name.equals(_name) && + olock._owner.equals(_owner); + } + + /** Identifies what is locked. */ + protected Name _name; + + /** The name of the node that owns the lock. */ + protected String _owner; +} diff --git a/src/java/com/threerings/presents/peer/data/NodeObject.java b/src/java/com/threerings/presents/peer/data/NodeObject.java index 2bc88a6d8..0e631e0a6 100644 --- a/src/java/com/threerings/presents/peer/data/NodeObject.java +++ b/src/java/com/threerings/presents/peer/data/NodeObject.java @@ -31,6 +31,23 @@ import com.threerings.presents.dobj.DSet; */ public class NodeObject extends DObject { + // AUTO-GENERATED: FIELDS START + /** The field name of the clients field. */ + public static final String CLIENTS = "clients"; + + /** The field name of the locks field. */ + public static final String LOCKS = "locks"; + + /** The field name of the acquiringLock field. */ + public static final String ACQUIRING_LOCK = "acquiringLock"; + + /** The field name of the releasingLock field. */ + public static final String RELEASING_LOCK = "releasingLock"; + + /** The field name of the cacheData field. */ + public static final String CACHE_DATA = "cacheData"; + // AUTO-GENERATED: FIELDS END + /** Used for informing peers of changes to persistent data. */ public static class CacheData implements Streamable { @@ -49,20 +66,21 @@ public class NodeObject extends DObject this.cache = cache; this.data = data; } - } - // AUTO-GENERATED: FIELDS START - /** The field name of the clients field. */ - public static final String CLIENTS = "clients"; - - /** The field name of the cacheData field. */ - public static final String CACHE_DATA = "cacheData"; - // AUTO-GENERATED: FIELDS END - /** Contains information on all clients connected to this node. */ public DSet clients = new DSet(); + /** This node's view of the peer lock set. A lock is acquired only when it has been added to + * all peers' sets, and is released only when it has been removed from all peers' sets. */ + public DSet locks = new DSet(); + + /** Used to broadcast a node's desire to acquire a lock. */ + public Lock.Name acquiringLock; + + /** Used to broadcast a node's desire to release a lock. */ + public Lock.Name releasingLock; + /** A field we use to broadcast changes to possible cached data. */ public CacheData cacheData; @@ -115,6 +133,86 @@ public class NodeObject extends DObject this.clients = clone; } + /** + * Requests that the specified entry be added to the + * locks set. The set will not change until the event is + * actually propagated through the system. + */ + public void addToLocks (Lock elem) + { + requestEntryAdd(LOCKS, locks, elem); + } + + /** + * Requests that the entry matching the supplied key be removed from + * the locks set. The set will not change until the + * event is actually propagated through the system. + */ + public void removeFromLocks (Comparable key) + { + requestEntryRemove(LOCKS, locks, key); + } + + /** + * Requests that the specified entry be updated in the + * locks set. The set will not change until the event is + * actually propagated through the system. + */ + public void updateLocks (Lock elem) + { + requestEntryUpdate(LOCKS, locks, elem); + } + + /** + * Requests that the locks field be set to the + * specified value. Generally one only adds, updates and removes + * entries of a distributed set, but certain situations call for a + * complete replacement of the set value. The local value will be + * updated immediately and an event will be propagated through the + * system to notify all listeners that the attribute did + * change. Proxied copies of this object (on clients) will apply the + * value change when they received the attribute changed notification. + */ + public void setLocks (DSet value) + { + requestAttributeChange(LOCKS, value, this.locks); + @SuppressWarnings("unchecked") DSet clone = + (value == null) ? null : value.typedClone(); + this.locks = clone; + } + + /** + * Requests that the acquiringLock field be set to the + * specified value. The local value will be updated immediately and an + * event will be propagated through the system to notify all listeners + * that the attribute did change. Proxied copies of this object (on + * clients) will apply the value change when they received the + * attribute changed notification. + */ + public void setAcquiringLock (Lock.Name value) + { + Lock.Name ovalue = this.acquiringLock; + requestAttributeChange( + ACQUIRING_LOCK, value, ovalue); + this.acquiringLock = value; + } + + /** + * Requests that the releasingLock field be set to the + * specified value. The local value will be updated immediately and an + * event will be propagated through the system to notify all listeners + * that the attribute did change. Proxied copies of this object (on + * clients) will apply the value change when they received the + * attribute changed notification. + */ + public void setReleasingLock (Lock.Name value) + { + Lock.Name ovalue = this.releasingLock; + requestAttributeChange( + RELEASING_LOCK, value, ovalue); + this.releasingLock = value; + } + /** * Requests that the cacheData field be set to the * specified value. The local value will be updated immediately and an diff --git a/src/java/com/threerings/presents/peer/server/PeerManager.java b/src/java/com/threerings/presents/peer/server/PeerManager.java index d41610dba..2724a8dbf 100644 --- a/src/java/com/threerings/presents/peer/server/PeerManager.java +++ b/src/java/com/threerings/presents/peer/server/PeerManager.java @@ -24,13 +24,18 @@ package com.threerings.presents.peer.server; import java.util.ArrayList; import java.util.Date; import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; import java.util.logging.Level; import com.samskivert.io.PersistenceException; import com.samskivert.jdbc.ConnectionProvider; import com.samskivert.util.Interval; import com.samskivert.util.Invoker; +import com.samskivert.util.ObjectUtil; import com.samskivert.util.ObserverList; +import com.samskivert.util.ResultListener; +import com.samskivert.util.Tuple; import com.threerings.io.Streamable; import com.threerings.util.Name; @@ -39,13 +44,19 @@ import com.threerings.presents.client.Client; import com.threerings.presents.client.ClientObserver; import com.threerings.presents.dobj.AttributeChangedEvent; import com.threerings.presents.dobj.AttributeChangeListener; +import com.threerings.presents.dobj.EntryAddedEvent; +import com.threerings.presents.dobj.EntryRemovedEvent; +import com.threerings.presents.dobj.EntryUpdatedEvent; import com.threerings.presents.dobj.ObjectAccessException; +import com.threerings.presents.dobj.SetAdapter; import com.threerings.presents.dobj.Subscriber; import com.threerings.presents.server.ClientManager; +import com.threerings.presents.server.InvocationException; import com.threerings.presents.server.PresentsClient; import com.threerings.presents.server.PresentsServer; import com.threerings.presents.peer.data.ClientInfo; +import com.threerings.presents.peer.data.Lock; import com.threerings.presents.peer.data.NodeObject; import com.threerings.presents.peer.net.PeerBootstrapData; import com.threerings.presents.peer.net.PeerCreds; @@ -181,6 +192,113 @@ public class PeerManager creds.getPassword()); } + /** + * Attempts to acquire the specified lock amongst our peers. When the peers complete the + * process of resolution, the supplied listener will be notified with the name of the lock's + * owner, which will either be this or another node (if that node preempted this one). If + * another node holds the lock, it will not release it before this node has a chance to make + * a request. + */ + public void acquireLock (final Lock.Name name, final ResultListener listener) + { + // wait for any resolution to end + queryLock(name, new ResultListener() { + public void requestCompleted (String result) { + if (result == null) { + // the lock is free; we will attempt to acquire it + continueAcquiringLock(name, listener); + } else { + // another node has acquired the lock + listener.requestCompleted(result); + } + } + public void requestFailed (Exception cause) { + listener.requestFailed(cause); + } + }); + } + + /** + * Attempts to release the specified lock. The supplied listener will receive the lock's + * owner (null if the lock was successfully released, or the node's name if + * the release was cancelled using {@link #reacquireLock}). + */ + public void releaseLock (final Lock.Name name, final ResultListener listener) + { + // make sure we actually hold the lock + Tuple result = queryLock(name, false); + if (!result.left) { + log.warning("Attempted to release lock in process of resolution [name=" + name + "]."); + } else if (!_nodeName.equals(result.right)) { + log.warning("Attempted to release lock not owned [name=" + name + ", owner=" + + result.right + "]."); + } + + // announce our desire to drop the lock + _nodeobj.setReleasingLock(name); + + // wait for all peers to agree + new LockResolutionListener(name, true) { + public void lockResolved (String owner) { + if (_acquiringLocks.contains(name)) { + queryLock(name, listener); + return; + } else if (owner == null) { + _nodeobj.removeFromLocks(name); + } else if (!owner.equals(_nodeName)) { + log.warning("Someone acquired the lock before we were done releasing it?! " + + "[name=" + name + ", owner=" + owner + "]."); + } + listener.requestCompleted(owner); + } + }.add(); + } + + /** + * Reacquires a lock for which we have called {@link #releaseLock}, but have not yet received + * a response. + */ + public void reacquireLock (final Lock.Name name, final ResultListener listener) + { + // broadcast our desire to reacquire + _nodeobj.setAcquiringLock(name); + + // remember the name + _acquiringLocks.add(name); + + // wait for resolution + new LockResolutionListener(name, false) { + public void lockResolved (String owner) { + _acquiringLocks.remove(name); + if (!_nodeName.equals(owner)) { + log.warning("Failed to reacquire lock? [name=" + name + ", owner=" + + owner + "]."); + } + listener.requestCompleted(owner); + } + }.add(); + } + + /** + * Attempts to determine the owner of the specified lock, if any. + * + * @param listener a listener to receive the name of the lock's owner (or null + * for none) once any necessary resolution has been performed + */ + public void queryLock (Lock.Name name, final ResultListener listener) + { + Tuple result = queryLock(name, false); + if (result.left) { + listener.requestCompleted(result.right); + } else { + new LockResolutionListener(name, false) { + public void lockResolved (String owner) { + listener.requestCompleted(owner); + } + }.add(); + } + } + // documentation inherited from interface ClientManager.ClientObserver public void clientSessionDidStart (PresentsClient client) { @@ -368,6 +486,53 @@ public class PeerManager return new PeerNode(record); } + /** + * Continues the process of acquiring a lock once it has been established that the lock is + * free for the taking. + */ + protected void continueAcquiringLock ( + final Lock.Name name, final ResultListener listener) + { + // announce our desire for the lock + _nodeobj.setAcquiringLock(name); + + // remember the name + _acquiringLocks.add(name); + + // wait for all other nodes to resolve before adding ourself to seal the deal + new LockResolutionListener(name, true) { + public void lockResolved (String owner) { + _acquiringLocks.remove(name); + if (_nodeName.equals(owner)) { + _nodeobj.addToLocks(new Lock(name, _nodeName)); + } + listener.requestCompleted(owner); + } + }.add(); + } + + /** + * Attempts to determine the owner of the specified lock, if any. + * + * @param peersOnly if true, only check the peers' locks (that is, exclude this node from + * the search) + * @return a tuple containing as its left value a boolean indicating whether the lock's + * state has been resolved. If the value is true, the right value contains + * the lock's owner (or null if the lock is free) + */ + protected Tuple queryLock (Lock.Name name, boolean peersOnly) + { + Iterator it = _peers.values().iterator(); + Lock lock = (peersOnly && it.hasNext()) ? + it.next().nodeobj.locks.get(name) : _nodeobj.locks.get(name); + while (it.hasNext()) { + if (!ObjectUtil.equals(lock, it.next().nodeobj.locks.get(name))) { + return new Tuple(false, null); + } + } + return new Tuple(true, (lock == null) ? null : lock.getOwner()); + } + /** * Contains all runtime information for one of our peer nodes. */ @@ -464,7 +629,7 @@ public class PeerManager // documentation inherited from interface ClientObserver public void clientDidLogoff (Client client) { - // TODO: clean things up? + nodeobj.removeListener(this); } // documentation inherited from interface ClientObserver @@ -477,8 +642,9 @@ public class PeerManager public void objectAvailable (NodeObject object) { nodeobj = object; + + // listen for lock and cache updates nodeobj.addListener(this); - // TODO: stuff! } // documentation inherited from interface Subscriber @@ -491,17 +657,113 @@ public class PeerManager // documentation inherited from interface AttributeChangeListener public void attributeChanged (AttributeChangedEvent event) { - if (NodeObject.CACHE_DATA.equals(event.getName())) { - NodeObject.CacheData cd = (NodeObject.CacheData)event.getValue(); - changedCacheData(cd.cache, cd.data); + String name = event.getName(); + if (name.equals(NodeObject.ACQUIRING_LOCK)) { + if (_acquiringLocks.contains(nodeobj.acquiringLock) && + _nodeName.compareTo(_record.nodeName) < 0) { + return; // this node has priority + } + Lock olock = _nodeobj.locks.get(nodeobj.acquiringLock), + nlock = new Lock(nodeobj.acquiringLock, _record.nodeName); + if (olock == null) { + _nodeobj.addToLocks(nlock); + } else if (_record.nodeName.compareTo(olock.getOwner()) < 0) { + _nodeobj.updateLocks(nlock); + } + } else if (name.equals(NodeObject.RELEASING_LOCK)) { + Lock lock = _nodeobj.locks.get(nodeobj.releasingLock); + if (lock != null && lock.getOwner().equals(_record.nodeName)) { + _nodeobj.removeFromLocks(nodeobj.releasingLock); + } else { + log.warning("Node attempting to release lock not held? [node=" + + _record.nodeName + ", name=" + nodeobj.releasingLock + ", lock=" + + lock + "]."); + } + } else if (name.equals(NodeObject.CACHE_DATA)) { + changedCacheData(nodeobj.cacheData.cache, nodeobj.cacheData.data); } } - + protected NodeRecord _record; protected Client _client; protected long _lastConnectStamp; } + /** + * Listens to all {@link NodeObject}s (possibly including our own) to determine when the lock + * state has been agreed upon. + */ + protected abstract class LockResolutionListener extends SetAdapter + { + public LockResolutionListener (Lock.Name name, boolean peersOnly) + { + _name = name; + _peersOnly = peersOnly; + } + + public void add () + { + for (PeerNode peer : _peers.values()) { + peer.nodeobj.addListener(this); + } + if (!_peersOnly) { + _nodeobj.addListener(this); + } + } + + @Override // documentation inherited + public void entryAdded (EntryAddedEvent event) + { + if (event.getName().equals(NodeObject.LOCKS)) { + checkLockResolved(); + } + } + + @Override // documentation inherited + public void entryRemoved (EntryRemovedEvent event) + { + if (event.getName().equals(NodeObject.LOCKS)) { + checkLockResolved(); + } + } + + @Override // documentation inherited + public void entryUpdated (EntryUpdatedEvent event) + { + if (event.getName().equals(NodeObject.LOCKS)) { + checkLockResolved(); + } + } + + protected void checkLockResolved () + { + Tuple result = queryLock(_name, _peersOnly); + if (!result.left) { + return; + } + + lockResolved(result.right); + + for (PeerNode peer : _peers.values()) { + peer.nodeobj.removeListener(this); + } + if (!_peersOnly) { + _nodeobj.removeListener(this); + } + } + + /** + * Called when the owner of the lock has been agreed upon. + */ + protected abstract void lockResolved (String owner); + + /** The name of the lock. */ + protected Lock.Name _name; + + /** If true, wait for all nodes except this one to agree. */ + protected boolean _peersOnly; + } + protected String _nodeName, _hostName, _publicHostName, _sharedSecret; protected int _port; protected Invoker _invoker; @@ -512,4 +774,7 @@ public class PeerManager /** Our stale cache observers. */ protected HashMap> _cacheobs = new HashMap>(); + + /** Contains the names of any locks we are currently trying to acquire. */ + protected HashSet _acquiringLocks = new HashSet(); }