From d3d841e3e54b17ef4859653240bf56e633e2a1f5 Mon Sep 17 00:00:00 2001 From: Andrzej Kapolka Date: Tue, 24 Apr 2007 19:47:12 +0000 Subject: [PATCH] Some fiddling to handle cases where peers become connected after having been disconnected (in which case there may be conflicts in lock ownership) or are connected asymmetrically (A is subscribed to B, B is not yet subscribed to A). git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4674 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../presents/peer/server/PeerManager.java | 116 ++++++++++++++++-- .../peer/server/persist/NodeRecord.java | 6 + .../peer/server/persist/NodeRepository.java | 9 ++ 3 files changed, 122 insertions(+), 9 deletions(-) diff --git a/src/java/com/threerings/presents/peer/server/PeerManager.java b/src/java/com/threerings/presents/peer/server/PeerManager.java index 2616c7459..f57d91d3d 100644 --- a/src/java/com/threerings/presents/peer/server/PeerManager.java +++ b/src/java/com/threerings/presents/peer/server/PeerManager.java @@ -90,6 +90,18 @@ public class PeerManager public void changedCacheData (Streamable data); } + /** + * Used by entities that wish to know when this peer has been forced into immediately releasing + * a lock. + */ + public static interface DroppedLockObserver + { + /** + * Called when this node has been forced to drop a lock. + */ + public void droppedLock (Lock lock); + } + /** * Creates a peer manager which will create a {@link NodeRepository} which will be used to * publish our existence and discover the other nodes. @@ -191,7 +203,20 @@ public class PeerManager // clear out our client observer registration PresentsServer.clmgr.removeClientObserver(this); - // TODO: clear our record from the node table + // clear our record from the node table + _invoker.postUnit(new Invoker.Unit() { + public boolean invoke () { + try { + _noderepo.deleteNode(_nodeName); + } catch (PersistenceException pe) { + log.warning("Failed to delete node record [nodeName=" + _nodeName + + ", error=" + pe + "]."); + } + return false; + } + }); + + // shut down the peers for (PeerNode peer : _peers.values()) { peer.shutdown(); } @@ -407,6 +432,22 @@ public class PeerManager return null; } + /** + * Adds an observer to notify when this peer has been forced to drop a lock immediately. + */ + public void addDroppedLockObserver (DroppedLockObserver observer) + { + _dropobs.add(observer); + } + + /** + * Removes a dropped lock observer from the list. + */ + public void removeDroppedLockObserver (DroppedLockObserver observer) + { + _dropobs.remove(observer); + } + // documentation inherited from interface PeerProvider public void ratifyLockAction (ClientObject caller, Lock lock, boolean acquire) { @@ -611,6 +652,19 @@ public class PeerManager }); } + /** + * Called when we have been forced to drop a lock. + */ + protected void droppedLock (final Lock lock) + { + _dropobs.apply(new ObserverList.ObserverOp() { + public boolean apply (DroppedLockObserver observer) { + observer.droppedLock(lock); + return true; + } + }); + } + /** * Initializes the supplied client info for the supplied client. */ @@ -627,6 +681,15 @@ public class PeerManager return new PeerNode(record); } + /** + * Determines whether the first node named has priority over the second when resolving + * lock disputes. + */ + protected static boolean hasPriority (String nodeName1, String nodeName2) + { + return nodeName1.compareTo(nodeName2) < 0; + } + /** * Contains all runtime information for one of our peer nodes. */ @@ -664,7 +727,7 @@ public class PeerManager // if our client hasn't updated its record since we last tried to logon, then just // chill - if (_lastConnectStamp > record.lastUpdated.getTime()) { + if ((_lastConnectStamp - record.lastUpdated.getTime()) > STALE_INTERVAL) { log.fine("Not reconnecting to stale client [record=" + _record + ", lastTry=" + new Date(_lastConnectStamp) + "]."); return; @@ -739,6 +802,7 @@ public class PeerManager handler.clientDidLogoff(); } } + nodeobj = null; } // documentation inherited from interface ClientObserver @@ -752,6 +816,22 @@ public class PeerManager { nodeobj = object; + // check for lock conflicts + for (Lock lock : nodeobj.locks) { + LockHandler handler = _locks.get(lock); + if (handler != null) { + log.warning("Client hijacked lock in process of resolution [handler=" + + handler + ", node=" + _record.nodeName + "]."); + handler.clientHijackedLock(_record.nodeName); + + } else if (_nodeobj.locks.contains(lock)) { + log.warning("Client hijacked lock owned by this node [lock=" + lock + + ", node=" + _record.nodeName + "]."); + _nodeobj.removeFromLocks(lock); + droppedLock(lock); + } + } + // listen for lock and cache updates nodeobj.addListener(this); } @@ -771,14 +851,14 @@ public class PeerManager Lock lock = nodeobj.acquiringLock; LockHandler handler = _locks.get(lock); if (handler == null) { + if (_nodeobj.locks.contains(lock)) { + log.warning("Peer trying to acquire lock owned by this node [lock=" + + lock + ", node=" + _record.nodeName + "]."); + return; + } handler = new LockHandler(this, lock, true); } else { - 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 + "]."); + if (hasPriority(handler.getNodeName(), _record.nodeName)) { return; } // this node has priority, so cancel the existing handler and take over @@ -786,7 +866,7 @@ public class PeerManager ResultListenerList olisteners = handler.listeners; handler.cancel(); handler = new LockHandler(this, lock, true); - handler.listeners.addAll(olisteners); + handler.listeners = olisteners; } _locks.put(lock, handler); @@ -917,6 +997,17 @@ public class PeerManager listeners.requestCompleted(null); } + /** + * Called when a client hijacks the lock by having it in its node object when it + * connects. + */ + public void clientHijackedLock (String nodeName) + { + cancel(); + _locks.remove(_lock); + listeners.requestCompleted(nodeName); + } + /** * Cancels this handler, as another one will be taking its place. */ @@ -1023,9 +1114,16 @@ public class PeerManager protected HashMap> _cacheobs = new HashMap>(); + /** Listeners for dropped locks. */ + protected ObserverList _dropobs = + new ObserverList(ObserverList.FAST_UNSAFE_NOTIFY); + /** Locks in the process of resolution. */ protected HashMap _locks = new HashMap(); + /** The amount of time after which a node record can be considered out of date and invalid. */ + protected static final long STALE_INTERVAL = 5L * 60L * 1000L; + /** 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/persist/NodeRecord.java b/src/java/com/threerings/presents/peer/server/persist/NodeRecord.java index f4af1d57c..a38da3e14 100644 --- a/src/java/com/threerings/presents/peer/server/persist/NodeRecord.java +++ b/src/java/com/threerings/presents/peer/server/persist/NodeRecord.java @@ -60,6 +60,12 @@ public class NodeRecord this.port = port; } + /** Used for queries. */ + public NodeRecord (String nodeName) + { + this.nodeName = nodeName; + } + /** * Generates a string representation of this instance. */ diff --git a/src/java/com/threerings/presents/peer/server/persist/NodeRepository.java b/src/java/com/threerings/presents/peer/server/persist/NodeRepository.java index f832f6839..a63fc9e5a 100644 --- a/src/java/com/threerings/presents/peer/server/persist/NodeRepository.java +++ b/src/java/com/threerings/presents/peer/server/persist/NodeRepository.java @@ -73,6 +73,15 @@ public class NodeRepository extends JORARepository store(_ntable, record); } + /** + * Deletes the identified node record. + */ + public void deleteNode (String nodeName) + throws PersistenceException + { + delete(_ntable, new NodeRecord(nodeName)); + } + @Override // documentation inherited protected void migrateSchema (Connection conn, DatabaseLiaison liaison) throws SQLException, PersistenceException