diff --git a/src/java/com/threerings/presents/peer/client/PeerService.java b/src/java/com/threerings/presents/peer/client/PeerService.java new file mode 100644 index 000000000..c855115ff --- /dev/null +++ b/src/java/com/threerings/presents/peer/client/PeerService.java @@ -0,0 +1,39 @@ +// +// $Id: CrowdPeerService.java 4509 2007-01-24 00:22:07Z dhoover $ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2006 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.peer.client; + +import com.threerings.presents.client.Client; +import com.threerings.presents.client.InvocationService; + +import com.threerings.presents.peer.data.NodeObject.Lock; + +/** + * Defines requests made from one peer to another. + */ +public interface PeerService extends InvocationService +{ + /** + * Informs the node that the sending peer ratifies its acquisition or release of the specified + * lock. + */ + public void ratifyLockAction (Client client, Lock lock, boolean acquire); +} diff --git a/src/java/com/threerings/presents/peer/data/Lock.java b/src/java/com/threerings/presents/peer/data/Lock.java deleted file mode 100644 index e58198f1e..000000000 --- a/src/java/com/threerings/presents/peer/data/Lock.java +++ /dev/null @@ -1,139 +0,0 @@ -// -// $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 0e631e0a6..a597b352d 100644 --- a/src/java/com/threerings/presents/peer/data/NodeObject.java +++ b/src/java/com/threerings/presents/peer/data/NodeObject.java @@ -21,6 +21,9 @@ package com.threerings.presents.peer.data; +import com.samskivert.util.ObjectUtil; + +import com.threerings.io.SimpleStreamableObject; import com.threerings.io.Streamable; import com.threerings.presents.dobj.DObject; @@ -32,6 +35,9 @@ import com.threerings.presents.dobj.DSet; public class NodeObject extends DObject { // AUTO-GENERATED: FIELDS START + /** The field name of the peerService field. */ + public static final String PEER_SERVICE = "peerService"; + /** The field name of the clients field. */ public static final String CLIENTS = "clients"; @@ -48,8 +54,60 @@ public class NodeObject extends DObject public static final String CACHE_DATA = "cacheData"; // AUTO-GENERATED: FIELDS END + /** Identifies a locked resource. */ + public static class Lock extends SimpleStreamableObject + implements Comparable, DSet.Entry + { + /** The resource type. Only resources of the same type will have their ids compared. */ + public String type; + + /** The resource identifier, which can be null for singleton resources. */ + public Comparable id; + + public Lock () + { + } + + public Lock (String type, Comparable id) + { + this.type = type; + this.id = id; + } + + // documentation inherited from interface Comparable + public int compareTo (Object other) + { + Lock olock = (Lock)other; + int v1 = type.compareTo(olock.type); + if (v1 != 0 || id == null) { + return v1; + } + @SuppressWarnings("unchecked") int v2 = id.compareTo(olock.id); + return v2; + } + + // documentation inherited from interface DSet.Entry + public Comparable getKey () + { + return this; + } + + @Override // documentation inherited + public int hashCode () + { + return type.hashCode() + (id == null ? 0 : id.hashCode()); + } + + @Override // documentation inherited + public boolean equals (Object other) + { + Lock olock = (Lock)other; + return type.equals(olock.type) && ObjectUtil.equals(id, olock.id); + } + } + /** Used for informing peers of changes to persistent data. */ - public static class CacheData implements Streamable + public static class CacheData extends SimpleStreamableObject { /** The cache that should be purged. */ public String cache; @@ -68,23 +126,41 @@ public class NodeObject extends DObject } } + /** The service used to make requests of the node. */ + public PeerMarshaller peerService; + /** 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. */ + /** The set of locks held by this node. */ public DSet locks = new DSet(); /** Used to broadcast a node's desire to acquire a lock. */ - public Lock.Name acquiringLock; + public Lock acquiringLock; /** Used to broadcast a node's desire to release a lock. */ - public Lock.Name releasingLock; + public Lock releasingLock; /** A field we use to broadcast changes to possible cached data. */ public CacheData cacheData; // AUTO-GENERATED: METHODS START + /** + * Requests that the peerService 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 setPeerService (PeerMarshaller value) + { + PeerMarshaller ovalue = this.peerService; + requestAttributeChange( + PEER_SERVICE, value, ovalue); + this.peerService = value; + } + /** * Requests that the specified entry be added to the * clients set. The set will not change until the event is @@ -138,7 +214,7 @@ public class NodeObject extends DObject * locks set. The set will not change until the event is * actually propagated through the system. */ - public void addToLocks (Lock elem) + public void addToLocks (NodeObject.Lock elem) { requestEntryAdd(LOCKS, locks, elem); } @@ -158,7 +234,7 @@ public class NodeObject extends DObject * locks set. The set will not change until the event is * actually propagated through the system. */ - public void updateLocks (Lock elem) + public void updateLocks (NodeObject.Lock elem) { requestEntryUpdate(LOCKS, locks, elem); } @@ -173,10 +249,10 @@ public class NodeObject extends DObject * 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) + public void setLocks (DSet value) { requestAttributeChange(LOCKS, value, this.locks); - @SuppressWarnings("unchecked") DSet clone = + @SuppressWarnings("unchecked") DSet clone = (value == null) ? null : value.typedClone(); this.locks = clone; } @@ -189,9 +265,9 @@ public class NodeObject extends DObject * clients) will apply the value change when they received the * attribute changed notification. */ - public void setAcquiringLock (Lock.Name value) + public void setAcquiringLock (NodeObject.Lock value) { - Lock.Name ovalue = this.acquiringLock; + NodeObject.Lock ovalue = this.acquiringLock; requestAttributeChange( ACQUIRING_LOCK, value, ovalue); this.acquiringLock = value; @@ -205,9 +281,9 @@ public class NodeObject extends DObject * clients) will apply the value change when they received the * attribute changed notification. */ - public void setReleasingLock (Lock.Name value) + public void setReleasingLock (NodeObject.Lock value) { - Lock.Name ovalue = this.releasingLock; + NodeObject.Lock ovalue = this.releasingLock; requestAttributeChange( RELEASING_LOCK, value, ovalue); this.releasingLock = value; diff --git a/src/java/com/threerings/presents/peer/data/PeerMarshaller.java b/src/java/com/threerings/presents/peer/data/PeerMarshaller.java new file mode 100644 index 000000000..fa1bca836 --- /dev/null +++ b/src/java/com/threerings/presents/peer/data/PeerMarshaller.java @@ -0,0 +1,50 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2006 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.peer.data; + +import com.threerings.presents.client.Client; +import com.threerings.presents.data.InvocationMarshaller; +import com.threerings.presents.dobj.InvocationResponseEvent; +import com.threerings.presents.peer.client.PeerService; +import com.threerings.presents.peer.data.NodeObject; + +/** + * Provides the implementation of the {@link PeerService} interface + * that marshalls the arguments and delivers the request to the provider + * on the server. Also provides an implementation of the response listener + * interfaces that marshall the response arguments and deliver them back + * to the requesting client. + */ +public class PeerMarshaller extends InvocationMarshaller + implements PeerService +{ + /** The method id used to dispatch {@link #ratifyLockAction} requests. */ + public static final int RATIFY_LOCK_ACTION = 1; + + // from interface PeerService + public void ratifyLockAction (Client arg1, NodeObject.Lock arg2, boolean arg3) + { + sendRequest(arg1, RATIFY_LOCK_ACTION, new Object[] { + arg2, Boolean.valueOf(arg3) + }); + } +} diff --git a/src/java/com/threerings/presents/peer/server/PeerDispatcher.java b/src/java/com/threerings/presents/peer/server/PeerDispatcher.java new file mode 100644 index 000000000..db569fbe9 --- /dev/null +++ b/src/java/com/threerings/presents/peer/server/PeerDispatcher.java @@ -0,0 +1,71 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2006 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.peer.server; + +import com.threerings.presents.client.Client; +import com.threerings.presents.data.ClientObject; +import com.threerings.presents.data.InvocationMarshaller; +import com.threerings.presents.peer.client.PeerService; +import com.threerings.presents.peer.data.NodeObject; +import com.threerings.presents.peer.data.PeerMarshaller; +import com.threerings.presents.server.InvocationDispatcher; +import com.threerings.presents.server.InvocationException; + +/** + * Dispatches requests to the {@link PeerProvider}. + */ +public class PeerDispatcher extends InvocationDispatcher +{ + /** + * Creates a dispatcher that may be registered to dispatch invocation + * service requests for the specified provider. + */ + public PeerDispatcher (PeerProvider provider) + { + this.provider = provider; + } + + // from InvocationDispatcher + public InvocationMarshaller createMarshaller () + { + return new PeerMarshaller(); + } + + @SuppressWarnings("unchecked") // from InvocationDispatcher + public void dispatchRequest ( + ClientObject source, int methodId, Object[] args) + throws InvocationException + { + switch (methodId) { + case PeerMarshaller.RATIFY_LOCK_ACTION: + ((PeerProvider)provider).ratifyLockAction( + source, + (NodeObject.Lock)args[0], ((Boolean)args[1]).booleanValue() + ); + return; + + default: + super.dispatchRequest(source, methodId, args); + return; + } + } +} diff --git a/src/java/com/threerings/presents/peer/server/PeerManager.java b/src/java/com/threerings/presents/peer/server/PeerManager.java index 2724a8dbf..df40cd6be 100644 --- a/src/java/com/threerings/presents/peer/server/PeerManager.java +++ b/src/java/com/threerings/presents/peer/server/PeerManager.java @@ -24,7 +24,6 @@ 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; @@ -35,6 +34,7 @@ import com.samskivert.util.Invoker; import com.samskivert.util.ObjectUtil; import com.samskivert.util.ObserverList; import com.samskivert.util.ResultListener; +import com.samskivert.util.ResultListenerList; import com.samskivert.util.Tuple; import com.threerings.io.Streamable; @@ -42,13 +42,16 @@ import com.threerings.util.Name; import com.threerings.presents.client.Client; import com.threerings.presents.client.ClientObserver; +import com.threerings.presents.data.ClientObject; 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.ObjectDeathListener; +import com.threerings.presents.dobj.ObjectDestroyedEvent; +import com.threerings.presents.dobj.SetListener; import com.threerings.presents.dobj.Subscriber; import com.threerings.presents.server.ClientManager; import com.threerings.presents.server.InvocationException; @@ -56,8 +59,9 @@ 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.data.NodeObject.Lock; +import com.threerings.presents.peer.data.PeerMarshaller; import com.threerings.presents.peer.net.PeerBootstrapData; import com.threerings.presents.peer.net.PeerCreds; import com.threerings.presents.peer.server.persist.NodeRecord; @@ -71,7 +75,7 @@ import static com.threerings.presents.Log.log; * servers and uses those objects to communicate cross-node information. */ public class PeerManager - implements ClientManager.ClientObserver + implements PeerProvider, ClientManager.ClientObserver { /** * Used by entities that wish to know when cached data has become stale due to a change on @@ -153,6 +157,11 @@ public class PeerManager } }); + // set the invocation service + _nodeobj.setPeerService( + (PeerMarshaller)PresentsServer.invmgr.registerDispatcher( + new PeerDispatcher(this), false)); + // register ourselves as a client observer PresentsServer.clmgr.addClientObserver(this); @@ -163,7 +172,7 @@ public class PeerManager refreshPeers(); } }.schedule(5000L, 60*1000L); - + // give derived classes an easy way to get in on the init action didInit(); } @@ -174,6 +183,11 @@ public class PeerManager */ public void shutdown () { + // clear out our invocation service + if (_nodeobj != null) { + PresentsServer.invmgr.clearDispatcher(_nodeobj.peerService); + } + // clear out our client observer registration PresentsServer.clmgr.removeClientObserver(this); @@ -193,22 +207,29 @@ public class PeerManager } /** - * 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. + * Returns the client object representing the connection to the named peer, or + * null if we are not currently connected to it. */ - public void acquireLock (final Lock.Name name, final ResultListener listener) + public Client getPeerClient (String nodeName) { - // wait for any resolution to end - queryLock(name, new ResultListener() { + PeerNode peer = _peers.get(nodeName); + return (peer == null) ? null : peer.getClient(); + } + + /** + * Acquires a lock on a resource shared amongst this node's peers. If the lock + * is successfully acquired, the supplied listener will receive this node's name. + * If another node acquires the lock first, then the listener will receive the + * name of that node. + */ + public void acquireLock (final Lock lock, final ResultListener listener) + { + // wait until any pending resolution is complete + queryLock(lock, new ResultListener() { public void requestCompleted (String result) { if (result == null) { - // the lock is free; we will attempt to acquire it - continueAcquiringLock(name, listener); + _locks.put(lock, new LockHandler(lock, true, listener)); } else { - // another node has acquired the lock listener.requestCompleted(result); } } @@ -219,83 +240,104 @@ public class PeerManager } /** - * 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}). + * Releases a lock. This can be cancelled using {@link #reacquireLock}, in which case the + * passed listener will receive this node's name as opposed to null, which + * signifies that the lock has been successfully released. */ - public void releaseLock (final Lock.Name name, final ResultListener listener) + public void releaseLock (final Lock lock, 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 + "]."); + // wait until any pending resolution is complete + queryLock(lock, new ResultListener() { + public void requestCompleted (String result) { + if (_nodeName.equals(result)) { + _locks.put(lock, new LockHandler(lock, false, listener)); + } else { + if (result != null) { + log.warning("Tried to release lock held by another peer [lock=" + + lock + ", owner=" + result + "]."); + } + listener.requestCompleted(result); + } + } + public void requestFailed (Exception cause) { + listener.requestFailed(cause); + } + }); + } + + /** + * Reacquires a lock after a call to {@link #releaseLock} but before the result listener + * supplied to that method has been notified with the result of the action. The result + * listener will receive the name of this node to indicate that the lock is still held. + * If a node requests to release a lock, then receives a lock-related request from another + * peer, it can use this method to cancel the release reliably, since the lock-related + * request will have been sent before the peer's ratification of the release. + */ + public void reacquireLock (Lock lock) + { + // make sure we're releasing it + LockHandler handler = _locks.get(lock); + if (handler == null || !handler.getNodeName().equals(_nodeName) || handler.isAcquiring()) { + log.warning("Tried to reacquire lock not being released [lock=" + lock + ", handler=" + + handler + "]."); + return; } - // announce our desire to drop the lock - _nodeobj.setReleasingLock(name); + // perform an update to let other nodes know that we're reacquiring + _nodeobj.updateLocks(lock); - // 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(); + // cancel the handler and report to any listeners + _locks.remove(lock); + handler.cancel(); + handler.listeners.requestCompleted(_nodeName); } /** - * Reacquires a lock for which we have called {@link #releaseLock}, but have not yet received - * a response. + * Determines the owner of the specified lock, waiting for any resolution to complete before + * notifying the supplied listener. */ - public void reacquireLock (final Lock.Name name, final ResultListener listener) + public void queryLock (Lock lock, ResultListener listener) { - // broadcast our desire to reacquire - _nodeobj.setAcquiringLock(name); + // if it's being resolved, add the listener to the list + LockHandler handler = _locks.get(lock); + if (handler != null) { + handler.listeners.add(listener); + return; + } - // 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(); + // otherwise, return its present value + listener.requestCompleted(queryLock(lock)); } /** - * 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 + * Finds the owner of the specified lock (if any) among this node and its peers. This answer + * is not definitive, as the lock may be in the process of resolving. */ - public void queryLock (Lock.Name name, final ResultListener listener) + public String queryLock (Lock lock) { - Tuple result = queryLock(name, false); - if (result.left) { - listener.requestCompleted(result.right); + // look for it in our own lock set + if (_nodeobj.locks.contains(lock)) { + return _nodeName; + } + + // then in our peers + for (PeerNode peer : _peers.values()) { + if (peer.nodeobj.locks.contains(lock)) { + return peer._record.nodeName; + } + } + return null; + } + + // documentation inherited from interface PeerProvider + public void ratifyLockAction (ClientObject caller, Lock lock, boolean acquire) + { + LockHandler handler = _locks.get(lock); + if (handler != null && handler.getNodeName().equals(_nodeName)) { + handler.ratify(acquire); } else { - new LockResolutionListener(name, false) { - public void lockResolved (String owner) { - listener.requestCompleted(owner); - } - }.add(); + // this is not an error condition, as we may have cancelled the handler or + // allowed another to take priority } } @@ -486,53 +528,6 @@ 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. */ @@ -659,26 +654,38 @@ public class PeerManager { 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); + Lock lock = nodeobj.acquiringLock; + LockHandler handler = _locks.get(lock); + if (handler == null) { + handler = new LockHandler(this, lock, true); } else { - log.warning("Node attempting to release lock not held? [node=" + - _record.nodeName + ", name=" + nodeobj.releasingLock + ", lock=" + - lock + "]."); + int val = handler.getNodeName().compareTo(_record.nodeName); + if (val < 0) { + return; // existing handler has priority + } else if (val == 0) { + log.warning("Received duplicate acquire request [handler=" + + handler + "]."); + return; + } + // this node has priority, so cancel the existing handler and take over + // its listeners + ResultListenerList olisteners = handler.listeners; + handler.cancel(); + handler = new LockHandler(this, lock, true); + handler.listeners.addAll(olisteners); } + _locks.put(lock, handler); + + } else if (name.equals(NodeObject.RELEASING_LOCK)) { + Lock lock = nodeobj.releasingLock; + LockHandler handler = _locks.get(lock); + if (handler == null) { + _locks.put(lock, new LockHandler(this, lock, false)); + } else { + log.warning("Received request to release resolving lock [node=" + + _record.nodeName + ", handler=" + handler + "]."); + } + } else if (name.equals(NodeObject.CACHE_DATA)) { changedCacheData(nodeobj.cacheData.cache, nodeobj.cacheData.data); } @@ -690,78 +697,150 @@ public class PeerManager } /** - * Listens to all {@link NodeObject}s (possibly including our own) to determine when the lock - * state has been agreed upon. + * Handles a lock in a state of resolution. */ - protected abstract class LockResolutionListener extends SetAdapter + protected class LockHandler + implements SetListener, ObjectDeathListener { - public LockResolutionListener (Lock.Name name, boolean peersOnly) - { - _name = name; - _peersOnly = peersOnly; - } + /** Listeners waiting for resolution. */ + public ResultListenerList listeners = new ResultListenerList(); - public void add () + /** + * Creates a handler to acquire or release a lock for this node. + */ + public LockHandler (Lock lock, boolean acquire, ResultListener listener) { - 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; + _lock = lock; + _acquire = acquire; + listeners.add(listener); + + // signal our desire to acquire or release the lock + if (acquire) { + _nodeobj.setAcquiringLock(lock); + } else { + _nodeobj.setReleasingLock(lock); } - lockResolved(result.right); + // find out exactly how many responses we need + _remaining = _peers.size(); - for (PeerNode peer : _peers.values()) { - peer.nodeobj.removeListener(this); - } - if (!_peersOnly) { - _nodeobj.removeListener(this); - } + // schedule a timeout to act if something goes wrong + (_timeout = new Interval(PresentsServer.omgr) { + public void expired () { + log.warning("Lock handler timed out, acting anyway [lock=" + _lock + + ", acquire=" + _acquire + "]."); + activate(); + } + }).schedule(LOCK_TIMEOUT); } /** - * Called when the owner of the lock has been agreed upon. + * Creates a handle that tracks another node's acquisition or release of a lock. */ - protected abstract void lockResolved (String owner); + public LockHandler (PeerNode peer, Lock lock, boolean acquire) + { + _peer = peer; + _lock = lock; + _acquire = acquire; + + // ratify the action + peer.nodeobj.peerService.ratifyLockAction(peer.getClient(), lock, acquire); + + // listen for the act to take place + peer.nodeobj.addListener(this); + } - /** The name of the lock. */ - protected Lock.Name _name; + public String getNodeName () + { + return (_peer == null) ? _nodeName : _peer._record.nodeName; + } - /** If true, wait for all nodes except this one to agree. */ - protected boolean _peersOnly; + public boolean isAcquiring () + { + return _acquire; + } + + public void ratify (boolean acquire) + { + if (_acquire == acquire && --_remaining == 0) { + _timeout.cancel(); + activate(); + } + } + + public void cancel () + { + if (_peer != null) { + _peer.nodeobj.removeListener(this); + } else { + _timeout.cancel(); + } + } + + // documentation inherited from interface SetListener + public void entryAdded (EntryAddedEvent event) + { + if (_acquire && event.getName().equals(NodeObject.LOCKS) && + event.getEntry().equals(_lock)) { + wasActivated(_peer._record.nodeName); + } + } + + // documentation inherited from interface SetListener + public void entryRemoved (EntryRemovedEvent event) + { + if (!_acquire && event.getName().equals(NodeObject.LOCKS) && + event.getOldEntry().equals(_lock)) { + wasActivated(null); + } + } + + // documentation inherited from interface SetListener + public void entryUpdated (EntryUpdatedEvent event) + { + if (!_acquire && event.getName().equals(NodeObject.LOCKS) && + event.getEntry().equals(_lock)) { + wasActivated(_peer._record.nodeName); + } + } + + // documentation inherited from interface ObjectDeathListener + public void objectDestroyed (ObjectDestroyedEvent event) + { + _locks.remove(_lock); + listeners.requestCompleted(null); + } + + @Override // documentation inherited + public String toString () + { + return "[node=" + getNodeName() + ", lock=" + _lock + ", acquire=" + _acquire + "]"; + } + + protected void activate () + { + _locks.remove(_lock); + if (_acquire) { + _nodeobj.addToLocks(_lock); + listeners.requestCompleted(_nodeName); + } else { + _nodeobj.removeFromLocks(_lock); + listeners.requestCompleted(null); + } + } + + protected void wasActivated (String owner) + { + _peer.nodeobj.removeListener(this); + _locks.remove(_lock); + listeners.requestCompleted(owner); + } + + protected PeerNode _peer; + protected Lock _lock; + protected boolean _acquire; + protected int _remaining; + protected Interval _timeout; } protected String _nodeName, _hostName, _publicHostName, _sharedSecret; @@ -775,6 +854,9 @@ public class PeerManager protected HashMap> _cacheobs = new HashMap>(); - /** Contains the names of any locks we are currently trying to acquire. */ - protected HashSet _acquiringLocks = new HashSet(); + /** Locks in the process of resolution. */ + protected HashMap _locks = new HashMap(); + + /** We wait this long for peer ratification to complete before acquiring/releasing the lock. */ + protected static final long LOCK_TIMEOUT = 5000L; } diff --git a/src/java/com/threerings/presents/peer/server/PeerProvider.java b/src/java/com/threerings/presents/peer/server/PeerProvider.java new file mode 100644 index 000000000..8d8133029 --- /dev/null +++ b/src/java/com/threerings/presents/peer/server/PeerProvider.java @@ -0,0 +1,40 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2006 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.peer.server; + +import com.threerings.presents.client.Client; +import com.threerings.presents.data.ClientObject; +import com.threerings.presents.peer.client.PeerService; +import com.threerings.presents.peer.data.NodeObject; +import com.threerings.presents.server.InvocationException; +import com.threerings.presents.server.InvocationProvider; + +/** + * Defines the server-side of the {@link PeerService}. + */ +public interface PeerProvider extends InvocationProvider +{ + /** + * Handles a {@link PeerService#ratifyLockAction} request. + */ + public void ratifyLockAction (ClientObject caller, NodeObject.Lock arg1, boolean arg2); +}